feat(txn)!: return values as Bytes sharing the cached page

BREAKING: `ReadTxn::get` and `WriteTxn::get` return `Option<Bytes>` rather
than `Option<Vec<u8>>`.

Every lookup copied its value out of the page it had just pinned, so a warm
read — no I/O, no decryption, the bytes already in memory — still allocated and
memcpy'd. Hold the page buffer as `Bytes` and hand back a slice of it: a
refcount bump instead of a copy, and the warm get path now allocates nothing at
all.

The returned handle keeps that buffer alive by itself, so it outlives the guard
and survives the page leaving the cache, and the bytes under it never change
because a write installs a new page rather than mutating the old one. That is
what makes the slice a stable snapshot of the version that was read rather than
a window onto whatever the page becomes.

It does hold the whole page for as long as it lives, which is the real cost:
code that keeps one small value from each of many pages should copy it out.
Both entry points say so.

Values that have no committed page to borrow from — an overflow chain, or a
write this transaction has not flushed — are still assembled into a buffer of
their own, and `Bytes` carries them without a second copy.

Chosen now because the return type is the part that cannot change later: the
internals can get faster after 0.1, the signature cannot.

Author: farhan-syah

src/btree/leaf.rs

@@ -271,11 +271,18 @@ impl Default for Leaf {
     }
 }
 
-/// Borrowed view of one leaf value (zero-copy).
-#[derive(Debug)]
-pub enum LeafValueRef<'a> {
-    Inline(&'a [u8]),
-    Overflow { total_len: u64, root_page_id: u64 },
+/// Where slot `idx`'s value lives, without reading it.
+///
+/// Inline values report their extent rather than their bytes so the read path
+/// can slice the pinned page instead of copying out of it.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum LeafValueLoc {
+    /// Body range holding the value.
+    Inline(std::ops::Range<usize>),
+    Overflow {
+        total_len: u64,
+        root_page_id: u64,
+    },
 }
 
 /// Zero-allocation accessor over an encoded leaf page body. Used on the read
@@ -365,24 +372,22 @@ impl<'a> LeafAccessor<'a> {
         None
     }
 
-    /// Decode the value at slot `idx`. Inline values borrow from the page body;
-    /// overflow values return only the chain root + total length.
+    /// Locate the value at slot `idx` without reading it: an inline value's
+    /// extent within the body, or an overflow chain's root and total length.
     #[must_use]
-    pub fn value_at(&self, idx: usize) -> LeafValueRef<'a> {
+    pub fn value_loc(&self, idx: usize) -> LeafValueLoc {
         let off = slot_offset(self.body, self.prefix_len, idx);
         let suffix_len = read_u16_le(self.body, off) as usize;
         let after_key = off + 2 + suffix_len;
         let value_len_raw = read_u16_le(self.body, after_key);
         if value_len_raw == OVERFLOW_SENTINEL {
-            let total_len = read_u64_le(self.body, after_key + 2);
-            let root_page_id = read_u64_le(self.body, after_key + 2 + 8);
-            LeafValueRef::Overflow {
-                total_len,
-                root_page_id,
+            LeafValueLoc::Overflow {
+                total_len: read_u64_le(self.body, after_key + 2),
+                root_page_id: read_u64_le(self.body, after_key + 2 + 8),
             }
         } else {
-            let vlen = value_len_raw as usize;
-            LeafValueRef::Inline(&self.body[after_key + 2..after_key + 2 + vlen])
+            let start = after_key + 2;
+            LeafValueLoc::Inline(start..start + value_len_raw as usize)
         }
     }
 }

src/btree/tests/tree_ops.rs

@@ -232,8 +232,8 @@ async fn random_100k_ops_match_ground_truth() {
             let got = tree.get(&key).await.unwrap();
             let expected = truth.get(&key).cloned();
             assert_eq!(
-                got,
-                expected,
+                got.as_deref(),
+                expected.as_deref(),
                 "op {op} key {:?}",
                 String::from_utf8_lossy(&key)
             );

src/btree/tree/read.rs

@@ -6,8 +6,10 @@ use crate::pager::{PageGuard, Pager};
 use crate::vfs::Vfs;
 use crate::{RealmId, Result};
 
+use bytes::Bytes;
+
 use crate::btree::internal::InternalAccessor;
-use crate::btree::leaf::{Leaf, LeafAccessor, LeafValue, LeafValueRef};
+use crate::btree::leaf::{Leaf, LeafAccessor, LeafValue, LeafValueLoc};
 use crate::btree::node::NodeKind;
 use crate::btree::overflow;
 
@@ -18,10 +20,11 @@ impl<V: Vfs> BTree<V> {
     ///
     /// Zero-allocation tree descent: each level reads the page through a
     /// [`PageGuard`], builds a borrowed accessor over the decrypted body, and
-    /// either descends or extracts the value. The only allocation on the hit
-    /// path is the final owned `Vec<u8>` copy of the inline value (or the
-    /// overflow chain reassembly for large values).
-    pub async fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>> {
+    /// either descends or extracts the value. An inline value is returned as a
+    /// slice sharing the cached page's buffer, so the hit path allocates
+    /// nothing; only an overflow chain, which is reassembled from several
+    /// pages, needs a buffer of its own.
+    pub async fn get(&self, key: &[u8]) -> Result<Option<Bytes>> {
         if self.root_page_id == 0 {
             return Ok(None);
         }
@@ -42,16 +45,18 @@ impl<V: Vfs> BTree<V> {
         realm_id: RealmId,
         leaf: &Leaf,
         key: &[u8],
-    ) -> Result<Option<Vec<u8>>> {
+    ) -> Result<Option<Bytes>> {
         match leaf.get(key) {
             None => Ok(None),
-            Some(LeafValue::Inline(v)) => Ok(Some(v.clone())),
+            // Uncommitted, so it lives in the decoded leaf rather than in any
+            // page a slice could borrow from.
+            Some(LeafValue::Inline(v)) => Ok(Some(Bytes::copy_from_slice(v))),
             Some(LeafValue::Overflow {
                 total_len,
                 root_page_id,
             }) => {
                 let v = overflow::read_chain(pager, realm_id, *root_page_id, *total_len).await?;
-                Ok(Some(v))
+                Ok(Some(Bytes::from(v)))
             }
         }
     }
@@ -64,7 +69,7 @@ impl<V: Vfs> BTree<V> {
         key: &[u8],
         root_guard: &PageGuard,
         root_kind: NodeKind,
-    ) -> Result<Option<Vec<u8>>> {
+    ) -> Result<Option<Bytes>> {
         if self.root_page_id == 0 {
             return Ok(None);
         }
@@ -82,7 +87,7 @@ impl<V: Vfs> BTree<V> {
         start_page_id: u64,
         start_guard: &PageGuard,
         start_kind: NodeKind,
-    ) -> Result<Option<Vec<u8>>> {
+    ) -> Result<Option<Bytes>> {
         // Handle the first level using the borrowed guard.
         let next_page_id = match start_kind {
             NodeKind::Leaf => {
@@ -120,7 +125,7 @@ impl<V: Vfs> BTree<V> {
         key: &[u8],
         leaf_page_id: u64,
         leaf_guard: &PageGuard,
-    ) -> Result<Option<Vec<u8>>> {
+    ) -> Result<Option<Bytes>> {
         // Within a write txn, the dirty + fresh caches shadow the buffer-pool
         // bytes. The cached decoded form is the source of truth for any leaf
         // they hold.
@@ -131,31 +136,34 @@ impl<V: Vfs> BTree<V> {
         {
             return match leaf.get(key) {
                 None => Ok(None),
-                Some(LeafValue::Inline(v)) => Ok(Some(v.clone())),
+                // Uncommitted, so it lives in the decoded leaf rather than in
+                // any page a slice could borrow from.
+                Some(LeafValue::Inline(v)) => Ok(Some(Bytes::copy_from_slice(v))),
                 Some(LeafValue::Overflow {
                     total_len,
                     root_page_id,
                 }) => {
                     let v =
                         overflow::read_chain(&self.pager, self.realm_id, *root_page_id, *total_len)
                             .await?;
-                    Ok(Some(v))
+                    Ok(Some(Bytes::from(v)))
                 }
             };
         }
         let leaf = LeafAccessor::from_guard(leaf_guard)?;
         let Some(idx) = leaf.find(key) else {
             return Ok(None);
         };
-        match leaf.value_at(idx) {
-            LeafValueRef::Inline(v) => Ok(Some(v.to_vec())),
-            LeafValueRef::Overflow {
+        match leaf.value_loc(idx) {
+            // Shares the pinned page's buffer; no copy of the value at all.
+            LeafValueLoc::Inline(range) => Ok(Some(leaf_guard.body_slice(range))),
+            LeafValueLoc::Overflow {
                 total_len,
                 root_page_id,
             } => {
                 let v = overflow::read_chain(&self.pager, self.realm_id, root_page_id, total_len)
                     .await?;
-                Ok(Some(v))
+                Ok(Some(Bytes::from(v)))
             }
         }
     }

src/pager/cache.rs

@@ -10,6 +10,7 @@ use std::collections::BTreeSet;
 use std::sync::Arc;
 use std::sync::atomic::AtomicBool;
 
+use bytes::Bytes;
 use rustc_hash::FxHashMap;
 
 /// File-identity discriminator for the cache key.
@@ -30,8 +31,13 @@ pub enum FileKey {
 /// `bytes` is logically immutable for the page's cache lifetime — set once at
 /// insert, never mutated. Mutation happens by replacing the whole `Arc<Page>`
 /// in the cache map.
+///
+/// Held as [`Bytes`] rather than `Vec<u8>` so a reader can hand out a slice of
+/// a value without copying it: the slice shares this buffer's refcount and
+/// keeps it alive on its own. That the buffer never changes is what makes such
+/// a slice a stable snapshot even after the page leaves the cache.
 pub struct Page {
-    pub bytes: Vec<u8>,
+    pub bytes: Bytes,
     /// Page kind recorded at write time; used by the Pager flush path to
     /// reconstruct the correct AAD for each dirty page.
     pub kind_byte: u8,
@@ -58,17 +64,18 @@ impl Page {
     #[must_use]
     pub fn new(bytes: Vec<u8>) -> Self {
         Self {
-            bytes,
+            bytes: Bytes::from(bytes),
             kind_byte: 0,
             realm_id_bytes: None,
             extents_validated: AtomicBool::new(false),
         }
     }
 
+    /// `bytes` is moved into a [`Bytes`], not copied.
     #[must_use]
     pub fn new_with_meta(bytes: Vec<u8>, kind_byte: u8, realm_id_bytes: [u8; 16]) -> Self {
         Self {
-            bytes,
+            bytes: Bytes::from(bytes),
             kind_byte,
             realm_id_bytes: Some(realm_id_bytes),
             extents_validated: AtomicBool::new(false),

src/pager/core.rs

@@ -114,6 +114,23 @@ impl PageGuard {
         body(&self.page.bytes)
     }
 
+    /// Slice `range` of the body as an owned [`Bytes`] sharing this page's
+    /// buffer — a refcount bump, not a copy.
+    ///
+    /// The returned handle keeps the buffer alive by itself, so it stays valid
+    /// after the guard drops and after the page leaves the cache. The bytes it
+    /// sees never change: a write installs a new page rather than mutating this
+    /// one, which makes the slice a snapshot of the version that was read.
+    ///
+    /// It does pin the whole page for as long as it lives. A caller keeping a
+    /// small value from each of many pages should copy it out instead.
+    #[must_use]
+    pub fn body_slice(&self, range: std::ops::Range<usize>) -> Bytes {
+        let start = HEADER_LEN + range.start;
+        let end = HEADER_LEN + range.end;
+        self.page.bytes.slice(start..end)
+    }
+
     /// Per-page memo for the structural extent check run by decoders over
     /// [`body_ref`](Self::body_ref). Scoped to the pinned cache entry, so it is
     /// discarded whenever the page is replaced or evicted.

src/snapshot/tests/basic.rs

@@ -3736,8 +3736,11 @@ async fn chained_incremental_applies_keep_the_follower_able_to_advance() {
         let rtxn = follower.begin_read().await.unwrap();
         for i in 0u32..200 {
             assert_eq!(
-                rtxn.get(format!("k{i:05}").as_bytes()).await.unwrap(),
-                Some(vec![4u8; 128]),
+                rtxn.get(format!("k{i:05}").as_bytes())
+                    .await
+                    .unwrap()
+                    .as_deref(),
+                Some(vec![4u8; 128].as_slice()),
                 "key k{i:05} should reflect generation 4 after the first delta"
             );
         }
@@ -3767,8 +3770,11 @@ async fn chained_incremental_applies_keep_the_follower_able_to_advance() {
         let rtxn = follower.begin_read().await.unwrap();
         for i in 0u32..200 {
             assert_eq!(
-                rtxn.get(format!("k{i:05}").as_bytes()).await.unwrap(),
-                Some(vec![9u8; 128]),
+                rtxn.get(format!("k{i:05}").as_bytes())
+                    .await
+                    .unwrap()
+                    .as_deref(),
+                Some(vec![9u8; 128].as_slice()),
                 "key k{i:05} should reflect generation 9 after the second delta"
             );
         }
@@ -3871,15 +3877,21 @@ async fn a_delta_that_recycles_follower_private_page_ids_applies_cleanly() {
         let rtxn = follower.begin_read().await.unwrap();
         for i in 0u32..256 {
             assert_eq!(
-                rtxn.get(format!("k{i:05}").as_bytes()).await.unwrap(),
-                Some(vec![0xBBu8; 128]),
+                rtxn.get(format!("k{i:05}").as_bytes())
+                    .await
+                    .unwrap()
+                    .as_deref(),
+                Some(vec![0xBBu8; 128].as_slice()),
                 "refilled key k{i:05} must reflect the source's current state"
             );
         }
         for i in 256u32..512 {
             assert_eq!(
-                rtxn.get(format!("k{i:05}").as_bytes()).await.unwrap(),
-                Some(vec![0xAAu8; 128]),
+                rtxn.get(format!("k{i:05}").as_bytes())
+                    .await
+                    .unwrap()
+                    .as_deref(),
+                Some(vec![0xAAu8; 128].as_slice()),
                 "untouched key k{i:05} must reflect the source's current state"
             );
         }
@@ -4041,8 +4053,11 @@ async fn chained_deltas_survive_many_rounds_of_delete_and_refill_churn() {
         let rtxn = follower.begin_read().await.unwrap();
         for i in 0u32..512 {
             assert_eq!(
-                rtxn.get(format!("k{i:05}").as_bytes()).await.unwrap(),
-                Some(vec![round as u8; 128]),
+                rtxn.get(format!("k{i:05}").as_bytes())
+                    .await
+                    .unwrap()
+                    .as_deref(),
+                Some(vec![round as u8; 128].as_slice()),
                 "round {round}: key k{i:05} should reflect this round's generation"
             );
         }
@@ -4121,7 +4136,7 @@ async fn a_follower_stays_consistent_across_churn_by_falling_back_to_a_full_snap
         // Pre-round follower state, needed only if this round's delta gets
         // refused -- the refusal must be a pure no-op.
         let pre_round_commit = follower.latest_commit();
-        let pre_round_values: Vec<(String, Option<Vec<u8>>)> = {
+        let pre_round_values: Vec<(String, Option<bytes::Bytes>)> = {
             let rtxn = follower.begin_read().await.unwrap();
             let mut values = Vec::with_capacity(512);
             for i in 0u32..512 {
@@ -4193,8 +4208,11 @@ async fn a_follower_stays_consistent_across_churn_by_falling_back_to_a_full_snap
                 let rtxn = follower.begin_read().await.unwrap();
                 for i in 0u32..512 {
                     assert_eq!(
-                        rtxn.get(format!("k{i:05}").as_bytes()).await.unwrap(),
-                        Some(vec![round as u8; 128]),
+                        rtxn.get(format!("k{i:05}").as_bytes())
+                            .await
+                            .unwrap()
+                            .as_deref(),
+                        Some(vec![round as u8; 128].as_slice()),
                         "round {round}: key k{i:05} should reflect this round's generation after a successful apply"
                     );
                 }
@@ -4278,8 +4296,11 @@ async fn a_follower_stays_consistent_across_churn_by_falling_back_to_a_full_snap
                 let rtxn = new_follower.begin_read().await.unwrap();
                 for i in 0u32..512 {
                     assert_eq!(
-                        rtxn.get(format!("k{i:05}").as_bytes()).await.unwrap(),
-                        Some(vec![round as u8; 128]),
+                        rtxn.get(format!("k{i:05}").as_bytes())
+                            .await
+                            .unwrap()
+                            .as_deref(),
+                        Some(vec![round as u8; 128].as_slice()),
                         "round {round}: key k{i:05} should reflect this round's generation on the remedy follower"
                     );
                 }

src/txn/read/txn.rs

@@ -1,3 +1,4 @@
+use bytes::Bytes;
 use tokio::sync::OnceCell;
 
 use crate::btree::BTree;
@@ -92,7 +93,15 @@ impl<'db, V: Vfs + Clone> ReadTxn<'db, V> {
         Ok(())
     }
 
-    pub async fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>> {
+    /// Look up `key` in this snapshot.
+    ///
+    /// The value is returned as [`Bytes`] sharing the cached page's buffer
+    /// rather than copied out of it, so a hit allocates nothing. The handle
+    /// keeps that buffer alive on its own and the bytes never change under it,
+    /// so it stays valid and consistent after the transaction ends. It does
+    /// hold the whole page though: code retaining one small value from each of
+    /// many pages should copy it out with `to_vec`.
+    pub async fn get(&self, key: &[u8]) -> Result<Option<Bytes>> {
         self.check_abort()?;
         if self.root_page_id == 0 {
             return Ok(None);