test(capture): specify atomic populated native migration

Cargo.lock

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

crates/capture-authority/Cargo.toml

@@ -10,10 +10,8 @@ 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,40 +1,3 @@
-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,
@@ -315,6 +278,10 @@ 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()?;
@@ -405,14 +372,6 @@ 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 {
@@ -421,9 +380,18 @@ impl Store {
         if start > end {
             return Err(StoreError::InvalidLogSequence);
         }
-        read_native_range_in(&tx, start..end)
+        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)
     }
-
     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

@@ -1,26 +0,0 @@
-//! 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)
-    }
-}