feat(capture): stream snapshot rows with a permanent failure fence

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,
@@ -133,14 +170,145 @@ pub(crate) fn decode_native_entry(entry: crate::LogEntry) -> Result<crate::Captu
     })
 }
 
+fn apply_created(
+    tx: &redb::WriteTransaction,
+    id: LogId,
+    command: &DecideCreated,
+    canonical: &[u8],
+    binding: &CaptureBinding,
+) -> Result<CreatedResult, StoreError> {
+    let mut operations = tx.open_table(OPERATIONS)?;
+    let existing = operations
+        .get(command.operation_id.as_slice())?
+        .map(|v| decode_operation(v.value()))
+        .transpose()?;
+    let result = if let Some(existing) = existing {
+        if existing.canonical_command == canonical {
+            existing.result
+        } else {
+            CreatedResult::OperationConflict
+        }
+    } else {
+        let mut heads = tx.open_table(HEADS)?;
+        let mut versions = tx.open_table(VERSIONS)?;
+        let object = object_bytes(&command.object)?;
+        let key = version_key(&command.object, command.prepared.version_id)?;
+        let actual = heads
+            .get(object.as_slice())?
+            .map(|v| {
+                let mut d = Decoder::new(v.value());
+                let id = d.uuid()?;
+                d.finish()?;
+                Ok::<_, StoreError>(id)
+            })
+            .transpose()?;
+        let outcome = if command.binding_revision != binding.revision {
+            CreatedResult::BindingMismatch
+        } else if actual != command.expected_head {
+            CreatedResult::HeadMismatch { actual_head: actual }
+        } else if versions.get(key.as_slice())?.is_some() {
+            CreatedResult::VersionConflict
+        } else {
+            let mut version = Vec::new();
+            put_prepared(&mut version, &command.prepared);
+            version.extend(encode_id(id));
+            version.extend(binding.revision.to_be_bytes());
+            versions.insert(key.as_slice(), version.as_slice())?;
+            let mut event = object;
+            put_prepared(&mut event, &command.prepared);
+            event.extend(encode_id(id));
+            event.extend(binding_bytes(&binding)?);
+            tx.open_table(EVENTS)?
+                .insert(command.operation_id.as_slice(), event.as_slice())?;
+            heads.insert(object_bytes(&command.object)?.as_slice(), command.prepared.version_id.as_slice())?;
+            CreatedResult::Created {
+                version_id: command.prepared.version_id,
+                event_id: command.operation_id,
+                decision_log_id: id,
+            }
+        };
+        let mut op = u64::try_from(canonical.len())
+            .map_err(|_| StoreError::CorruptRecord)?
+            .to_be_bytes()
+            .to_vec();
+        op.extend(canonical);
+        op.extend(result_bytes(&outcome));
+        operations.insert(command.operation_id.as_slice(), op.as_slice())?;
+        outcome
+    };
+    Ok(result)
+}
+
+fn validate_response_kind(entry: &crate::CaptureRaftEntry, response: &crate::CaptureResponse) -> Result<(), StoreError> {
+    let kind = match entry.payload {
+        openraft::EntryPayload::Blank => crate::CaptureResponseKind::Blank,
+        openraft::EntryPayload::Normal(_) => crate::CaptureResponseKind::Created,
+        openraft::EntryPayload::Membership(_) => crate::CaptureResponseKind::Membership,
+    };
+    if response.kind() != kind || *response.log_id() != LogId::from(entry.log_id) {
+        return Err(StoreError::CorruptRecord);
+    }
+    Ok(())
+}
+fn encode_response(response: &crate::CaptureResponse) -> Result<Vec<u8>, StoreError> {
+    let (kind, body) = match response.kind() {
+        crate::CaptureResponseKind::Blank => (0, vec![]),
+        crate::CaptureResponseKind::Created => (1, result_bytes(response.created_result().ok_or(StoreError::CorruptRecord)?)),
+        crate::CaptureResponseKind::Membership => (2, vec![]),
+    };
+    let mut bytes = b"CRSP\x01".to_vec();
+    bytes.push(kind);
+    bytes.extend(encode_id(*response.log_id()));
+    bytes.extend(
+        u64::try_from(body.len())
+            .map_err(|_| StoreError::CorruptRecord)?
+            .to_be_bytes(),
+    );
+    bytes.extend(body);
+    Ok(bytes)
+}
+fn decode_response(bytes: &[u8]) -> Result<crate::CaptureResponse, StoreError> {
+    if bytes.len() < 38
+        || &bytes[..5] != b"CRSP\x01"
+        || number(&bytes[30..38])? != u64::try_from(bytes.len() - 38).map_err(|_| StoreError::CorruptRecord)?
+    {
+        return Err(StoreError::CorruptRecord);
+    }
+    let id = crate::decode_id(&bytes[6..30])?;
+    match bytes[5] {
+        0 if bytes.len() == 38 => crate::CaptureResponse::blank(id),
+        1 => {
+            let mut body = encode_id(id);
+            body.extend(&bytes[38..]);
+            crate::CaptureResponse::created(decode_apply(&body)?)
+        }
+        2 if bytes.len() == 38 => crate::CaptureResponse::membership(id),
+        _ => Err(StoreError::CorruptRecord),
+    }
+}
+fn entry_command(entry: &crate::LogEntry, native: bool) -> Result<DecideCreated, StoreError> {
+    if native {
+        match decode_native_entry(entry.clone())?.payload {
+            openraft::EntryPayload::Normal(command) => Ok(command.created().clone()),
+            _ => Err(StoreError::CorruptRecord),
+        }
+    } else {
+        DecideCreated::decode(&entry.payload)
+    }
+}
+
 pub(crate) fn native_application_state(tx: &ReadTransaction) -> Result<(Option<LogId>, crate::CaptureMembership), StoreError> {
     let meta = tx.open_table(META)?;
     if number(required(meta.get("schema")?)?.value())? != 4 {
         return Err(StoreError::UnsupportedSchema);
     }
     let last = read_applied(&meta)?;
-    let membership = crate::CaptureMembership::decode(required(meta.get("native_membership")?)?.value())?;
-    if last.is_some() || membership.log_id().is_some() || !membership.configs().is_empty() || !membership.nodes().is_empty() {
+    let membership = crate::CaptureMembership::decode(required(meta.get("native_membership")?)?.value())
+        .map_err(|_| StoreError::CorruptRecord)?;
+    if membership
+        .log_id()
+        .is_some_and(|id| last.is_none_or(|last| id.index > last.index))
+    {
         return Err(StoreError::CorruptRecord);
     }
     Ok((last, membership))
@@ -158,11 +326,13 @@ impl Store {
             }
             read_binding(&meta)?;
             if schema == 4 {
-                let membership = crate::CaptureMembership::decode(required(meta.get("native_membership")?)?.value())?;
-                if read_applied(&meta)?.is_some()
-                    || membership.log_id().is_some()
-                    || !membership.configs().is_empty()
-                    || !membership.nodes().is_empty()
+                let membership = crate::CaptureMembership::decode(required(meta.get("native_membership")?)?.value())
+                    .map_err(|_| StoreError::CorruptRecord)?;
+                let last = read_applied(&meta)?;
+                if membership
+                    .log_id()
+                    .is_some_and(|id| last.is_none_or(|last| id.index > last.index))
+                    || (last.is_none() && (!membership.configs().is_empty() || !membership.nodes().is_empty()))
                 {
                     return Err(StoreError::CorruptRecord);
                 }
@@ -235,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 {
@@ -243,24 +421,67 @@ 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);
+        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;
+        {
+            let mut meta = tx.open_table(META)?;
+            if number(required(meta.get("schema")?)?.value())? != 4 {
+                return Err(StoreError::UnsupportedSchema);
             }
-            entries.push(decode_native_entry(entry)?);
+            let binding = read_binding(&meta)?;
+            let logs = tx.open_table(LOGS)?;
+            let committed = validate_committed(&meta, &logs)?.ok_or(StoreError::InvalidCommittedLog)?;
+            let entry = decode_log(required(logs.get(id.index)?)?.value())?;
+            if entry.id != id || id.index > committed.index {
+                return Err(StoreError::InvalidCommittedLog);
+            }
+            let native = decode_native_entry(entry)?;
+            let last = read_applied(&meta)?;
+            let mut results = tx.open_table(APPLY_RESULTS)?;
+            if last.is_some_and(|last| id.index <= last.index) {
+                let saved = decode_response(required(results.get(id.index)?)?.value())?;
+                if *saved.log_id() != id {
+                    return Err(StoreError::InvalidApplySequence);
+                }
+                validate_response_kind(&native, &saved)?;
+                return Ok(saved);
+            }
+            if id.index != next_index(last.map(|last| last.index))? {
+                return Err(StoreError::InvalidApplySequence);
+            }
+            response = match native.payload {
+                openraft::EntryPayload::Blank => crate::CaptureResponse::blank(id)?,
+                openraft::EntryPayload::Normal(command) => {
+                    let canonical = command.created().encode()?;
+                    let result = apply_created(&tx, id, command.created(), &canonical, &binding)?;
+                    crate::CaptureResponse::created(CreatedApplyResult { log_id: id, result })?
+                }
+                openraft::EntryPayload::Membership(membership) => {
+                    let stored = openraft::StoredMembership::new(Some(id.into()), membership);
+                    let checked = crate::CaptureMembership::try_from(&stored)?;
+                    meta.insert("native_membership", checked.encode()?.as_slice())?;
+                    crate::CaptureResponse::membership(id)?
+                }
+            };
+            results.insert(id.index, encode_response(&response)?.as_slice())?;
+            let mut last = vec![1];
+            last.extend(encode_id(id));
+            meta.insert("last_applied", last.as_slice())?;
         }
-        Ok(entries)
+        self.commit(tx)?;
+        Ok(response)
     }
-    pub fn apply_next_native(&mut self, _id: LogId) -> Result<crate::CaptureResponse, StoreError> {
-        let _tx = self.write_transaction()?;
-        Err(StoreError::UnsupportedSchema)
-    }
-    pub fn native_apply_response(&self, _index: u64) -> Result<Option<crate::CaptureResponse>, StoreError> {
-        Err(StoreError::UnsupportedSchema)
+    pub fn native_apply_response(&self, index: u64) -> Result<Option<crate::CaptureResponse>, StoreError> {
+        let tx = self.database.begin_read()?;
+        if number(required(tx.open_table(META)?.get("schema")?)?.value())? != 4 {
+            return Err(StoreError::UnsupportedSchema);
+        }
+        let results = tx.open_table(APPLY_RESULTS)?;
+        results.get(index)?.map(|v| decode_response(v.value())).transpose()
     }
     pub fn native_application_state(&self) -> Result<(Option<LogId>, crate::CaptureMembership), StoreError> {
         native_application_state(&self.database.begin_read()?)
@@ -326,65 +547,7 @@ impl Store {
                 return Err(StoreError::InvalidApplySequence);
             }
             let command = DecideCreated::decode(&entry.payload)?;
-            let mut operations = tx.open_table(OPERATIONS)?;
-            let existing = operations
-                .get(command.operation_id.as_slice())?
-                .map(|v| decode_operation(v.value()))
-                .transpose()?;
-            result = if let Some(existing) = existing {
-                if existing.canonical_command == entry.payload {
-                    existing.result
-                } else {
-                    CreatedResult::OperationConflict
-                }
-            } else {
-                let mut heads = tx.open_table(HEADS)?;
-                let mut versions = tx.open_table(VERSIONS)?;
-                let object = object_bytes(&command.object)?;
-                let key = version_key(&command.object, command.prepared.version_id)?;
-                let actual = heads
-                    .get(object.as_slice())?
-                    .map(|v| {
-                        let mut d = Decoder::new(v.value());
-                        let id = d.uuid()?;
-                        d.finish()?;
-                        Ok::<_, StoreError>(id)
-                    })
-                    .transpose()?;
-                let outcome = if command.binding_revision != binding.revision {
-                    CreatedResult::BindingMismatch
-                } else if actual != command.expected_head {
-                    CreatedResult::HeadMismatch { actual_head: actual }
-                } else if versions.get(key.as_slice())?.is_some() {
-                    CreatedResult::VersionConflict
-                } else {
-                    let mut version = Vec::new();
-                    put_prepared(&mut version, &command.prepared);
-                    version.extend(encode_id(id));
-                    version.extend(binding.revision.to_be_bytes());
-                    versions.insert(key.as_slice(), version.as_slice())?;
-                    let mut event = object;
-                    put_prepared(&mut event, &command.prepared);
-                    event.extend(encode_id(id));
-                    event.extend(binding_bytes(&binding)?);
-                    tx.open_table(EVENTS)?
-                        .insert(command.operation_id.as_slice(), event.as_slice())?;
-                    heads.insert(object_bytes(&command.object)?.as_slice(), command.prepared.version_id.as_slice())?;
-                    CreatedResult::Created {
-                        version_id: command.prepared.version_id,
-                        event_id: command.operation_id,
-                        decision_log_id: id,
-                    }
-                };
-                let mut op = u64::try_from(entry.payload.len())
-                    .map_err(|_| StoreError::CorruptRecord)?
-                    .to_be_bytes()
-                    .to_vec();
-                op.extend(&entry.payload);
-                op.extend(result_bytes(&outcome));
-                operations.insert(command.operation_id.as_slice(), op.as_slice())?;
-                outcome
-            };
+            result = apply_created(&tx, id, &command, &entry.payload, &binding)?;
             let mut applied = encode_id(id);
             applied.extend(result_bytes(&result));
             results.insert(id.index, applied.as_slice())?;
@@ -443,27 +606,29 @@ impl Store {
     pub fn created_apply_result(&self, index: u64) -> Result<Option<CreatedApplyResult>, StoreError> {
         let tx = self.database.begin_read()?;
         let t = tx.open_table(APPLY_RESULTS)?;
-        t.get(index)?.map(|v| decode_apply(v.value())).transpose()
+        if number(required(tx.open_table(META)?.get("schema")?)?.value())? == 4 {
+            t.get(index)?
+                .map(|v| {
+                    decode_response(v.value()).map(|r| {
+                        r.created_result().cloned().map(|result| CreatedApplyResult {
+                            log_id: *r.log_id(),
+                            result,
+                        })
+                    })
+                })
+                .transpose()
+                .map(Option::flatten)
+        } else {
+            t.get(index)?.map(|v| decode_apply(v.value())).transpose()
+        }
     }
 }
 
 pub(crate) fn validate_application(tx: &ReadTransaction) -> Result<(), StoreError> {
     let meta = tx.open_table(META)?;
     let schema = number(required(meta.get("schema")?)?.value())?;
-    if schema == 4 {
-        read_binding(&meta)?;
-        native_application_state(tx)?;
-        for table in [HEADS, VERSIONS, EVENTS, OPERATIONS] {
-            if !tx.open_table(table)?.is_empty()? {
-                return Err(StoreError::CorruptRecord);
-            }
-        }
-        if !tx.open_table(APPLY_RESULTS)?.is_empty()? {
-            return Err(StoreError::CorruptRecord);
-        }
-        return Ok(());
-    }
-    if schema != 3 {
+    let native = schema == 4;
+    if !matches!(schema, 3 | 4) {
         return Ok(());
     }
     let binding = read_binding(&meta).map_err(|error| match error {
@@ -484,24 +649,85 @@ pub(crate) fn validate_application(tx: &ReadTransaction) -> Result<(), StoreErro
     let mut next = Some(0);
     let mut observed = None;
     let mut first_operations = BTreeMap::new();
+    let mut expected_membership = crate::CaptureMembership::new(None, vec![], BTreeMap::new())?;
+    let mut replay_operations = BTreeMap::<[u8; 16], (Vec<u8>, CreatedResult)>::new();
+    let mut replay_heads = BTreeMap::<Vec<u8>, [u8; 16]>::new();
+    let mut replay_versions = std::collections::BTreeSet::new();
     for row in results.iter()? {
         let (k, v) = row?;
-        let r = decode_apply(v.value())?;
-        if next != Some(k.value()) || r.log_id.index != k.value() {
+        let entry = decode_log(required(logs.get(k.value())?)?.value())?;
+        let r = if native {
+            let response = decode_response(v.value())?;
+            let native_entry = decode_native_entry(entry.clone())?;
+            validate_response_kind(&native_entry, &response)?;
+            if next != Some(k.value()) || response.log_id().index != k.value() {
+                return Err(StoreError::CorruptRecord);
+            }
+            observed = Some(*response.log_id());
+            next = k.value().checked_add(1);
+            match native_entry.payload {
+                openraft::EntryPayload::Membership(membership) => {
+                    expected_membership =
+                        crate::CaptureMembership::try_from(&openraft::StoredMembership::new(Some(entry.id.into()), membership))?;
+                    continue;
+                }
+                openraft::EntryPayload::Blank => continue,
+                openraft::EntryPayload::Normal(_) => CreatedApplyResult {
+                    log_id: *response.log_id(),
+                    result: response.created_result().cloned().ok_or(StoreError::CorruptRecord)?,
+                },
+            }
+        } else {
+            decode_apply(v.value())?
+        };
+        if !native && (next != Some(k.value()) || r.log_id.index != k.value()) {
             return Err(StoreError::CorruptRecord);
         }
-        let entry = decode_log(required(logs.get(k.value())?)?.value())?;
         if entry.id != r.log_id {
             return Err(StoreError::CorruptRecord);
         }
-        let c = DecideCreated::decode(&entry.payload)?;
+        let c = entry_command(&entry, native)?;
+        let canonical = c.encode()?;
         let op = decode_operation(required(operations.get(c.operation_id.as_slice())?)?.value())?;
-        let canonical_matches = op.canonical_command == entry.payload;
+        let canonical_matches = op.canonical_command == canonical;
         let expected = if canonical_matches {
             op.result
         } else {
             CreatedResult::OperationConflict
         };
+        if native {
+            let replay = if let Some((previous, result)) = replay_operations.get(&c.operation_id) {
+                if *previous == canonical {
+                    result.clone()
+                } else {
+                    CreatedResult::OperationConflict
+                }
+            } else {
+                let object = object_bytes(&c.object)?;
+                let key = version_key(&c.object, c.prepared.version_id)?;
+                let actual = replay_heads.get(&object).copied();
+                let result = if c.binding_revision != binding.revision {
+                    CreatedResult::BindingMismatch
+                } else if actual != c.expected_head {
+                    CreatedResult::HeadMismatch { actual_head: actual }
+                } else if replay_versions.contains(&key) {
+                    CreatedResult::VersionConflict
+                } else {
+                    replay_heads.insert(object, c.prepared.version_id);
+                    replay_versions.insert(key);
+                    CreatedResult::Created {
+                        version_id: c.prepared.version_id,
+                        event_id: c.operation_id,
+                        decision_log_id: r.log_id,
+                    }
+                };
+                replay_operations.insert(c.operation_id, (canonical.clone(), result.clone()));
+                result
+            };
+            if replay != r.result {
+                return Err(StoreError::CorruptRecord);
+            }
+        }
         if expected != r.result {
             return Err(StoreError::CorruptRecord);
         }
@@ -514,6 +740,19 @@ pub(crate) fn validate_application(tx: &ReadTransaction) -> Result<(), StoreErro
     if observed != last {
         return Err(StoreError::CorruptRecord);
     }
+    if native {
+        if native_application_state(tx)?.1 != expected_membership
+            || heads.len()? != u64::try_from(replay_heads.len()).map_err(|_| StoreError::CorruptRecord)?
+            || versions.len()? != u64::try_from(replay_versions.len()).map_err(|_| StoreError::CorruptRecord)?
+        {
+            return Err(StoreError::CorruptRecord);
+        }
+        for (object, id) in replay_heads {
+            if required(heads.get(object.as_slice())?)?.value() != id {
+                return Err(StoreError::CorruptRecord);
+            }
+        }
+    }
     for row in operations.iter()? {
         let (k, v) = row?;
         let op = decode_operation(v.value())?;
@@ -592,7 +831,7 @@ pub(crate) fn validate_application(tx: &ReadTransaction) -> Result<(), StoreErro
             return Err(StoreError::CorruptRecord);
         }
         let entry = decode_log(required(logs.get(version.decision_log_id.index)?)?.value())?;
-        let c = DecideCreated::decode(&entry.payload)?;
+        let c = entry_command(&entry, native)?;
         if entry.id != version.decision_log_id
             || c.object != object
             || c.prepared != version.prepared

crates/capture-authority/src/lib.rs

@@ -11,6 +11,7 @@ mod committed;
 mod created;
 mod node;
 mod raft_types;
+pub mod snapshot;
 pub use created::{
     CaptureBinding, CreatedApplyResult, CreatedEvent, CreatedOperation, CreatedResult, CreatedVersion, DecideCreated,
     MAX_OBJECT_KEY_LENGTH, ObjectIdentity, PreparedIdentity,
@@ -23,6 +24,135 @@ 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)),
+        }
+    }
+}
+
+/// Async native writes and readers sharing one durable Store.
+#[derive(Clone)]
+pub struct NativeLogWriter {
+    store: std::sync::Arc<std::sync::Mutex<Store>>,
+    ordering: std::sync::Arc<tokio::sync::Mutex<()>>,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum NativeWriteError {
+    #[error(transparent)]
+    Store(#[from] StoreError),
+    #[error("native log writer lock poisoned")]
+    Poisoned,
+    #[error("native log writer worker failed")]
+    Worker(#[source] tokio::task::JoinError),
+}
+
+impl NativeLogWriter {
+    pub fn new(store: Store) -> Self {
+        Self {
+            store: std::sync::Arc::new(std::sync::Mutex::new(store)),
+            ordering: std::sync::Arc::new(tokio::sync::Mutex::new(())),
+        }
+    }
+
+    pub fn reader(&self) -> NativeLogReader {
+        NativeLogReader {
+            store: self.store.clone(),
+        }
+    }
+
+    async fn write<F>(&self, write: F) -> Result<(), NativeWriteError>
+    where
+        F: FnOnce(&mut Store) -> Result<(), StoreError> + Send + 'static,
+    {
+        // Lock order is ordering then Store. The worker owns the ordering guard
+        // until commit completes, even when its awaiting caller is cancelled.
+        let ordering = self.ordering.clone().lock_owned().await;
+        let store = self.store.clone();
+        tokio::task::spawn_blocking(move || {
+            let _ordering = ordering;
+            let mut store = store.lock().map_err(|_| NativeWriteError::Poisoned)?;
+            write(&mut store).map_err(NativeWriteError::Store)
+        })
+        .await
+        .map_err(NativeWriteError::Worker)?
+    }
+
+    pub async fn save_vote(&self, vote: openraft::Vote<u64>) -> Result<(), NativeWriteError> {
+        self.write(move |store| store.save_vote(vote.into())).await
+    }
+
+    pub async fn read_vote(&self) -> Result<Option<openraft::Vote<u64>>, NativeWriteError> {
+        let store = self.store.clone();
+        tokio::task::spawn_blocking(move || {
+            let store = store.lock().map_err(|_| NativeWriteError::Poisoned)?;
+            store
+                .read_vote()
+                .map(|vote| vote.map(Into::into))
+                .map_err(NativeWriteError::Store)
+        })
+        .await
+        .map_err(NativeWriteError::Worker)?
+    }
+
+    pub async fn append_native(&self, entries: Vec<CaptureRaftEntry>) -> Result<(), NativeWriteError> {
+        self.write(move |store| store.append_native(&entries)).await
+    }
+
+    pub async fn truncate_suffix(&self, from: u64) -> Result<(), NativeWriteError> {
+        self.write(move |store| store.truncate_suffix(from)).await
+    }
+}
+
+#[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");
@@ -211,7 +341,11 @@ impl Store {
             let (key, value) = row?;
             let entry = decode_log(value.value())?;
             if next != Some(key.value()) || entry.id.index != key.value() {
-                return Err(StoreError::InvalidLogSequence);
+                return Err(if number(schema.value())? == 4 {
+                    StoreError::CorruptRecord
+                } else {
+                    StoreError::InvalidLogSequence
+                });
             }
             if number(schema.value())? == 4 {
                 application::decode_native_entry(entry.clone())?;
@@ -223,7 +357,11 @@ impl Store {
         application::validate_application(&tx)?;
         let tail = meta.get("tail")?.map(|v| decode_id(v.value())).transpose()?;
         if tail != last {
-            return Err(StoreError::InvalidLogSequence);
+            return Err(if number(schema.value())? == 4 {
+                StoreError::CorruptRecord
+            } else {
+                StoreError::InvalidLogSequence
+            });
         }
         drop(schema);
         drop(identity);
@@ -451,6 +589,294 @@ 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 u20_vote_durable_roundtrip() {
+        let (_dir, backend, store) = u17a_store();
+        let writer = NativeLogWriter::new(store);
+        let expected = openraft::Vote::from(vote());
+        u19_runtime().block_on(async {
+            writer.save_vote(expected).await.expect("U20_VOTE_DURABLE save");
+            assert_eq!(writer.read_vote().await.expect("read vote"), Some(expected), "U20_VOTE_DURABLE");
+        });
+        drop(writer);
+        backend.cut();
+        let reopened = Store::from_database(CrashBackend::new(&backend.path).database(), identity()).expect("reopen vote");
+        assert_eq!(reopened.read_vote().expect("durable vote"), Some(vote()), "U20_VOTE_DURABLE");
+    }
+
+    #[test]
+    fn u20_native_append_reader_visibility() {
+        use openraft::RaftLogReader;
+        let (_dir, backend, store) = u17a_store();
+        let writer = NativeLogWriter::new(store);
+        let entries = vec![u17a_blank(0), u17a_membership(1).0, u17a_created(2, u17a_command())];
+        let mut reader = writer.reader();
+        u19_runtime().block_on(async {
+            assert!(writer.append_native(entries.clone()).await.is_ok(), "U20_WRITE_SUCCESS");
+            assert_eq!(reader.try_get_log_entries(..).await.expect("shared reader"), entries);
+        });
+        drop(reader);
+        drop(writer);
+        backend.cut();
+        let reopened = Store::from_database(CrashBackend::new(&backend.path).database(), identity()).expect("reopen entries");
+        assert_eq!(reopened.read_native_range(..).expect("durable entries"), entries);
+    }
+
+    #[test]
+    fn u20_native_truncate_replacement() {
+        use openraft::RaftLogReader;
+        let (_dir, backend, mut store) = u17a_store();
+        let original = vec![u17a_blank(0), u17a_blank(1), u17a_blank(2)];
+        store.append_native(&original).expect("committed fixture entries");
+        store.save_committed(u17a_id(0)).expect("committed prefix");
+        let writer = NativeLogWriter::new(store);
+        let replacement = u17a_created(1, u17a_command());
+        let expected = vec![original[0].clone(), replacement.clone()];
+        u19_runtime().block_on(async {
+            assert!(matches!(
+                writer.truncate_suffix(0).await,
+                Err(NativeWriteError::Store(StoreError::CommittedLogProtected))
+            ));
+            writer.truncate_suffix(1).await.expect("truncate suffix");
+            writer.append_native(vec![replacement]).await.expect("replace suffix");
+            assert_eq!(writer.reader().try_get_log_entries(..).await.expect("replacement reader"), expected);
+        });
+        drop(writer);
+        backend.cut();
+        let reopened = Store::from_database(CrashBackend::new(&backend.path).database(), identity()).expect("reopen replacement");
+        assert_eq!(reopened.read_native_range(..).expect("durable replacement"), expected);
+    }
+
+    #[test]
+    fn u20_worker_thread() {
+        let (_dir, backend, store) = u17a_store();
+        let writer = NativeLogWriter::new(store);
+        backend.media.lock().expect("media").sync_threads.clear();
+        let executor = std::thread::current().id();
+        u19_runtime().block_on(writer.save_vote(vote().into())).expect("worker vote");
+        let media = backend.media.lock().expect("media");
+        assert!(!media.sync_threads.is_empty(), "U20_WORKER_THREAD no real sync");
+        assert!(media.sync_threads.iter().all(|thread| *thread != executor), "U20_WORKER_THREAD");
+    }
+
+    #[test]
+    fn u20_cancelled_write_stays_serialized() {
+        let (_dir, backend, store) = u17a_store();
+        let writer = NativeLogWriter::new(store);
+        let (entered, resume) = backend.pause_sync(1, false);
+        u19_runtime().block_on(async {
+            let first_writer = writer.clone();
+            let first = tokio::spawn(async move { first_writer.save_vote(vote().into()).await });
+            tokio::task::spawn_blocking(move || entered.wait())
+                .await
+                .expect("first sync entered");
+            first.abort();
+            assert!(first.await.expect_err("cancel first await").is_cancelled());
+            let held = writer.ordering.try_lock().is_err();
+            let second_writer = writer.clone();
+            let second = tokio::spawn(async move { second_writer.append_native(vec![u17a_blank(0)]).await });
+            tokio::task::yield_now().await;
+            let waiting = !second.is_finished();
+            tokio::task::spawn_blocking(move || resume.wait())
+                .await
+                .expect("release sync");
+            second.await.expect("second task").expect("second commit");
+            assert!(held && waiting, "U20_WRITE_ORDER cancelled caller released ordering");
+            assert_eq!(writer.read_vote().await.expect("first committed vote"), Some(vote().into()));
+        });
+    }
+
+    #[test]
+    fn u20_failed_commit_disables_mutations() {
+        let (_dir, backend, store) = u17a_store();
+        let writer = NativeLogWriter::new(store);
+        backend.arm(1, false, false);
+        u19_runtime().block_on(async {
+            assert!(
+                matches!(
+                    writer.save_vote(vote().into()).await,
+                    Err(NativeWriteError::Store(StoreError::CommitIndeterminate(_)))
+                ),
+                "U20_WRITE_FAILURE"
+            );
+            assert!(matches!(
+                writer.save_vote(vote().into()).await,
+                Err(NativeWriteError::Store(StoreError::WriteDisabled))
+            ));
+            assert!(matches!(
+                writer.append_native(vec![u17a_blank(0)]).await,
+                Err(NativeWriteError::Store(StoreError::WriteDisabled))
+            ));
+            assert!(matches!(
+                writer.truncate_suffix(0).await,
+                Err(NativeWriteError::Store(StoreError::WriteDisabled))
+            ));
+        });
+    }
+
+    #[test]
+    fn u20_worker_poison_errors() {
+        let (_dir, _backend, store) = u17a_store();
+        let writer = NativeLogWriter::new(store);
+        u19_runtime().block_on(async {
+            assert!(matches!(
+                writer.write(|_| panic!("injected worker panic")).await,
+                Err(NativeWriteError::Worker(_))
+            ));
+            assert!(matches!(writer.save_vote(vote().into()).await, Err(NativeWriteError::Poisoned)));
+            assert!(matches!(writer.append_native(vec![]).await, Err(NativeWriteError::Poisoned)));
+            assert!(matches!(writer.truncate_suffix(0).await, Err(NativeWriteError::Poisoned)));
+            assert!(matches!(writer.read_vote().await, Err(NativeWriteError::Poisoned)));
+        });
+    }
+
+    #[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,
@@ -527,6 +953,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);
@@ -621,6 +1130,7 @@ mod tests {
             let response = store.apply_next_native(u17a_id(index)).expect("exact replay");
             assert_eq!(store.native_apply_response(index).expect("receipt"), Some(response));
         }
+        store.enable_native_empty().expect("U17A_ENABLE_VALID_NATIVE_REPLAY");
         assert_eq!(backend.media.lock().expect("sync count").syncs, syncs);
         drop(store);
         let reopened = Store::open(&backend.path, identity()).expect("heterogeneous reopen");
@@ -898,6 +1408,56 @@ mod tests {
                 .expect("history frontier"),
             Some(u17a_id(3))
         );
+        let (_dir, backend, mut store) = u17a_store();
+        let first = u17a_command();
+        let mut rejected = first.clone();
+        rejected.prepared.version_id = [19; 16];
+        rejected.expected_head = Some(first.prepared.version_id);
+        let mut latest = rejected.clone();
+        latest.operation_id = [20; 16];
+        latest.prepared.version_id = [21; 16];
+        latest.prepared.preparation_id = [22; 16];
+        store
+            .append_native(&[
+                u17a_created(0, first),
+                u17a_created(1, rejected.clone()),
+                u17a_created(2, latest),
+            ])
+            .expect("rejected history batch");
+        store.save_committed(u17a_id(2)).expect("commit rejected history");
+        for index in 0..=2 {
+            store.apply_next_native(u17a_id(index)).expect("rejected history apply");
+        }
+        assert_eq!(
+            store
+                .native_apply_response(1)
+                .expect("rejected receipt")
+                .expect("receipt exists")
+                .created_result(),
+            Some(&CreatedResult::OperationConflict)
+        );
+        drop(store);
+        let mut store = Store::open(&backend.path, identity()).expect("valid rejected history reopen");
+        let tx = store.write_transaction().expect("orphan version transaction");
+        let mut key = crate::created::object_bytes(&rejected.object).expect("orphan key");
+        key.extend(rejected.prepared.version_id);
+        let mut value = Vec::new();
+        crate::created::put_prepared(&mut value, &rejected.prepared);
+        value.extend(encode_id(u17a_id(1)));
+        value.extend(u17a_binding().revision.to_be_bytes());
+        tx.open_table(application::VERSIONS)
+            .expect("version table")
+            .insert(key.as_slice(), value.as_slice())
+            .expect("inject syntactically valid rejected version");
+        store.commit(tx).expect("persist rejected version");
+        drop(store);
+        assert_eq!(
+            Store::open(&backend.path, identity())
+                .err()
+                .expect("U17A_REJECTED_VERSION_NOT_MATERIALIZED")
+                .code(),
+            "corrupt_record"
+        );
     }
     #[test]
     fn u17a_native_reopen_rejects_corruption() {
@@ -1229,6 +1789,7 @@ mod tests {
     }
     #[derive(Debug)]
     struct Media {
+        sync_threads: Vec<std::thread::ThreadId>,
         volatile: Vec<u8>,
         dead: bool,
         syncs: usize,
@@ -1250,6 +1811,7 @@ mod tests {
             Self {
                 path: path.to_owned(),
                 media: Arc::new(Mutex::new(Media {
+                    sync_threads: Vec::new(),
                     volatile: bytes,
                     dead: false,
                     syncs: 0,
@@ -1327,6 +1889,7 @@ mod tests {
                 return Err(io::Error::other("power lost"));
             }
             m.syncs += 1;
+            m.sync_threads.push(std::thread::current().id());
             let fault = m.fault.filter(|(n, _, _)| *n == m.syncs);
             let pause = if m.pause.as_ref().is_some_and(|(n, _, _, _)| *n == m.syncs) {
                 m.pause.take()