cube(2): CUBE_OP_FLAG_SCAN — classify at write, retrieve by class

The store's flag field is a substrate, and this is its read half: a v4
index entry carries a 16-bit class mask, and a scan by class is a walk that
reads the mask and tests it. Nothing about the merge changes — the same log
overlay, the same order — so a caller pays for its own class rather than for
the store.

The op takes a space, a mask, a mode (any/all), a cursor and a buffer, in
its own size-versioned block. A mask of zero matches nothing, because naming
no class is asking no question. Records come back as
key | flags(2) | value_len | value: the mask travels, since a record can
carry bits the scan did not name and no other op returns a mask.

Both layouts the mask can be in are read. With the record still in the log
it comes from the v2 log entry; after a fold it comes from the v4 index
entry. The non-indexed path is not a corner — it is the state of every store
between the write that classified something and the fold, so answering it
with 'nothing' would make the substrate work only after a checkpoint.

Also settles what an append writes into which log, since the entry's frame
has to match the header a reader frames it by: a log that already holds v1
entries keeps taking v1 entries (a device folds first — that is the v4
migration — and a bare image keeps the log it has), an empty log is framed
v2 on a device and left alone on a bare image, and a log with no header is
framed v2 on a device and v1 on a bare image. The bare layout is the legacy
one: it has no index for a mask to be folded into, so a mask written there
could only be scanned and never checkpointed — half a feature, bought by
making every existing reader of that layout grow a version it cannot use.

drivers/cube/cube_syscall.c

@@ -56,6 +56,15 @@ int cubelinux_kernel_range(const __u8 *space,
 			   __u64 cursor, void *buf, size_t cap,
 			   __u64 *out_len, __u64 *out_cursor);
 
+/*
+ * The flag scan (CUBE_OP_FLAG_SCAN). The mask and its mode travel as plain numbers: which bits mean
+ * what is a vocabulary's business, and the kernel never interprets one — it compares masks, which is
+ * what lets a new vocabulary attach without a format change.
+ */
+int cubelinux_kernel_flag_scan(const __u8 *space, __u16 mask, __u16 mode,
+			       __u64 cursor, void *buf, size_t cap,
+			       __u64 *out_len, __u64 *out_cursor);
+
 /* The store device path, resolved from the `cube_store=` boot parameter at boot. */
 const char *cubelinux_store_device(void);
 
@@ -361,7 +370,60 @@ static long cube_range_op(void __user *uargs)
 }
 
 /*
- * One syscall, three argument blocks. They share a prefix — `size`, then `op` — so the size the
+ * The flag scan — CUBE_OP_FLAG_SCAN — which travels in `struct cube_flag_scan_args`.
+ *
+ * The walks' shape again, because it is the walks' contract: a batch fills the caller's buffer and
+ * returns how much was used plus the cursor to pass next; a record that does not fit ends the
+ * batch; a record that cannot fit in any buffer the caller offered comes back as -ERANGE with `len`
+ * saying what it would need.
+ *
+ * The one thing a caller must know beyond the walk's rules: the cursor counts the records that
+ * **matched**, not the records examined, because the records the mask rejected are not answers.
+ */
+static long cube_flag_scan_op(void __user *uargs)
+{
+	struct cube_flag_scan_args f;
+	void *buf = NULL;
+	long ret = 0;
+	u64 out_len = 0, out_cursor = 0;
+
+	if (copy_from_user(&f, uargs, sizeof(f)))
+		return -EFAULT;
+	if (f.size != sizeof(struct cube_flag_scan_args) || f.op != CUBE_OP_FLAG_SCAN)
+		return -EINVAL;
+	if (f.mode != CUBE_FLAG_ANY && f.mode != CUBE_FLAG_ALL)
+		return -EINVAL;
+
+	if (f.len > CUBE_MAX_WALK)
+		return -E2BIG;
+	if (f.len > 0) {
+		buf = kvmalloc(f.len, GFP_KERNEL);
+		if (!buf)
+			return -ENOMEM;
+	}
+
+	ret = cubelinux_kernel_flag_scan(f.space, f.mask, f.mode, f.cursor,
+					 buf, f.len, &out_len, &out_cursor);
+	if (ret == 0) {
+		if (out_len > 0 && copy_to_user((void __user *)f.value, buf, out_len))
+			ret = -EFAULT;
+		f.len = out_len;
+		f.cursor = out_cursor;
+		if (copy_to_user(uargs, &f, sizeof(f)))
+			ret = -EFAULT;
+	} else if (ret == -ERANGE) {
+		/* Nothing was written; `len` now says how much one record needs. */
+		f.len = out_len;
+		if (copy_to_user(uargs, &f, sizeof(f)))
+			ret = -EFAULT;
+	}
+
+	kvfree(buf);
+	return ret;
+}
+
+/*
+ * One syscall, four argument blocks. They share a prefix — `size`, then `op` — so the size the
  * caller declares is what says which one arrived. That is the whole point of putting `size`
  * first: an interface that cannot grow has to be replaced, and this one grows by being given a
  * new block with a new size.
@@ -378,5 +440,7 @@ SYSCALL_DEFINE2(cube, unsigned int, op, void __user *, uargs)
 		return cube_enum_op(op, uargs);
 	if (size == sizeof(struct cube_range_args))
 		return cube_range_op(uargs);
+	if (size == sizeof(struct cube_flag_scan_args))
+		return cube_flag_scan_op(uargs);
 	return -EINVAL;
 }

drivers/cube/cubelinux_store.rs

@@ -1167,24 +1167,34 @@ struct Mutation {
     value: KVVec<u8>,
 }
 
-/// Build a log entry: `op | crc32 | space | key | flags | len | value`.
+/// Build a log entry: `op | crc32 | space | key | [flags] | len | value`.
+///
+/// `version` decides whether the class mask is in the frame, and it must match the log's header:
+/// a reader frames every entry by the header's version, so an entry is only well-formed against the
+/// log it is written into. Version 1 has no mask field — the legacy log simply cannot carry one —
+/// and an entry going into such a log is written without it rather than not written at all.
 fn encode_entry(
     op: u8,
     space: &[u8; 32],
     key: &[u8; 24],
     flags: u16,
     value: &[u8],
+    version: u8,
 ) -> Result<KVVec<u8>, AllocError> {
-    let total = ENTRY_FIXED_V2 + value.len();
+    let flagged = version == WAL_VERSION_V2;
-    let mut entry = KVVec::<u8>::with_capacity(total, GFP_KERNEL)?;
+    let fixed = if flagged { ENTRY_FIXED_V2 } else { ENTRY_FIXED };
+    let mut entry = KVVec::<u8>::with_capacity(fixed + value.len(), GFP_KERNEL)?;
     entry.extend_from_slice(&[op], GFP_KERNEL)?;
     entry.extend_from_slice(&[0u8; 4][..], GFP_KERNEL)?;
     entry.extend_from_slice(space, GFP_KERNEL)?;
     entry.extend_from_slice(key, GFP_KERNEL)?;
-    entry.extend_from_slice(&flags.to_le_bytes(), GFP_KERNEL)?;
+    if flagged {
+        entry.extend_from_slice(&flags.to_le_bytes(), GFP_KERNEL)?;
+    }
     entry.extend_from_slice(&(value.len() as u32).to_le_bytes(), GFP_KERNEL)?;
     entry.extend_from_slice(value, GFP_KERNEL)?;
-    // The checksum covers everything after the crc field.
+    // The checksum covers everything after the crc field, so a v1 entry's checksum covers what a v1
+    // reader reads — the two versions are different framings of the same rule, not two rules.
     let crc = crc32(&entry.as_slice()[5..]);
     entry.as_mut_slice()[1..5].copy_from_slice(&crc.to_le_bytes());
     Ok(entry)
@@ -1247,8 +1257,6 @@ fn append(
     m: &Mutation,
     op: u8,
 ) -> Result<Option<Control>, Error> {
-    let entry = encode_entry(op, &m.space, &m.key, m.flags, m.value.as_slice())?;
-
     // Where it goes: after the bytes in use on a store device, after the valid prefix on a
     // bare image, where nothing records the length.
     let region_at = core::cmp::min(layout.log_off, image.len());
@@ -1271,21 +1279,62 @@ fn append(
             (used, region.len())
         }
     };
-    // The entries this function writes are v2, so a log whose header is v1 cannot take one:
+
-    // the reader frames every entry by the header's version, and a v1 header would mis-frame
+    // The entry's frame has to match the log's header, because the header is what a reader frames
-    // the v2 entries (and the v1 entries already there, if a header were simply rewritten).
+    // it by. So the header decides the version, and there are exactly three cases:
-    // A v1 log that still holds entries is folded into the image first — that is the v4
+    //
-    // migration — and the entry then lands in a fresh v2 log. A v1 log that is empty only
+    //   * a log that already holds v1 entries must keep taking v1 entries — rewriting the header
-    // needs its header upgraded.
+    //     would mis-frame every entry already there. A store *device* folds first instead (that is
-    if region.len() >= WAL_HEADER_LEN && &region[0..4] == WAL_MAGIC && region[4] == WAL_VERSION && used > 0 {
+    //     the v4 migration, and it has a spare slot to fold into); a bare image has nowhere to fold
-        if layout.control.is_some() {
+    //     to, so it keeps writing the log it has. The mask is lost for that append, because a v1
-            fold_now(image, layout)?;
+    //     entry has no field to carry it — a limitation of the legacy layout stated plainly, and
-            let (device, layout) = device_and_layout()?;
+    //     better than refusing to write to a log the store already holds data in;
-            return append(&device, &layout, m, op);
+    //   * no header, a v2 header, or a v1 header over an *empty* log: the frame is v2, and the
+    //     header is written (or upgraded) to say so.
+    let header_version = if region.len() >= WAL_HEADER_LEN && &region[0..4] == WAL_MAGIC {
+        Some(region[4])
+    } else {
+        None
+    };
+    let entry_version = match header_version {
+        // A log that already holds entries and declares v1 keeps taking v1 — rewriting the header
+        // would mis-frame everything already there. A store *device* first folds instead (that is
+        // the v4 migration, and it has a spare slot to fold into); a bare image has nowhere to fold
+        // to, so it keeps writing the log it has.
+        Some(WAL_VERSION) if used > 0 => {
+            if layout.control.is_some() {
+                fold_now(image, layout)?;
+                let (device, layout) = device_and_layout()?;
+                return append(&device, &layout, m, op);
+            }
+            WAL_VERSION
         }
-        pr_err!("cubelinux: a bare image cannot migrate its v1 log; refold it with the userspace tools\n");
+        // An empty log that already declares a version: a device's is upgraded to v2, since v2 is
-        return Err(EINVAL);
+        // the layout that carries the class mask and the one a fold turns into a v4 index. A bare
-    }
+        // image keeps what its log says.
+        Some(v) => {
+            if layout.control.is_some() {
+                WAL_VERSION_V2
+            } else {
+                v
+            }
+        }
+        // No header at all: this append writes the first one, and so chooses the framing for this
+        // log's life. A store device gets v2. A bare image gets v1, because the bare layout is the
+        // legacy one: it has no index, so a mask written into its log could only ever be read back
+        // by a scan and never be checkpointed — half a feature, bought by making every existing
+        // reader of that layout grow a version it has no use for. The format that carries a class
+        // mask is the checkpointable one, and that is the one that gets it.
+        None => {
+            if layout.control.is_some() {
+                WAL_VERSION_V2
+            } else {
+                WAL_VERSION
+            }
+        }
+    };
+
+    let entry = encode_entry(op, &m.space, &m.key, m.flags, m.value.as_slice(), entry_version)?;
 
     if WAL_HEADER_LEN + used + entry.len() > capacity {
         pr_err!(
@@ -1301,14 +1350,15 @@ fn append(
     let file = store_file()?;
 
     let mut result: Result<(), Error> = Ok(());
-    // A log region that has never been written is zeros, not a log: give it a header, the
+    // A log region that has never been written is zeros, not a log: give it a header, the same
-    // same thing the userspace log does when its file does not exist. A log region whose
+    // thing the userspace log does when its file does not exist. A header that does not already
-    // header is not v2 is upgraded, because the entry being written is v2 and the header is
+    // say the version this entry is framed by is (re)written to say it, because the header is what
-    // what a reader frames it by.
+    // a reader frames the entry by — and an entry under a header that disagrees with it is not a
-    if region.len() < WAL_HEADER_LEN || &region[0..4] != WAL_MAGIC || region[4] != WAL_VERSION_V2 {
+    // record, it is a mis-framing.
+    if region.len() < WAL_HEADER_LEN || &region[0..4] != WAL_MAGIC || region[4] != entry_version {
         let mut hdr = [0u8; WAL_HEADER_LEN];
         hdr[0..4].copy_from_slice(&WAL_MAGIC);
-        hdr[4] = WAL_VERSION_V2;
+        hdr[4] = entry_version;
         hdr[5] = 0; // morton
         result = write_and_sync(file, layout.log_off as u64, &hdr);
     }
@@ -2050,7 +2100,7 @@ enum Effect<'a> {
 fn log_effect<'a>(log: &'a [u8], space: &[u8; SPACE_ID_LEN], key: &[u8; RAW_KEY_LEN]) -> Option<Effect<'a>> {
     let mut entries = log_entries(log).ok()?;
     let mut found = None;
-    while let Some((entry_space, entry_key, op, value_at, value_len)) = entries.next() {
+    while let Some((entry_space, entry_key, op, _flags, value_at, value_len)) = entries.next() {
         if entry_space == space && entry_key == key {
             found = Some(if op == 2 {
                 Effect::Delete
@@ -2069,14 +2119,17 @@ struct LogEntries<'a> {
     /// The entry stride for this log's version, and where its length sits.
     fixed: usize,
     len_at: usize,
+    /// Whether this log's entries carry a class mask (version 2). A version 1 entry has no mask
+    /// field, and reads as "no class" — the same rule the index uses for a v3 entry.
+    flagged: bool,
 }
 
 impl<'a> LogEntries<'a> {
-    /// The next entry: `(space, key, op, where its value starts, its value's length)`.
+    /// The next entry: `(space, key, op, flags, where its value starts, its value's length)`.
     ///
     /// The value's *position* rather than a slice, because a caller collecting edits keeps them in
     /// a fixed-size vector and a borrow of the log would tie that vector's type to the log's life.
-    fn next(&mut self) -> Option<(&'a [u8], &'a [u8], u8, usize, usize)> {
+    fn next(&mut self) -> Option<(&'a [u8], &'a [u8], u8, u16, usize, usize)> {
         if self.off + self.fixed > self.log.len() {
             return None;
         }
@@ -2101,8 +2154,13 @@ impl<'a> LogEntries<'a> {
         if crc32(&self.log[start + 5..frame_end]) != crc {
             return None;
         }
+        let flags = if self.flagged {
+            cube_format::le_u16(self.log, start + 5 + SPACE_ID_LEN + RAW_KEY_LEN)
+        } else {
+            0
+        };
         self.off = frame_end;
-        Some((space, key, op, start + self.fixed, len))
+        Some((space, key, op, flags, start + self.fixed, len))
     }
 }
 
@@ -2120,6 +2178,7 @@ fn log_entries(log: &[u8]) -> Result<LogEntries<'_>, &'static str> {
         off: WAL_HEADER_LEN,
         fixed,
         len_at,
+        flagged: log[4] == WAL_VERSION_V2,
     })
 }
 
@@ -2133,6 +2192,9 @@ struct LogEdit<'a> {
     /// Position in the log, so the later of two edits to one key is the one that counts.
     seq: u32,
     value: &'a [u8],
+    /// The class mask the writer stamped, carried through the merge so a flag scan sees the log's
+    /// word on a record and not only the image's.
+    flags: u16,
     deleted: bool,
 }
 
@@ -2151,7 +2213,7 @@ fn log_edits<'a>(
         Err(_) => return Ok(()),
     };
     let mut seq: u32 = 0;
-    while let Some((edit_space, key, op, value_at, value_len)) = entries.next() {
+    while let Some((edit_space, key, op, flags, value_at, value_len)) = entries.next() {
         if edit_space == space {
             let key: &[u8; RAW_KEY_LEN] = match key.try_into() {
                 Ok(k) => k,
@@ -2162,6 +2224,7 @@ fn log_edits<'a>(
                     key,
                     seq,
                     value: &log[value_at..value_at + value_len],
+                    flags,
                     deleted: op == 2,
                 },
                 GFP_KERNEL,
@@ -2314,7 +2377,7 @@ fn log_spaces(log: &[u8], out: &mut KVVec<[u8; SPACE_ID_LEN]>) -> Result<()> {
         Ok(e) => e,
         Err(_) => return Ok(()),
     };
-    while let Some((space, _, _, _, _)) = entries.next() {
+    while let Some((space, _, _, _, _, _)) = entries.next() {
         let mut s = [0u8; SPACE_ID_LEN];
         s.copy_from_slice(&space[..SPACE_ID_LEN]);
         if !out.as_slice().contains(&s) {
@@ -2424,6 +2487,27 @@ impl Batch<'_> {
             }
         }
     }
+
+    /// Offer one *flagged* record. The counting rule is `offer`'s, and it is what makes the cursor
+    /// mean "matches already returned" rather than an index position: only a record that passed the
+    /// mask is counted, so a filtered walk resumes where it left off.
+    fn offer_flagged(&mut self, key: &[u8], flags: u16, value: &[u8]) -> bool {
+        self.seen += 1;
+        if self.seen <= self.cursor {
+            return true;
+        }
+        match pack_flagged(key, flags, value, self.out, self.written) {
+            Some(n) => {
+                self.written += n;
+                self.returned += 1;
+                true
+            }
+            None => {
+                self.too_big = RAW_KEY_LEN + FLAGS_LEN + 4 + value.len();
+                false
+            }
+        }
+    }
 }
 
 // ── The region walk: a box, its key span, and what decides membership ────────────────────
@@ -2602,22 +2686,28 @@ impl Addressed {
         Ok(None)
     }
 
-    /// One index entry, by its position in the index.
+    /// One index entry, by its position in the index: `(value_off, value_len, flags)`.
-    fn index_entry(&mut self, at: u64, buf: &mut KVVec<u8>) -> Result<(u64, u64)> {
+    ///
+    /// The class mask comes back with the address because it is in the same stride and costs
+    /// nothing extra to read — and because a flag scan is exactly a walk that reads it and nothing
+    /// else. A v3 entry has no mask and answers zero, the same way its log entries do.
+    fn index_entry(&mut self, at: u64, buf: &mut KVVec<u8>) -> Result<(u64, u64, u16)> {
         let stride = self.header.stride();
         let off = self.header.index_off + at * stride as u64;
         self.read_image_at(off, stride, buf)?;
         let entry = buf.as_slice();
-        let vo = if self.header.version == VERSION_V4 {
+        let flagged = self.header.version == VERSION_V4;
-            RAW_KEY_LEN + FLAGS_LEN
+        let flags = if flagged {
+            cube_format::le_u16(entry, RAW_KEY_LEN)
         } else {
-            RAW_KEY_LEN
+            0
         };
+        let vo = if flagged { RAW_KEY_LEN + FLAGS_LEN } else { RAW_KEY_LEN };
         let mut w = [0u8; 8];
         w.copy_from_slice(&entry[vo..vo + 8]);
         let value_off = u64::from_le_bytes(w);
         w.copy_from_slice(&entry[vo + 8..vo + 16]);
-        Ok((value_off, u64::from_le_bytes(w)))
+        Ok((value_off, u64::from_le_bytes(w), flags))
     }
 
     /// The address of `key` inside a space's index range, or nothing if the space does not hold it.
@@ -2636,7 +2726,9 @@ impl Addressed {
             } else if found > &key[..] {
                 hi = mid;
             } else {
-                return self.index_entry(first + mid, &mut buf).map(Some);
+                return self
+                    .index_entry(first + mid, &mut buf)
+                    .map(|(value_off, value_len, _)| Some((value_off, value_len)));
             }
         }
         Ok(None)
@@ -2843,7 +2935,7 @@ unsafe fn v3_enum(
         let mut entry = KVVec::<u8>::new();
         let mut at = first + cursor;
         while at < first + records {
-            let (value_off, value_len) = match image.index_entry(at, &mut entry) {
+            let (value_off, value_len, _) = match image.index_entry(at, &mut entry) {
                 Ok(pair) => pair,
                 Err(e) => return -(e.to_errno() as i32),
             };
@@ -2884,7 +2976,7 @@ unsafe fn v3_enum(
                 None => None,
             };
             match (edit, image_entry, image_key) {
-                (Some(e), Some((_, value_len)), Some(key)) if e.key == key => {
+                (Some(e), Some((_, value_len, _)), Some(key)) if e.key == key => {
                     let _ = value_len;
                     edit_at += 1;
                     at += 1;
@@ -2892,14 +2984,14 @@ unsafe fn v3_enum(
                         break;
                     }
                 }
-                (Some(e), Some((value_off, value_len)), Some(key)) if e.key < key => {
+                (Some(e), Some((value_off, value_len, _)), Some(key)) if e.key < key => {
                     edit_at += 1;
                     if !e.deleted && !batch.offer(e.key, e.value) {
                         break;
                     }
                     let _ = (value_off, value_len);
                 }
-                (_, Some((value_off, value_len)), Some(_)) => {
+                (_, Some((value_off, value_len, _)), Some(_)) => {
                     let value = match image.value_at(value_off, value_len) {
                         Ok(v) => v,
                         Err(e) => return -(e.to_errno() as i32),
@@ -3069,7 +3161,7 @@ unsafe fn v3_range(
                 }
             }
             // The image's own record, which the log says nothing about.
-            (_, Some((value_off, value_len)), Some(key)) => {
+            (_, Some((value_off, value_len, _)), Some(key)) => {
                 let value = match image.value_at(value_off, value_len) {
                     Ok(v) => v,
                     Err(e) => return -(e.to_errno() as i32),
@@ -3102,6 +3194,176 @@ unsafe fn v3_range(
     }
     0
 }
+
+// ── The flag scan: classify at write, retrieve by class ──────────────────────────────────
+//
+// The class mask lives in the index entry (v4) and in the log entry (WAL v2), so a scan by class is
+// a walk that reads the mask and tests it. Nothing else changes: the same merge with the log, the
+// same order. What it buys is that a record which does not match is never offered, so the caller
+// pays for its own class rather than for the store.
+
+/// A record matches when it shares *any* bit with the scan mask: "every error", "every Wi-Fi event".
+const FLAG_MODE_ANY: u16 = 0;
+/// A record matches when it carries *every* bit of the scan mask: "every Wi-Fi error".
+const FLAG_MODE_ALL: u16 = 1;
+
+/// Whether a record's class mask answers a scan for `mask` under `mode`.
+///
+/// A mask of zero matches nothing. Asking "what is this" with no class named is asking no question,
+/// and a scan that returned everything on an empty mask would be a walk wearing a scan's hat — so
+/// this is a refusal rather than a wildcard. `cube-store`'s `scan_by_flag` states the same rule,
+/// and the gate diffs the two.
+fn flag_matches(flags: u16, mask: u16, mode: u16) -> bool {
+    if mask == 0 {
+        return false;
+    }
+    if mode == FLAG_MODE_ALL {
+        flags & mask == mask
+    } else {
+        flags & mask != 0
+    }
+}
+
+/// `CUBE_OP_FLAG_SCAN` against a v3/v4 image: the live records of one space whose mask matches.
+///
+/// This is [`v3_enum`]'s merge with one difference, and it is the operation's whole point: a record
+/// is offered only when its class mask answers the scan, so the cursor counts **matches** rather
+/// than records walked. That is why this re-walks from the space's first record on every batch (as
+/// the region walk does) instead of starting at `first + cursor` the way the plain walk can: the
+/// index position of the cursor-th match is not arithmetic, because the records before it are not
+/// all matches.
+///
+/// A v3 image is a store whose records were written before the mask existed. Its entries read as
+/// "no class", so a scan over it matches nothing and says so by returning an empty batch — which is
+/// the honest answer, not an error.
+///
+/// # Safety
+/// `buf` must hold `cap` writable bytes; `out_len` and `out_cursor` must point to writable `u64`s.
+unsafe fn v3_flag_scan(
+    mut image: Addressed,
+    wanted: [u8; SPACE_ID_LEN],
+    mask: u16,
+    mode: u16,
+    cursor: u64,
+    buf: *mut u8,
+    cap: usize,
+    out_len: *mut u64,
+    out_cursor: *mut u64,
+) -> i32 {
+    let (first, records) = match image.space_entry(&wanted) {
+        Ok(Some(entry)) => entry,
+        Ok(None) => (0, 0),
+        Err(e) => return -(e.to_errno() as i32),
+    };
+    // Own the log: the edits borrow it, and walking the image needs `&mut image`.
+    let mut log = KVVec::<u8>::new();
+    if log.extend_from_slice(image.log(), GFP_KERNEL).is_err() {
+        return -12; // -ENOMEM
+    }
+    let mut edits = KVVec::<LogEdit<'_>>::new();
+    if log_edits(log.as_slice(), &wanted, &mut edits).is_err() {
+        return -12;
+    }
+
+    // SAFETY: the shim guarantees `cap` writable bytes at `buf`.
+    let out = unsafe { core::slice::from_raw_parts_mut(buf, cap) };
+    let mut batch = Batch {
+        out,
+        written: 0,
+        returned: 0,
+        // `seen` counts matches, and the cursor skips the matches already returned; starting it at
+        // `cursor` would skip the first `cursor` matches of a fresh walk *and* still count from
+        // there, re-serving the first batch forever. It starts at 0 and `offer_flagged` is the only
+        // thing that counts.
+        seen: 0,
+        cursor,
+        too_big: 0,
+    };
+
+    let mut entry = KVVec::<u8>::new();
+    let mut at = first;
+    let mut edit_at = 0usize;
+    loop {
+        // The entries of one key arrive in log order, so the last of a key's group is the newest
+        // word on it — and the newest word carries the mask that counts.
+        while edit_at + 1 < edits.len() && edits[edit_at + 1].key == edits[edit_at].key {
+            edit_at += 1;
+        }
+        let edit = edits.as_slice().get(edit_at).copied();
+        let image_entry = if at < first + records {
+            match image.index_entry(at, &mut entry) {
+                Ok(triple) => Some(triple),
+                Err(e) => return -(e.to_errno() as i32),
+            }
+        } else {
+            None
+        };
+        let image_key: Option<&[u8; RAW_KEY_LEN]> = match image_entry {
+            Some(_) => entry.as_slice()[..RAW_KEY_LEN].try_into().ok(),
+            None => None,
+        };
+
+        match (edit, image_entry, image_key) {
+            // The log's word on a coordinate the image holds: it wins, mask and all.
+            (Some(e), Some(_), Some(key)) if e.key == key => {
+                edit_at += 1;
+                at += 1;
+                if !e.deleted
+                    && flag_matches(e.flags, mask, mode)
+                    && !batch.offer_flagged(e.key, e.flags, e.value)
+                {
+                    break;
+                }
+            }
+            // A record only the log holds, in its key's place.
+            (Some(e), Some(_), Some(key)) if e.key < key => {
+                edit_at += 1;
+                if !e.deleted
+                    && flag_matches(e.flags, mask, mode)
+                    && !batch.offer_flagged(e.key, e.flags, e.value)
+                {
+                    break;
+                }
+            }
+            // The image's own record, which the log says nothing about.
+            (_, Some((value_off, value_len, flags)), Some(key)) => {
+                let value = match image.value_at(value_off, value_len) {
+                    Ok(v) => v,
+                    Err(e) => return -(e.to_errno() as i32),
+                };
+                at += 1;
+                if flag_matches(flags, mask, mode) && !batch.offer_flagged(key, flags, value.as_slice())
+                {
+                    break;
+                }
+            }
+            (Some(e), None, _) => {
+                edit_at += 1;
+                if !e.deleted
+                    && flag_matches(e.flags, mask, mode)
+                    && !batch.offer_flagged(e.key, e.flags, e.value)
+                {
+                    break;
+                }
+            }
+            (None, None, _) => break,
+            _ => break,
+        }
+    }
+
+    // SAFETY: both out-pointers are writable under this function's contract.
+    unsafe {
+        if batch.too_big > 0 && batch.written == 0 {
+            *out_len = batch.too_big as u64;
+            *out_cursor = cursor;
+            return -34; // -ERANGE
+        }
+        *out_len = batch.written as u64;
+        *out_cursor = cursor + batch.returned;
+    }
+    0
+}
+
 ///
 /// Returns the record's length, or a negative errno. When the value does not fit in `len` the
 /// bytes are *not* copied and the length is returned anyway, so the caller can size its buffer
@@ -3175,6 +3437,26 @@ fn pack_record(key: &[u8], value: &[u8], out: &mut [u8], at: usize) -> Option<us
     Some(need)
 }
 
+/// Pack one record for a flag scan: `key(24) | flags(2, LE) | value_len(u32, LE) | value`.
+///
+/// The mask travels with the record, unlike the walk's frame, because a scan's answer has to say
+/// *what class* each record answered with. A record can carry bits the scan did not ask for, and
+/// no other op returns a mask — so dropping it here would make those bits unreachable, which is
+/// the opposite of what a classification substrate is for. The field order echoes the v4 index
+/// entry (`key | flags | …`), so the wire shape is the index's shape.
+fn pack_flagged(key: &[u8], flags: u16, value: &[u8], out: &mut [u8], at: usize) -> Option<usize> {
+    let need = RAW_KEY_LEN + FLAGS_LEN + 4 + value.len();
+    if at + need > out.len() {
+        return None;
+    }
+    out[at..at + RAW_KEY_LEN].copy_from_slice(&key[..RAW_KEY_LEN]);
+    out[at + RAW_KEY_LEN..at + RAW_KEY_LEN + FLAGS_LEN].copy_from_slice(&flags.to_le_bytes());
+    out[at + RAW_KEY_LEN + FLAGS_LEN..at + RAW_KEY_LEN + FLAGS_LEN + 4]
+        .copy_from_slice(&(value.len() as u32).to_le_bytes());
+    out[at + RAW_KEY_LEN + FLAGS_LEN + 4..at + need].copy_from_slice(value);
+    Some(need)
+}
+
 /// `CUBE_OP_ENUM`: walk the records of one space into the caller's buffer, `cursor` records in.
 ///
 /// The cursor is a COUNT OF RECORDS ALREADY RETURNED, not a position in the image. It is opaque to
@@ -3368,6 +3650,117 @@ pub unsafe extern "C" fn cubelinux_kernel_range(
     0
 }
 
+/// `CUBE_OP_FLAG_SCAN`: the live records of one space whose class mask matches, in key order.
+///
+/// Its cursor counts **matches already returned**, exactly as the region walk's counts records in
+/// the box — for the same reason, and with the same consequence: a batch holds as many whole
+/// records as fit, most batches come back short, and reading a short batch as the end truncates the
+/// answer. A finished scan answers with no records and the cursor unchanged.
+///
+/// The answer's frame is `key(24) | flags(2, LE) | value_len(u32, LE) | value` — the walk's frame
+/// with the class mask in it, because a scan's answer has to say what class each record answered
+/// with and no other op returns a mask.
+///
+/// # Safety
+/// `space` must point to 32 readable bytes; `buf` must hold `cap` writable bytes; `out_len` and
+/// `out_cursor` must each point to a writable `u64`.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn cubelinux_kernel_flag_scan(
+    space: *const u8,
+    mask: u16,
+    mode: u16,
+    cursor: u64,
+    buf: *mut u8,
+    cap: usize,
+    out_len: *mut u64,
+    out_cursor: *mut u64,
+) -> i32 {
+    let mut wanted = [0u8; SPACE_ID_LEN];
+    // SAFETY: the caller guarantees 32 readable bytes at `space`.
+    unsafe { core::ptr::copy_nonoverlapping(space, wanted.as_mut_ptr(), SPACE_ID_LEN) };
+
+    // A mode this build does not know is refused rather than defaulted: a caller that asked for
+    // "all" and silently got "any" would get a superset, and a superset of a security question is
+    // the worst way to be wrong.
+    if mode != FLAG_MODE_ANY && mode != FLAG_MODE_ALL {
+        return -22; // -EINVAL
+    }
+
+    match Addressed::open() {
+        Ok(Some(image)) => {
+            return unsafe {
+                v3_flag_scan(image, wanted, mask, mode, cursor, buf, cap, out_len, out_cursor)
+            }
+        }
+        Ok(None) => {}
+        Err(e) => return -(e.to_errno() as i32),
+    }
+
+    let view = match read_view() {
+        Ok(v) => v,
+        Err(e) => return -(e.to_errno() as i32),
+    };
+
+    // A v1/v2 image holds packed records, which have no mask field at all — they read as "no class"
+    // and so can never match a scan. What *can* match is the log: its entries are the flagged ones,
+    // and they carry the mask the writer stamped. So the answer here is the log's live word on each
+    // coordinate it holds for the space, in key order — a walk's merge with the image's half of it
+    // known in advance to be empty of matches.
+    //
+    // This is not a corner: it is the state of every store between the write that classified
+    // something and the fold that moves the mask into the index — which is the ordinary state of a
+    // kernel that has just recorded an event. Answering it with "nothing" would make the substrate
+    // work only after a checkpoint, which is the opposite of "classify at write time".
+    let mut log = KVVec::<u8>::new();
+    if log.extend_from_slice(view.log(), GFP_KERNEL).is_err() {
+        return -12; // -ENOMEM
+    }
+    let mut edits = KVVec::<LogEdit<'_>>::new();
+    if log_edits(log.as_slice(), &wanted, &mut edits).is_err() {
+        return -12;
+    }
+
+    // SAFETY: the shim guarantees `cap` writable bytes at `buf`.
+    let out = unsafe { core::slice::from_raw_parts_mut(buf, cap) };
+    let mut batch = Batch {
+        out,
+        written: 0,
+        returned: 0,
+        seen: 0,
+        cursor,
+        too_big: 0,
+    };
+
+    let mut edit_at = 0usize;
+    while edit_at < edits.len() {
+        // The entries of one key arrive in log order, so the last of a key's group is the newest
+        // word on it — and only the newest can match, because it is what a read would return.
+        while edit_at + 1 < edits.len() && edits[edit_at + 1].key == edits[edit_at].key {
+            edit_at += 1;
+        }
+        let e = edits.as_slice()[edit_at];
+        edit_at += 1;
+        if !e.deleted
+            && flag_matches(e.flags, mask, mode)
+            && !batch.offer_flagged(e.key, e.flags, e.value)
+        {
+            break;
+        }
+    }
+
+    // SAFETY: both out-pointers are writable under this function's contract.
+    unsafe {
+        if batch.too_big > 0 && batch.written == 0 {
+            *out_len = batch.too_big as u64;
+            *out_cursor = cursor;
+            return -34; // -ERANGE
+        }
+        *out_len = batch.written as u64;
+        *out_cursor = cursor + batch.returned;
+    }
+    0
+}
+
 /// `CUBE_OP_SPACES`: the `cursor`-th distinct space that holds a record, or -ENOENT at the end.
 ///
 /// An index rather than a count of records: a caller that wants a space's records walks it with

include/uapi/linux/cube.h

@@ -43,6 +43,7 @@ struct cube_args {
 #define CUBE_OP_ENUM	5	/* walk the records of a space, in batches */
 #define CUBE_OP_SPACES	6	/* walk the spaces that hold records */
 #define CUBE_OP_RANGE	7	/* walk the records of a space that lie in a box */
+#define CUBE_OP_FLAG_SCAN 8	/* walk the records of a space whose class mask matches */
 
 /*
  * The walk's argument block: its own block rather than a wider `cube_args`, because it needs a
@@ -111,4 +112,49 @@ struct cube_range_args {
 				 */
 };
 
+/*
+ * The flag scan's argument block: `CUBE_OP_FLAG_SCAN`, the class-mask half of the store's
+ * classification substrate (DESIGN-flag-vocabularies.md). Its own block for the walks' reason — it
+ * needs a cursor and a buffer — and versioned by `size` like the other three.
+ *
+ * `mask` is a raw 16-bit class mask and this interface does not interpret a bit of it: the
+ * vocabulary that owns the bits (events today, sealing / lineage / lifecycle later) is the only
+ * thing that knows what they mean. `mode` says how to read the mask:
+ *
+ *	CUBE_FLAG_ANY	the record shares at least one bit with `mask` — "every error"
+ *	CUBE_FLAG_ALL	the record carries every bit of `mask` — "every Wi-Fi error"
+ *
+ * A `mask` of zero matches *nothing*, not everything: naming no class is asking no question, and a
+ * scan that answered a walk's worth of records to an empty question would be a walk wearing a
+ * scan's hat.
+ *
+ * The cursor counts **matches already returned**, as the region walk's counts records in its box,
+ * and for the same reason: the records examined before a match are not matches, so the index
+ * position of the cursor-th match is not arithmetic. A batch holds as many whole records as fit, so
+ * most batches come back short; reading a short batch as the end truncates the answer. A finished
+ * scan answers with no records and the cursor unchanged.
+ *
+ * Records come back as `key(24) | flags(2, little-endian) | value_len(u32, little-endian) | value`
+ * — the walk's frame with the class mask in it. The mask travels because a scan's answer has to say
+ * what class each record answered with: a record can carry bits beyond the one asked for, and no
+ * other operation returns a mask.
+ */
+struct cube_flag_scan_args {
+	__u32 size;		/* sizeof(struct cube_flag_scan_args) as the caller built it */
+	__u32 op;		/* CUBE_OP_FLAG_SCAN */
+	__u8 space[32];		/* in: the space to scan */
+	__u16 mask;		/* in: the class mask to match; 0 matches nothing */
+	__u16 mode;		/* in: CUBE_FLAG_ANY or CUBE_FLAG_ALL */
+	__u64 cursor;		/* in: 0 to start, or what the last call returned;
+				 * out: what to pass next — see the end-of-scan rule above
+				 */
+	__u64 value;		/* user pointer: where to put the records */
+	__u64 len;		/* in: the buffer's capacity;
+				 * out: bytes written, or on -ERANGE what would be needed
+				 */
+};
+
+#define CUBE_FLAG_ANY	0	/* the record shares at least one bit with the mask */
+#define CUBE_FLAG_ALL	1	/* the record carries every bit of the mask */
+
 #endif /* _UAPI_LINUX_CUBE_H */