test(capture): specify coherent snapshot view ownership

Cargo.lock

@@ -9808,8 +9808,10 @@ dependencies = [
  "redb",
  "serde",
  "serde_json",
+ "sha2 0.11.0",
  "tempfile",
  "thiserror 2.0.21",
+ "tokio",
  "url",
 ]
 

crates/capture-authority/Cargo.toml

@@ -10,8 +10,10 @@ description = "Durable local storage for capture authority"
 [dependencies]
 openraft = { workspace = true }
 redb = { workspace = true }
+sha2 = { 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,
@@ -278,10 +315,6 @@ pub(crate) fn native_application_state(tx: &ReadTransaction) -> Result<(Option<L
 }
 
 impl Store {
-    pub fn migrate_legacy_created_to_native(&mut self) -> Result<(), StoreError> {
-        Err(StoreError::UnsupportedSchema)
-    }
-
     /// Explicitly opt an empty, bound store into native entry persistence.
     pub fn enable_native_empty(&mut self) -> Result<(), StoreError> {
         let tx = self.write_transaction()?;
@@ -372,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 {
@@ -380,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/native_snapshot.rs

@@ -0,0 +1,26 @@
+//! Owned consistent native application snapshot view.
+use crate::{StoreError, snapshot};
+use std::io::Write;
+
+#[derive(Debug, thiserror::Error)]
+pub enum NativeSnapshotError {
+    #[error(transparent)]
+    Store(#[from] StoreError),
+    #[error(transparent)]
+    Snapshot(#[from] snapshot::SnapshotError),
+    #[error("native snapshot Store lock poisoned")]
+    Poisoned,
+    #[error("native snapshot worker failed")]
+    Worker(#[source] tokio::task::JoinError),
+    #[error("native snapshot view unavailable")]
+    Unavailable,
+}
+
+pub struct NativeSnapshotView {
+    _private: (),
+}
+impl NativeSnapshotView {
+    pub fn stream<W: Write>(self, _writer: W, _limits: snapshot::Limits) -> Result<W, NativeSnapshotError> {
+        Err(NativeSnapshotError::Unavailable)
+    }
+}