feat(capture-authority): add native OpenRaft log reader

Cargo.lock

@@ -9810,6 +9810,7 @@ dependencies = [
  "serde_json",
  "tempfile",
  "thiserror 2.0.21",
+ "tokio",
  "url",
 ]
 

crates/capture-authority/Cargo.toml

@@ -12,6 +12,7 @@ openraft = { workspace = true }
 redb = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 thiserror = { workspace = true }
+tokio = { workspace = true }
 url = { workspace = true }
 
 [dev-dependencies]

crates/capture-authority/src/application.rs

@@ -1,3 +1,40 @@
+fn read_native_range_in<R: std::ops::RangeBounds<u64>>(
+    tx: &redb::ReadTransaction,
+    range: R,
+) -> Result<Vec<crate::CaptureRaftEntry>, StoreError> {
+    use std::ops::Bound::{Excluded, Included, Unbounded};
+    let logs = tx.open_table(LOGS)?;
+    let start = match range.start_bound() {
+        Included(value) => *value,
+        Excluded(value) => match value.checked_add(1) {
+            Some(value) => value,
+            None => return Ok(Vec::new()),
+        },
+        Unbounded => 0,
+    };
+    let end = match range.end_bound() {
+        Included(value) => *value,
+        Excluded(value) => match value.checked_sub(1) {
+            Some(value) => value,
+            None => return Ok(Vec::new()),
+        },
+        Unbounded => u64::MAX,
+    };
+    if start > end {
+        return Ok(Vec::new());
+    }
+    let mut entries = Vec::new();
+    for row in logs.range(start..=end)? {
+        let (key, value) = row?;
+        let entry = decode_log(value.value())?;
+        if entry.id.index != key.value() {
+            return Err(StoreError::CorruptRecord);
+        }
+        entries.push(decode_native_entry(entry)?);
+    }
+    Ok(entries)
+}
+
 use crate::{
     LOGS, LogId, META, Store, StoreError, committed::validate_committed, created::*, decode_log, encode_id, encode_log,
     next_index, number,
@@ -368,6 +405,14 @@ impl Store {
             })
             .transpose()
     }
+    /// Read typed native entries within the requested bounds.
+    pub fn read_native_range<R: std::ops::RangeBounds<u64>>(&self, range: R) -> Result<Vec<crate::CaptureRaftEntry>, StoreError> {
+        let tx = self.database.begin_read()?;
+        if number(required(tx.open_table(META)?.get("schema")?)?.value())? != 4 {
+            return Err(StoreError::UnsupportedSchema);
+        }
+        read_native_range_in(&tx, range)
+    }
     pub fn read_native_entries(&self, start: u64, end: u64) -> Result<Vec<crate::CaptureRaftEntry>, StoreError> {
         let tx = self.database.begin_read()?;
         if number(required(tx.open_table(META)?.get("schema")?)?.value())? != 4 {
@@ -376,18 +421,9 @@ impl Store {
         if start > end {
             return Err(StoreError::InvalidLogSequence);
         }
-        let logs = tx.open_table(LOGS)?;
-        let mut entries = Vec::new();
-        for row in logs.range(start..end)? {
-            let (key, value) = row?;
-            let entry = decode_log(value.value())?;
-            if entry.id.index != key.value() {
-                return Err(StoreError::CorruptRecord);
-            }
-            entries.push(decode_native_entry(entry)?);
-        }
-        Ok(entries)
+        read_native_range_in(&tx, start..end)
     }
+
     pub fn apply_next_native(&mut self, id: LogId) -> Result<crate::CaptureResponse, StoreError> {
         let tx = self.write_transaction()?;
         let response;

crates/capture-authority/src/lib.rs

@@ -23,6 +23,61 @@ pub use raft_types::{
 mod application_tests;
 mod truncate;
 
+/// OpenRaft log reader backed by exclusive local authority storage.
+#[derive(Clone)]
+pub struct NativeLogReader {
+    store: std::sync::Arc<std::sync::Mutex<Store>>,
+}
+
+impl NativeLogReader {
+    pub fn new(store: Store) -> Self {
+        Self {
+            store: std::sync::Arc::new(std::sync::Mutex::new(store)),
+        }
+    }
+}
+
+#[derive(Debug, thiserror::Error)]
+enum NativeReaderError {
+    #[error(transparent)]
+    Store(#[from] StoreError),
+    #[error("native log reader lock poisoned")]
+    Poisoned,
+    #[error("native log reader worker failed")]
+    Worker,
+}
+
+fn native_read_error(error: NativeReaderError) -> openraft::StorageError<u64> {
+    openraft::StorageError::IO {
+        source: openraft::StorageIOError::read_logs(openraft::AnyError::new(&error)),
+    }
+}
+
+async fn native_read_worker<F>(
+    store: std::sync::Arc<std::sync::Mutex<Store>>,
+    read: F,
+) -> Result<Vec<CaptureRaftEntry>, openraft::StorageError<u64>>
+where
+    F: FnOnce(&Store) -> Result<Vec<CaptureRaftEntry>, StoreError> + Send + 'static,
+{
+    tokio::task::spawn_blocking(move || {
+        let store = store.lock().map_err(|_| native_read_error(NativeReaderError::Poisoned))?;
+        read(&store).map_err(|error| native_read_error(NativeReaderError::Store(error)))
+    })
+    .await
+    .map_err(|_| native_read_error(NativeReaderError::Worker))?
+}
+
+impl openraft::RaftLogReader<CaptureTypeConfig> for NativeLogReader {
+    async fn try_get_log_entries<RB: std::ops::RangeBounds<u64> + Clone + std::fmt::Debug + openraft::OptionalSend>(
+        &mut self,
+        range: RB,
+    ) -> Result<Vec<CaptureRaftEntry>, openraft::StorageError<u64>> {
+        let bounds = (range.start_bound().cloned(), range.end_bound().cloned());
+        native_read_worker(self.store.clone(), move |store| store.read_native_range(bounds)).await
+    }
+}
+
 const META: TableDefinition<&str, &[u8]> = TableDefinition::new("meta");
 const VOTE: TableDefinition<u8, &[u8]> = TableDefinition::new("vote");
 const LOGS: TableDefinition<u64, &[u8]> = TableDefinition::new("logs");
@@ -459,6 +514,153 @@ mod tests {
             node_id: 0,
         }
     }
+    fn u19_runtime() -> tokio::runtime::Runtime {
+        tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("reader runtime")
+    }
+
+    fn u19_assert_io(error: openraft::StorageError<u64>) {
+        match error {
+            openraft::StorageError::IO { source } => {
+                assert!(source.to_string().starts_with("when Read Logs:"), "U19_READER_ERROR: {source}");
+            }
+            other => panic!("U19_READER_ERROR: unexpected error {other:?}"),
+        }
+    }
+
+    #[test]
+    fn u19_reader_roundtrip() {
+        use openraft::RaftLogReader;
+        let (_dir, _backend, mut store) = u17a_store();
+        let entries = vec![u17a_blank(0), u17a_membership(1).0, u17a_created(2, u17a_command())];
+        store.append_native(&entries).expect("append reader fixture");
+        let mut reader = NativeLogReader::new(store);
+        let read = u19_runtime().block_on(reader.try_get_log_entries(..)).expect("trait read");
+        assert_eq!(read, entries, "U19_TRAIT_READ_SUCCESS");
+    }
+
+    #[test]
+    fn u19_reader_bounds_reopen() {
+        use openraft::RaftLogReader;
+        use std::ops::Bound::{Excluded, Included, Unbounded};
+        let (dir, backend, mut store) = u17a_store();
+        let entries = vec![u17a_blank(0), u17a_membership(1).0, u17a_created(2, u17a_command())];
+        store.append_native(&entries).expect("append range fixture");
+        drop(store);
+        drop(backend);
+        let store = Store::open(dir.path().join("db"), identity()).expect("reopen reader file");
+        let mut reader = NativeLogReader::new(store);
+        u19_runtime().block_on(async {
+            assert_eq!(
+                reader.try_get_log_entries(1..2).await.expect("bounded"),
+                entries[1..2],
+                "U19_READER_RANGE"
+            );
+            assert_eq!(reader.try_get_log_entries(..=1).await.expect("inclusive"), entries[..2]);
+            assert_eq!(
+                reader
+                    .try_get_log_entries((Excluded(0), Included(2)))
+                    .await
+                    .expect("excluded"),
+                entries[1..]
+            );
+            assert_eq!(reader.try_get_log_entries(2..).await.expect("unbounded end"), entries[2..]);
+            assert_eq!(
+                reader
+                    .try_get_log_entries((Unbounded, Included(u64::MAX)))
+                    .await
+                    .expect("max"),
+                entries
+            );
+            for range in [
+                (Included(2), Excluded(2)),
+                (Included(2), Included(1)),
+                (Excluded(u64::MAX), Unbounded),
+            ] {
+                assert!(reader.try_get_log_entries(range).await.expect("empty bounds").is_empty());
+            }
+            let start = 1;
+            let end = 2;
+            assert_eq!(
+                reader
+                    .try_get_log_entries((Included(&start), Excluded(&end)))
+                    .await
+                    .expect("borrowed bounds"),
+                entries[1..2]
+            );
+        });
+    }
+
+    #[test]
+    fn u19_reader_legacy_corrupt() {
+        use openraft::RaftLogReader;
+        let (_legacy_dir, _legacy_backend, legacy) = u17a_bound();
+        let mut legacy = NativeLogReader::new(legacy);
+        let runtime = u19_runtime();
+        u19_assert_io(
+            runtime
+                .block_on(legacy.try_get_log_entries(0..0))
+                .expect_err("U19_READER_ERROR: legacy empty range"),
+        );
+        let (_dir, _backend, mut store) = u17a_store();
+        store.append_native(&[u17a_blank(0)]).expect("append corruption fixture");
+        let tx = store.database.begin_write().expect("corruption transaction");
+        {
+            let mut logs = tx.open_table(LOGS).expect("logs");
+            logs.insert(0, b"corrupt".as_slice()).expect("corrupt row");
+        }
+        tx.commit().expect("persist corruption");
+        let mut reader = NativeLogReader::new(store);
+        u19_assert_io(
+            runtime
+                .block_on(reader.try_get_log_entries(..))
+                .expect_err("U19_READER_ERROR: corrupt row"),
+        );
+    }
+
+    #[test]
+    fn u19_reader_worker_thread() {
+        use openraft::RaftLogReader;
+        let (_dir, _backend, mut store) = u17a_store();
+        let entries = vec![u17a_blank(0)];
+        store.append_native(&entries).expect("append worker fixture");
+        let mut reader = NativeLogReader::new(store);
+        let runtime = u19_runtime();
+        let async_thread = std::thread::current().id();
+        let (sender, receiver) = mpsc::channel();
+        runtime.block_on(async {
+            let read = native_read_worker(reader.store.clone(), move |store| {
+                sender.send(std::thread::current().id()).expect("send worker thread");
+                store.read_native_range(..)
+            })
+            .await
+            .expect("worker read");
+            assert_eq!(read, entries);
+            assert_ne!(receiver.recv().expect("worker thread"), async_thread, "U19_BLOCKING_THREAD");
+            assert_eq!(reader.try_get_log_entries(..).await.expect("trait read"), entries);
+        });
+    }
+
+    #[test]
+    fn u19_reader_worker_errors() {
+        use openraft::RaftLogReader;
+        let (_dir, _backend, store) = u17a_store();
+        let mut reader = NativeLogReader::new(store);
+        let runtime = u19_runtime();
+        let error = runtime
+            .block_on(native_read_worker(reader.store.clone(), |_| panic!("test worker panic")))
+            .expect_err("worker panic must fail");
+        assert!(error.to_string().contains("native log reader worker failed"));
+        u19_assert_io(error);
+        let error = runtime
+            .block_on(reader.try_get_log_entries(..))
+            .expect_err("poison must fail");
+        assert!(error.to_string().contains("native log reader lock poisoned"));
+        u19_assert_io(error);
+    }
+
     fn u17a_id(index: u64) -> LogId {
         LogId {
             term: 11,
@@ -535,6 +737,89 @@ mod tests {
         )
     }
     #[test]
+    fn u18_native_read_roundtrip() {
+        let (_dir, _backend, mut store) = u17a_store();
+        let command = u17a_command();
+        let entries = vec![u17a_blank(0), u17a_membership(1).0, u17a_created(2, command.clone())];
+        store.append_native(&entries).expect("append native fixture");
+        let result = store.read_native_range(..);
+        assert!(result.is_ok(), "U18_NATIVE_READ_SUCCESS: {result:?}");
+        let read = result.expect("native read");
+        assert_eq!(read, entries, "full identities and payloads");
+        assert_eq!(
+            store.read_entries(2, 3).expect("canonical row")[0].payload[14..],
+            command.encode().expect("canonical command")
+        );
+    }
+    #[test]
+    fn u18_native_read_range_bounds() {
+        use std::ops::Bound::{Excluded, Included, Unbounded};
+        let (_dir, _backend, mut store) = u17a_store();
+        let entries = vec![u17a_blank(0), u17a_membership(1).0, u17a_created(2, u17a_command())];
+        store.append_native(&entries).expect("append range fixture");
+        assert_eq!(store.read_native_range(1..2).expect("bounded read"), entries[1..2], "U18_RANGE_BOUNDARY");
+        assert_eq!(store.read_native_range(..=1).expect("inclusive end"), entries[..2]);
+        assert_eq!(store.read_native_range((Excluded(0), Included(2))).expect("excluded start"), entries[1..]);
+        assert_eq!(store.read_native_range(2..).expect("unbounded end"), entries[2..]);
+        assert_eq!(store.read_native_range(..).expect("all"), entries);
+        for range in [
+            (Included(2), Excluded(2)),
+            (Included(2), Included(1)),
+            (Unbounded, Excluded(0)),
+            (Excluded(u64::MAX), Unbounded),
+            (Included(u64::MAX), Included(u64::MAX)),
+        ] {
+            assert!(store.read_native_range(range).expect("empty mathematical range").is_empty());
+        }
+    }
+    #[test]
+    fn u18_native_read_reopen() {
+        let (dir, backend, mut store) = u17a_store();
+        let entries = vec![u17a_blank(0), u17a_membership(1).0, u17a_created(2, u17a_command())];
+        store.append_native(&entries).expect("append reopen fixture");
+        drop(store);
+        drop(backend);
+        let store = Store::open(dir.path().join("db"), identity()).expect("reopen native file");
+        assert_eq!(store.read_native_range(..).expect("reopened native read"), entries);
+    }
+    #[test]
+    fn u18_native_read_legacy_rejected() {
+        let (_dir, _backend, store) = u17a_bound();
+        assert!(
+            matches!(store.read_native_range(0..0), Err(StoreError::UnsupportedSchema)),
+            "U18_SCHEMA_GUARD"
+        );
+        assert!(matches!(store.read_native_range(..), Err(StoreError::UnsupportedSchema)));
+    }
+    #[test]
+    fn u18_native_read_corrupt_row() {
+        let (_dir, _backend, mut store) = u17a_store();
+        store.append_native(&[u17a_blank(0)]).expect("append corruption fixture");
+        let original = store.read_entry(0).expect("read stored entry").expect("present entry");
+        let tx = store.database.begin_write().expect("corruption transaction");
+        {
+            let mut logs = tx.open_table(LOGS).expect("logs");
+            let bytes = logs.get(0).expect("get row").expect("present row").value().to_vec();
+            logs.insert(1, bytes.as_slice()).expect("mismatched index row");
+        }
+        tx.commit().expect("persist mismatched row");
+        assert!(matches!(store.read_native_range(1..=1), Err(StoreError::CorruptRecord)));
+        let tx = store.database.begin_write().expect("malformed transaction");
+        {
+            let mut logs = tx.open_table(LOGS).expect("logs");
+            let mut entry = original;
+            entry.payload = b"CENR\x01".to_vec();
+            let bytes = encode_log(&entry).expect("encode malformed native envelope");
+            logs.insert(0, bytes.as_slice()).expect("malformed native row");
+        }
+        tx.commit().expect("persist malformed native row");
+        assert!(matches!(store.read_native_range(..1), Err(StoreError::CorruptRecord)));
+        let tx = store.database.begin_write().expect("missing-table transaction");
+        assert!(tx.delete_table(LOGS).expect("delete logs table"));
+        tx.commit().expect("persist missing logs table");
+        assert!(store.read_native_entries(0, 0).is_err(), "U18_EMPTY_RANGE_TABLE_ERROR");
+    }
+    #[test]
     fn u17a_native_enable_empty_idempotent() {
         let (_dir, backend, mut store) = u17a_bound();
         assert_eq!(committed_schema(&store), 3);