feat(range-tombstone): add structured error context to decode_range_tombstones

src/error.rs

@@ -52,6 +52,15 @@ pub enum Error {
 
     /// UTF-8 error
     Utf8(std::str::Utf8Error),
+
+    /// Range tombstone block decode failure
+    RangeTombstoneDecode {
+        /// Which wire-format field failed (e.g. `start_len`, `end_buf`, `seqno`)
+        field: &'static str,
+
+        /// Byte offset in the block when the failure occurred
+        offset: u64,
+    },
 }
 
 impl std::fmt::Display for Error {

src/table/mod.rs

@@ -634,7 +634,7 @@ impl Table {
     /// Will return `Err` if the block data is malformed.
     fn decode_range_tombstones(block: &Block) -> crate::Result<Vec<RangeTombstone>> {
         use byteorder::{ReadBytesExt, LE};
-        use std::io::{Cursor, Read};
+        use std::io::Cursor;
 
         let mut tombstones = Vec::new();
         let data = block.data.as_ref();
@@ -645,52 +645,88 @@ impl Table {
             reason = "block size always fits in usize"
         )]
         while (cursor.position() as usize) < data.len() {
-            let start_len = cursor
-                .read_u16::<LE>()
-                .map_err(|_| crate::Error::Unrecoverable)? as usize;
+            let offset = cursor.position();
+            let start_len =
+                cursor
+                    .read_u16::<LE>()
+                    .map_err(|_| crate::Error::RangeTombstoneDecode {
+                        field: "start_len",
+                        offset,
+                    })? as usize;
 
             // Validate length against remaining data before allocating
             let remaining = data.len() - cursor.position() as usize;
             if start_len > remaining {
                 log::error!(
                     "Range tombstone block: start_len {start_len} exceeds remaining {remaining}"
                 );
-                return Err(crate::Error::Unrecoverable);
+                return Err(crate::Error::RangeTombstoneDecode {
+                    field: "start_len",
+                    offset,
+                });
             }
 
-            let mut start_buf = vec![0u8; start_len];
-            cursor
-                .read_exact(&mut start_buf)
-                .map_err(|_| crate::Error::Unrecoverable)?;
-
-            let end_len = cursor
-                .read_u16::<LE>()
-                .map_err(|_| crate::Error::Unrecoverable)? as usize;
+            // Bounds already validated: start_len <= remaining
+            let pos = cursor.position() as usize;
+            let start_buf = data
+                .get(pos..pos + start_len)
+                .ok_or(crate::Error::RangeTombstoneDecode {
+                    field: "start_buf",
+                    offset: pos as u64,
+                })?
+                .to_vec();
+            cursor.set_position((pos + start_len) as u64);
+
+            let offset = cursor.position();
+            let end_len =
+                cursor
+                    .read_u16::<LE>()
+                    .map_err(|_| crate::Error::RangeTombstoneDecode {
+                        field: "end_len",
+                        offset,
+                    })? as usize;
 
             let remaining = data.len() - cursor.position() as usize;
             if end_len > remaining {
                 log::error!(
                     "Range tombstone block: end_len {end_len} exceeds remaining {remaining}"
                 );
-                return Err(crate::Error::Unrecoverable);
+                return Err(crate::Error::RangeTombstoneDecode {
+                    field: "end_len",
+                    offset,
+                });
             }
 
-            let mut end_buf = vec![0u8; end_len];
-            cursor
-                .read_exact(&mut end_buf)
-                .map_err(|_| crate::Error::Unrecoverable)?;
-
-            let seqno = cursor
-                .read_u64::<LE>()
-                .map_err(|_| crate::Error::Unrecoverable)?;
+            // Bounds already validated: end_len <= remaining
+            let pos = cursor.position() as usize;
+            let end_buf = data
+                .get(pos..pos + end_len)
+                .ok_or(crate::Error::RangeTombstoneDecode {
+                    field: "end_buf",
+                    offset: pos as u64,
+                })?
+                .to_vec();
+            cursor.set_position((pos + end_len) as u64);
+
+            let offset = cursor.position();
+            let seqno =
+                cursor
+                    .read_u64::<LE>()
+                    .map_err(|_| crate::Error::RangeTombstoneDecode {
+                        field: "seqno",
+                        offset,
+                    })?;
 
             let start = UserKey::from(start_buf);
             let end = UserKey::from(end_buf);
 
             // Validate invariant: start < end (reject corrupted data)
             if start >= end {
                 log::error!("Range tombstone block: invalid interval (start >= end)");
-                return Err(crate::Error::Unrecoverable);
+                return Err(crate::Error::RangeTombstoneDecode {
+                    field: "interval",
+                    offset: cursor.position(),
+                });
             }
 
             tombstones.push(RangeTombstone::new(start, end, seqno));

src/table/tests.rs

@@ -1430,6 +1430,130 @@ fn table_global_seqno() -> crate::Result<()> {
     Ok(())
 }
 
+/// Build a [`Block`] from raw bytes for `decode_range_tombstones` tests.
+#[expect(
+    clippy::expect_used,
+    reason = "test helper: data length is controlled and fits in u32"
+)]
+fn rt_block(data: Vec<u8>) -> Block {
+    let data_length = u32::try_from(data.len()).expect("test buffer fits in u32");
+    Block {
+        header: block::Header {
+            block_type: block::BlockType::RangeTombstone,
+            checksum: crate::Checksum::from_raw(0),
+            data_length,
+            uncompressed_length: data_length,
+        },
+        data: data.into(),
+    }
+}
+
+/// Assert `decode_range_tombstones` returns [`RangeTombstoneDecode`](crate::Error::RangeTombstoneDecode) with the given field.
+fn assert_rt_decode_error(data: Vec<u8>, expected_field: &str) {
+    let block = rt_block(data);
+    match Table::decode_range_tombstones(&block) {
+        Err(crate::Error::RangeTombstoneDecode { field, .. }) => {
+            assert_eq!(
+                field, expected_field,
+                "expected field '{expected_field}', got '{field}'"
+            );
+        }
+        other => panic!(
+            "expected RangeTombstoneDecode {{ field: \"{expected_field}\" }}, got: {other:?}"
+        ),
+    }
+}
+
+#[test]
+#[expect(clippy::unwrap_used)]
+fn decode_range_tombstones_invalid_interval_returns_error() {
+    use byteorder::{WriteBytesExt, LE};
+
+    // Build a single tombstone where start ("z") >= end ("a")
+    let mut buf = Vec::new();
+    buf.write_u16::<LE>(1).unwrap(); // start_len
+    buf.extend_from_slice(b"z");
+    buf.write_u16::<LE>(1).unwrap(); // end_len
+    buf.extend_from_slice(b"a");
+    buf.write_u64::<LE>(1).unwrap(); // seqno
+
+    assert_rt_decode_error(buf, "interval");
+}
+
+#[test]
+fn decode_range_tombstones_truncated_start_len_returns_error() {
+    // Only 1 byte — not enough for u16 start_len
+    assert_rt_decode_error(vec![0x01], "start_len");
+}
+
+#[test]
+fn decode_range_tombstones_empty_block_returns_ok() {
+    // Empty block has no tombstones — should succeed with empty vec
+    let block = rt_block(Vec::new());
+    match Table::decode_range_tombstones(&block) {
+        Ok(tombstones) => assert!(tombstones.is_empty()),
+        Err(err) => panic!("empty block should decode, got error: {err}"),
+    }
+}
+
+#[test]
+#[expect(clippy::unwrap_used)]
+fn decode_range_tombstones_start_len_exceeds_remaining_returns_error() {
+    use byteorder::{WriteBytesExt, LE};
+
+    // start_len = 100 but only 1 byte of data follows
+    let mut buf = Vec::new();
+    buf.write_u16::<LE>(100).unwrap();
+    buf.push(0xFF);
+
+    assert_rt_decode_error(buf, "start_len");
+}
+
+#[test]
+#[expect(clippy::unwrap_used)]
+fn decode_range_tombstones_truncated_end_len_returns_error() {
+    use byteorder::{WriteBytesExt, LE};
+
+    // Valid start_len + start, then truncated before end_len completes
+    let mut buf = Vec::new();
+    buf.write_u16::<LE>(1).unwrap(); // start_len = 1
+    buf.push(b'a'); // start key
+    buf.push(0x01); // only 1 byte of end_len (need 2)
+
+    assert_rt_decode_error(buf, "end_len");
+}
+
+#[test]
+#[expect(clippy::unwrap_used)]
+fn decode_range_tombstones_end_len_exceeds_remaining_returns_error() {
+    use byteorder::{WriteBytesExt, LE};
+
+    // Valid start, then end_len = 100 but only 1 byte follows
+    let mut buf = Vec::new();
+    buf.write_u16::<LE>(1).unwrap(); // start_len
+    buf.push(b'a'); // start key
+    buf.write_u16::<LE>(100).unwrap(); // end_len = 100
+    buf.push(0xFF); // only 1 byte
+
+    assert_rt_decode_error(buf, "end_len");
+}
+
+#[test]
+#[expect(clippy::unwrap_used)]
+fn decode_range_tombstones_truncated_seqno_returns_error() {
+    use byteorder::{WriteBytesExt, LE};
+
+    // Valid start + end, but seqno truncated (only 4 of 8 bytes)
+    let mut buf = Vec::new();
+    buf.write_u16::<LE>(1).unwrap(); // start_len
+    buf.push(b'a'); // start key
+    buf.write_u16::<LE>(1).unwrap(); // end_len
+    buf.push(b'z'); // end key
+    buf.extend_from_slice(&[0x01, 0x00, 0x00, 0x00]); // 4 bytes of seqno (need 8)
+
+    assert_rt_decode_error(buf, "seqno");
+}
+
 /// Exercises the `load_block` cache-miss and cache-hit paths for
 /// `BlockType::RangeTombstone`, verifying that the dedicated RT metrics
 /// counters are incremented instead of the data-block counters.

tests/range_tombstone.rs

@@ -476,8 +476,12 @@ fn range_tombstone_tampered_rt_block_fails_recovery() -> lsm_tree::Result<()> {
     )
     .open()
     {
-        Err(lsm_tree::Error::ChecksumMismatch { .. }) | Err(lsm_tree::Error::Unrecoverable) => {}
-        Err(other) => panic!("expected ChecksumMismatch or Unrecoverable, got: {other:?}"),
+        Err(lsm_tree::Error::ChecksumMismatch { .. })
+        | Err(lsm_tree::Error::Unrecoverable)
+        | Err(lsm_tree::Error::RangeTombstoneDecode { .. }) => {}
+        Err(other) => panic!(
+            "expected ChecksumMismatch, Unrecoverable, or RangeTombstoneDecode, got: {other:?}"
+        ),
         Ok(_) => panic!("tampered RT block must fail recovery, not reopen successfully"),
     }