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

src/error.rs

@@ -52,6 +52,16 @@ pub enum Error {
 
     /// UTF-8 error
     Utf8(std::str::Utf8Error),
+
+    /// Range tombstone block decode failure.
+    RangeTombstoneDecode {
+        /// Which field or validation failed (e.g. `start_len`, `start`, `seqno`, `interval`)
+        field: &'static str,
+
+        /// Byte offset within the block to the start of the field whose decoding failed
+        /// (captured before reading bytes for that field).
+        offset: u64,
+    },
 }
 
 impl std::fmt::Display for Error {

src/table/mod.rs

@@ -632,65 +632,127 @@ impl Table {
     /// # Errors
     ///
     /// Will return `Err` if the block data is malformed.
+    #[expect(
+        clippy::cast_possible_truncation,
+        reason = "block sizes are bounded well within usize on all supported platforms"
+    )]
     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();
+
+        // A dedicated RT block with empty payload is corruption — the writer
+        // only creates an RT block handle when at least one tombstone exists.
+        if data.is_empty() {
+            log::error!("Range tombstone block: missing start_len");
+            return Err(crate::Error::RangeTombstoneDecode {
+                field: "start_len",
+                offset: 0,
+            });
+        }
+
         let mut cursor = Cursor::new(data);
 
-        #[expect(
-            clippy::cast_possible_truncation,
-            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 entry_offset = cursor.position();
+            let start_len_offset = entry_offset;
+            let start_len =
+                cursor
+                    .read_u16::<LE>()
+                    .map_err(|_| crate::Error::RangeTombstoneDecode {
+                        field: "start_len",
+                        offset: 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: 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.
+            // Using .get() instead of direct indexing to satisfy clippy::indexing_slicing.
+            let pos = cursor.position() as usize;
+            let start_end =
+                pos.checked_add(start_len)
+                    .ok_or(crate::Error::RangeTombstoneDecode {
+                        field: "start",
+                        offset: pos as u64,
+                    })?;
+            let start_buf = data
+                .get(pos..start_end)
+                .ok_or(crate::Error::RangeTombstoneDecode {
+                    field: "start",
+                    offset: pos as u64,
+                })?
+                .to_vec();
+            cursor.set_position(start_end as u64);
+
+            let end_len_offset = cursor.position();
+            let end_len =
+                cursor
+                    .read_u16::<LE>()
+                    .map_err(|_| crate::Error::RangeTombstoneDecode {
+                        field: "end_len",
+                        offset: 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: 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.
+            // Using .get() instead of direct indexing to satisfy clippy::indexing_slicing.
+            let pos = cursor.position() as usize;
+            let end_pos = pos
+                .checked_add(end_len)
+                .ok_or(crate::Error::RangeTombstoneDecode {
+                    field: "end",
+                    offset: pos as u64,
+                })?;
+            let end_buf = data
+                .get(pos..end_pos)
+                .ok_or(crate::Error::RangeTombstoneDecode {
+                    field: "end",
+                    offset: pos as u64,
+                })?
+                .to_vec();
+            cursor.set_position(end_pos as u64);
+
+            let seqno_offset = cursor.position();
+            let seqno =
+                cursor
+                    .read_u64::<LE>()
+                    .map_err(|_| crate::Error::RangeTombstoneDecode {
+                        field: "seqno",
+                        offset: 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: entry_offset,
+                });
             }
 
             tombstones.push(RangeTombstone::new(start, end, seqno));

src/table/tests.rs

@@ -1430,6 +1430,135 @@ 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 and expected byte offset.
+fn assert_rt_decode_error(data: Vec<u8>, expected_field: &str, expected_offset: u64) {
+    let block = rt_block(data);
+    match Table::decode_range_tombstones(&block) {
+        Err(crate::Error::RangeTombstoneDecode { field, offset }) => {
+            assert_eq!(
+                field, expected_field,
+                "expected field '{expected_field}', got '{field}'"
+            );
+            assert_eq!(
+                offset, expected_offset,
+                "expected offset {expected_offset}, got {offset}"
+            );
+        }
+        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", 0);
+}
+
+#[test]
+fn decode_range_tombstones_truncated_start_len_returns_error() {
+    // Only 1 byte — not enough for u16 start_len; offset = 0 (entry start)
+    assert_rt_decode_error(vec![0x01], "start_len", 0);
+}
+
+#[test]
+fn decode_range_tombstones_empty_block_returns_error() {
+    // Empty RT block payload is corruption — writer only creates an RT block
+    // handle when at least one tombstone exists.
+    assert_rt_decode_error(Vec::new(), "start_len", 0);
+}
+
+#[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; offset = 0 (entry start)
+    let mut buf = Vec::new();
+    buf.write_u16::<LE>(100).unwrap();
+    buf.push(0xFF);
+
+    assert_rt_decode_error(buf, "start_len", 0);
+}
+
+#[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
+    // offset = 3 (after u16 start_len + 1-byte key)
+    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", 3);
+}
+
+#[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
+    // offset = 3 (after u16 start_len + 1-byte key)
+    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", 3);
+}
+
+#[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)
+    // offset = 6 (after u16+1+u16+1 = 6 bytes for start/end fields)
+    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", 6);
+}
+
 /// 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"),
     }