fix(catalog): use scan_prefix instead of 0xFF-bounded range scans

Prefix scans over catalog rows built their end bound by appending a
single 0xFF byte to the start key. For a realm or segment id whose
first byte is itself 0xFF, that end bound sorts before (or equal to)
some in-range keys, silently dropping rows from segment listing,
stats, and rekey migration. One such migration failed with an
unverifiable-header corruption. Replace every appended-sentinel range
scan with scan_prefix, which bounds correctly regardless of key
content.

Author: farhan-syah

src/compaction/helpers.rs

@@ -60,9 +60,7 @@ pub(super) async fn list_all_segments_inner<V: Vfs + Clone>(
         pager.page_size(),
     );
     let start = vec![CatalogRowKind::Segment as u8];
-    let mut end = start.clone();
-    end.push(0xFF);
-    let rows = tree.collect_range(&start, &end).await?;
+    let rows = tree.scan_prefix(&start).await?;
     let mut out = Vec::with_capacity(rows.len());
     for (_k, v) in rows {
         let meta = Catalog::decode_segment_meta(&v)?;
@@ -88,9 +86,7 @@ pub(super) async fn find_segment_name_inner<V: Vfs + Clone>(
         pager.page_size(),
     );
     let start = vec![CatalogRowKind::Segment as u8];
-    let mut end = start.clone();
-    end.push(0xFF);
-    let rows = tree.collect_range(&start, &end).await?;
+    let rows = tree.scan_prefix(&start).await?;
     for (k, v) in rows {
         let meta = Catalog::decode_segment_meta(&v)?;
         if meta.segment_id == *segment_id && k.len() > 17 {

src/txn/db/misc.rs

@@ -161,12 +161,7 @@ impl<V: Vfs + Clone> Db<V> {
             );
 
             let seg_start = vec![CatalogRowKind::Segment as u8];
-            let mut seg_end = seg_start.clone();
-            seg_end.push(0xFF);
-            let seg_rows = tree
-                .collect_range(&seg_start, &seg_end)
-                .await
-                .unwrap_or_default();
+            let seg_rows = tree.scan_prefix(&seg_start).await.unwrap_or_default();
             let seg_count = u32::try_from(seg_rows.len()).unwrap_or(u32::MAX);
             let seg_bytes: u64 = seg_rows
                 .iter()

src/txn/db/rekey/segments.rs

@@ -348,9 +348,7 @@ impl<V: Vfs + Clone> Db<V> {
         }
         let tree = self.rekey_catalog_tree(state);
         let start = vec![CatalogRowKind::RekeySegmentProgress as u8];
-        let mut end = start.clone();
-        end.push(0xFF);
-        let rows = tree.collect_range(&start, &end).await?;
+        let rows = tree.scan_prefix(&start).await?;
         let mut out = BTreeMap::new();
         for (key, value) in rows {
             if key.len() != 17 {
@@ -371,9 +369,7 @@ impl<V: Vfs + Clone> Db<V> {
         }
         let tree = self.rekey_catalog_tree(state);
         let start = vec![CatalogRowKind::Segment as u8];
-        let mut end = start.clone();
-        end.push(0xFF);
-        tree.collect_range(&start, &end)
+        tree.scan_prefix(&start)
             .await?
             .into_iter()
             .map(|(key, value)| {

src/txn/db/segment.rs

@@ -72,9 +72,7 @@ impl<V: Vfs + Clone> Db<V> {
             self.page_size,
         );
         let start = Catalog::segment_key(realm, prefix.as_bytes())?;
-        let mut end = start.clone();
-        end.push(0xFF);
-        let rows = tree.collect_range(&start, &end).await?;
+        let rows = tree.scan_prefix(&start).await?;
         let mut out = Vec::with_capacity(rows.len());
         for (_k, v) in rows {
             let meta = Catalog::decode_segment_meta(&v)?;
@@ -296,9 +294,7 @@ impl<V: Vfs + Clone> Db<V> {
             self.page_size,
         );
         let start = vec![CatalogRowKind::Segment as u8];
-        let mut end = start.clone();
-        end.push(0xFF);
-        let rows = tree.collect_range(&start, &end).await?;
+        let rows = tree.scan_prefix(&start).await?;
         let mut out = Vec::with_capacity(rows.len());
         for (_k, v) in rows {
             let meta = Catalog::decode_segment_meta(&v)?;

src/txn/read/txn.rs

@@ -153,9 +153,7 @@ impl<'db, V: Vfs + Clone> ReadTxn<'db, V> {
         }
         let tree = self.catalog_tree();
         let start = Catalog::segment_key(self.db.realm_id, prefix.as_bytes())?;
-        let mut end = start.clone();
-        end.push(0xFF);
-        let rows = tree.collect_range(&start, &end).await?;
+        let rows = tree.scan_prefix(&start).await?;
         let mut out = Vec::with_capacity(rows.len());
         for (_k, v) in rows {
             out.push(Catalog::decode_segment_meta(&v)?);

src/txn/write/txn.rs

@@ -323,9 +323,7 @@ impl<'db, V: Vfs + Clone> WriteTxn<'db, V> {
         let mut prefix = Vec::with_capacity(17);
         prefix.push(0x01u8); // CatalogRowKind::Segment
         prefix.extend_from_slice(&realm.0);
-        let mut end = prefix.clone();
-        end.push(0xFF);
-        let rows = self.catalog_tree.collect_range(&prefix, &end).await?;
+        let rows = self.catalog_tree.scan_prefix(&prefix).await?;
         let mut committed: u64 = 0;
         for (_, v) in rows {
             let meta = Catalog::decode_segment_meta(&v)?;

tests/segment_basic.rs

@@ -308,3 +308,41 @@ async fn read_txn_pins_catalog_snapshot() {
     let page = reader.read_page(1).await.unwrap();
     assert!(page.starts_with(b"v1"));
 }
+
+/// Catalog segment rows are keyed `[0x01, realm(16), name]`. A "kind-prefix"
+/// scan (start `[0x01]`, as used by the live-segment stats) that bounded its end
+/// with an appended `0xFF` byte stopped at `[0x01, 0xFF]` (exclusive) and thus
+/// dropped every row of a realm whose first byte is `0xFF`. The scan must bound
+/// by prefix (`scan_prefix`), not by an appended sentinel. The same boundary
+/// defect made rekey migration of a segment whose random `segment_id` began with
+/// `0xFF` fail with an unverifiable-header corruption.
+#[tokio::test(flavor = "current_thread")]
+async fn segments_under_high_byte_realm_are_listed_and_counted() {
+    let realm = RealmId::new([0xFF; 16]);
+    let db = Db::open_internal(MemVfs::new(), [9u8; 32], PAGE, realm)
+        .await
+        .unwrap();
+
+    let mut w = db
+        .create_segment(realm, SegmentKind::Unspecified)
+        .await
+        .unwrap();
+    w.append_page(SegmentPageKind::Data, b"high-realm")
+        .await
+        .unwrap();
+    let meta = w.seal().await.unwrap();
+    {
+        let mut t = db.begin_write().await.unwrap();
+        t.link_segment("engine.idx", &meta).await.unwrap();
+        t.commit().await.unwrap();
+    }
+
+    let rx = db.begin_read().await.unwrap();
+    let listed = rx.list_segments("").await.unwrap();
+    assert_eq!(
+        listed.len(),
+        1,
+        "segment hidden from list_segments under 0xFF realm"
+    );
+    assert_eq!(db.stats().await.unwrap().segments_live, 1);
+}