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/lib.rs

@@ -11,7 +11,6 @@ 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,
@@ -24,135 +23,6 @@ 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");
@@ -589,294 +459,286 @@ 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:?}"),
+    fn u17b_snapshot(store: &Store) -> Vec<Vec<(Vec<u8>, Vec<u8>)>> {
+        use crate::application::{APPLY_RESULTS, EVENTS, HEADS, OPERATIONS, VERSIONS};
+        let tx = store.database.begin_read().expect("snapshot transaction");
+        let meta = tx.open_table(META).expect("snapshot metadata");
+        let mut snapshot = vec![
+            meta.iter()
+                .expect("metadata iterator")
+                .map(|row| {
+                    let (key, value) = row.expect("metadata row");
+                    (key.value().as_bytes().to_vec(), value.value().to_vec())
+                })
+                .collect(),
+        ];
+        for table in [LOGS, APPLY_RESULTS] {
+            snapshot.push(
+                tx.open_table(table)
+                    .expect("snapshot indexed table")
+                    .iter()
+                    .expect("indexed iterator")
+                    .map(|row| {
+                        let (key, value) = row.expect("indexed row");
+                        (key.value().to_be_bytes().to_vec(), value.value().to_vec())
+                    })
+                    .collect(),
+            );
         }
-    }
-
-    #[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"
+        for table in [HEADS, VERSIONS, EVENTS, OPERATIONS] {
+            snapshot.push(
+                tx.open_table(table)
+                    .expect("snapshot business table")
+                    .iter()
+                    .expect("business iterator")
+                    .map(|row| {
+                        let (key, value) = row.expect("business row");
+                        (key.value().to_vec(), value.value().to_vec())
+                    })
+                    .collect(),
             );
-            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"),
-        );
+        snapshot
     }
-
-    #[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(..)
+    fn u17b_history() -> (tempfile::TempDir, CrashBackend, Store, Vec<DecideCreated>) {
+        let (dir, backend, mut store) = u17a_bound();
+        let first = u17a_command();
+        let mut rejected = first.clone();
+        rejected.operation_id = [10; 16];
+        rejected.prepared.version_id = [11; 16];
+        let mut second = first.clone();
+        second.operation_id = [12; 16];
+        second.prepared.version_id = [13; 16];
+        second.expected_head = Some(first.prepared.version_id);
+        second.prepared.content_length = 8193;
+        let mut suffix = second.clone();
+        suffix.operation_id = [14; 16];
+        suffix.prepared.version_id = [15; 16];
+        suffix.expected_head = Some(second.prepared.version_id);
+        let commands = vec![first.clone(), first, rejected, second, suffix];
+        let entries = commands
+            .iter()
+            .enumerate()
+            .map(|(index, command)| LogEntry {
+                id: u17a_id(u64::try_from(index).expect("fixture index")),
+                payload: command.encode().expect("canonical legacy Created"),
             })
-            .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);
-        });
+            .collect::<Vec<_>>();
+        store.append(&entries).expect("legacy history and suffix");
+        store.save_committed(u17a_id(4)).expect("committed suffix");
+        for index in 0..4 {
+            store.apply_next_created(u17a_id(index)).expect("legacy history apply");
+        }
+        assert_eq!(committed_schema(&store), 3);
+        (dir, backend, store, commands)
     }
-
     #[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 u17b_native_migration_populated_suffix() {
+        let (_dir, backend, mut store, commands) = u17b_history();
+        let before = u17b_snapshot(&store);
+        let results = (0..4)
+            .map(|index| {
+                store
+                    .created_apply_result(index)
+                    .expect("legacy result")
+                    .expect("applied result")
+            })
+            .collect::<Vec<_>>();
+        assert!(
+            store.migrate_legacy_created_to_native().is_ok(),
+            "U17B_MIGRATION_SUCCESS canonical populated schema3 must migrate"
+        );
+        assert_eq!(committed_schema(&store), 4);
+        assert_eq!(&u17b_snapshot(&store)[3..], &before[3..], "business bytes preserved");
+        assert_eq!(store.capture_binding().expect("binding"), Some(u17a_binding()));
+        assert_eq!(store.log_state().expect("tail").last, Some(u17a_id(4)));
+        assert_eq!(store.read_committed().expect("committed"), Some(u17a_id(4)));
+        assert_eq!(
+            store.native_application_state().expect("application pair"),
+            (
+                Some(u17a_id(3)),
+                CaptureMembership::new(None, vec![], std::collections::BTreeMap::new()).expect("empty membership")
+            )
+        );
+        assert_eq!(
+            store.read_native_entries(0, 5).expect("all entries"),
+            commands
+                .iter()
+                .enumerate()
+                .map(|(index, command)| u17a_created(u64::try_from(index).expect("index"), command.clone()))
+                .collect::<Vec<_>>()
+        );
+        for result in results {
+            assert_eq!(
+                store.created_apply_result(result.log_id.index).expect("migrated result"),
+                Some(result.clone())
+            );
+            assert_eq!(
+                store
+                    .native_apply_response(result.log_id.index)
+                    .expect("native result")
+                    .expect("response")
+                    .created_result(),
+                Some(&result.result)
+            );
+        }
+        assert!(store.native_apply_response(4).expect("suffix response").is_none());
+        assert_eq!(
+            store
+                .created_operation(commands[0].operation_id)
+                .expect("first operation")
+                .expect("operation")
+                .result,
+            CreatedResult::Created {
+                version_id: commands[0].prepared.version_id,
+                event_id: commands[0].operation_id,
+                decision_log_id: u17a_id(0)
+            }
+        );
+        drop(store);
+        let mut store = Store::open(&backend.path, identity()).expect("migrated reopen");
+        assert_eq!(&u17b_snapshot(&store)[3..], &before[3..]);
+        let response = store.apply_next_native(u17a_id(4)).expect("real suffix application");
+        assert_eq!(
+            response.created_result(),
+            Some(&CreatedResult::Created {
+                version_id: commands[4].prepared.version_id,
+                event_id: commands[4].operation_id,
+                decision_log_id: u17a_id(4)
+            })
+        );
+        assert_eq!(
+            store.created_head(&commands[4].object).expect("suffix head"),
+            Some(commands[4].prepared.version_id)
+        );
+        assert_eq!(store.last_applied().expect("suffix frontier"), Some(u17a_id(4)));
+    }
+    #[test]
+    fn u17b_native_migration_opaque_refusal() {
+        let canonical = u17a_command().encode().expect("canonical command");
+        let mut noncanonical = canonical.clone();
+        noncanonical.push(0);
+        for payload in [
+            b"opaque payload".to_vec(),
+            canonical[..canonical.len() - 1].to_vec(),
+            noncanonical,
+        ] {
+            let (_dir, backend, mut store) = u17a_bound();
+            store
+                .append(&[LogEntry { id: u17a_id(0), payload }])
+                .expect("invalid legacy suffix append");
+            let before = u17b_snapshot(&store);
+            assert!(store.migrate_legacy_created_to_native().is_err(), "invalid payload must refuse");
+            assert_eq!(u17b_snapshot(&store), before, "refusal changes no table bytes");
+            drop(store);
+            let store = Store::open(&backend.path, identity()).expect("refused suffix reopen");
+            assert_eq!(u17b_snapshot(&store), before);
+            assert_eq!(committed_schema(&store), 3);
+        }
+    }
+    #[test]
+    fn u17b_native_migration_empty_idempotent() {
+        let (_dir, backend, store) = u17a_bound();
+        drop(store);
+        let mut store = Store::from_database(backend.database(), identity()).expect("open legacy without upgrade");
+        assert_eq!(committed_schema(&store), 3);
+        store.migrate_legacy_created_to_native().expect("empty migration");
+        assert_eq!(committed_schema(&store), 4);
+        let before = u17b_snapshot(&store);
+        backend.media.lock().expect("reset migration syncs").syncs = 0;
+        // Reopen through the instrumented backend before checking the no-commit path.
+        drop(store);
+        let mut store = Store::from_database(backend.database(), identity()).expect("native reopen");
+        backend.media.lock().expect("reset reopen syncs").syncs = 0;
+        store.migrate_legacy_created_to_native().expect("native repeat");
+        assert_eq!(backend.media.lock().expect("repeat syncs").syncs, 0, "repeat must not commit");
+        assert_eq!(u17b_snapshot(&store), before);
+        store.writable = false;
+        assert_eq!(
+            store
+                .migrate_legacy_created_to_native()
+                .expect_err("disabled even on schema4")
+                .code(),
+            "write_disabled"
+        );
+    }
+    #[test]
+    fn u17b_native_migration_ack_and_cut() {
+        let (_dir, backend, mut store, _) = u17b_history();
+        let old = u17b_snapshot(&store);
+        backend.media.lock().expect("reset migration syncs").syncs = 0;
+        store.migrate_legacy_created_to_native().expect("migration ACK");
+        let reached = backend.media.lock().expect("actual migration syncs").syncs;
+        assert!(reached > 0, "migration must reach durability IO");
+        let new = u17b_snapshot(&store);
+        backend.cut();
+        drop(store);
+        assert_eq!(u17b_snapshot(&Store::open(&backend.path, identity()).expect("ACK recovery")), new);
+        for sync in 1..=reached {
+            for after in [false, true] {
+                let (_dir, backend, mut store, _) = u17b_history();
+                assert_eq!(u17b_snapshot(&store), old);
+                let (entered, resume) = backend.pause_sync(sync, after);
+                let writer = std::thread::spawn(move || {
+                    let result = store.migrate_legacy_created_to_native();
+                    (store, result)
+                });
+                entered.wait();
+                backend.cut();
+                resume.wait();
+                let (store, result) = writer.join().expect("migration cut writer");
+                drop(store);
+                if let Some(store) = u17a_cut_recovery(&backend) {
+                    let actual = u17b_snapshot(&store);
+                    assert!(
+                        actual == old || actual == new,
+                        "migration cut must never mix schema/log/result/business tables"
+                    );
+                    if result.is_ok() {
+                        assert_eq!(actual, new, "ACK cannot recover old state");
+                    }
+                } else {
+                    assert!(result.is_err(), "ACK recovery must succeed");
+                }
+            }
+        }
+        let (_dir, backend, mut store, _) = u17b_history();
+        backend.arm(1, false, false);
+        assert_eq!(
+            store.migrate_legacy_created_to_native().expect_err("uncertain commit").code(),
+            "commit_indeterminate"
+        );
+        assert_eq!(
+            store
+                .migrate_legacy_created_to_native()
+                .expect_err("migration disabled")
+                .code(),
+            "write_disabled"
+        );
+        assert_eq!(store.append_native(&[]).expect_err("native append disabled").code(), "write_disabled");
+        assert_eq!(store.append(&[]).expect_err("legacy append disabled").code(), "write_disabled");
+        assert_eq!(
+            store.apply_next_native(u17a_id(4)).expect_err("native apply disabled").code(),
+            "write_disabled"
+        );
+        assert_eq!(
+            store
+                .apply_next_created(u17a_id(4))
+                .expect_err("legacy apply disabled")
+                .code(),
+            "write_disabled"
+        );
+        assert_eq!(
+            store
+                .pin_capture_binding(u17a_binding())
+                .expect_err("binding disabled")
+                .code(),
+            "write_disabled"
+        );
+        assert_eq!(store.save_committed(u17a_id(4)).expect_err("committed disabled").code(), "write_disabled");
+        assert_eq!(store.save_vote(vote()).expect_err("vote disabled").code(), "write_disabled");
+        assert_eq!(store.truncate_suffix(5).expect_err("truncate disabled").code(), "write_disabled");
     }
-
     fn u17a_id(index: u64) -> LogId {
         LogId {
             term: 11,
@@ -953,89 +815,6 @@ 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);
@@ -1789,7 +1568,6 @@ mod tests {
     }
     #[derive(Debug)]
     struct Media {
-        sync_threads: Vec<std::thread::ThreadId>,
         volatile: Vec<u8>,
         dead: bool,
         syncs: usize,
@@ -1811,7 +1589,6 @@ mod tests {
             Self {
                 path: path.to_owned(),
                 media: Arc::new(Mutex::new(Media {
-                    sync_threads: Vec::new(),
                     volatile: bytes,
                     dead: false,
                     syncs: 0,
@@ -1889,7 +1666,6 @@ 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()

docs/architecture/README.md

@@ -32,7 +32,6 @@ Required headings and strings in these files are asserted by `scripts/check_arch
 
 | Document | Use this when |
 |---|---|
-| [capture-checkpoint-contract.md](capture-checkpoint-contract.md) | changing Capture checkpoint capture, provenance, installation, recovery, or log cleanup |
 | [local-descriptor-cache.md](local-descriptor-cache.md) | changing descriptor cache keys/open modes or invalidation after local mutations |
 | [erasure-coding.md](erasure-coding.md) | changing anything under `crates/ecstore/src/erasure/`, `crates/filemeta/`, `crates/ecstore/src/set_disk/`, storage-class or layout code, or any decode, quorum, or heal boundary (normative spec) |
 | [placement-repair-invariants.md](placement-repair-invariants.md) | changing anything that resolves an object to a pool, set, or disk, or that admits scanner or heal work |

docs/architecture/capture-checkpoint-contract.md

@@ -1,211 +0,0 @@
-**Use this when:** changing Capture checkpoint capture, provenance, installation, recovery, or log cleanup.
-
-# Capture checkpoint contract
-
-This document defines the requirements for a recoverable Capture checkpoint.
-It is a design contract, not evidence that production capture, installation,
-purge, quorum integration, or S3 behavior implements these requirements.
-
-## Existing evidence and the missing boundary
-
-In `crates/capture-authority/src/application.rs`, `apply_created` inserts one
-canonical command and result on the first decision for an operation ID. An
-identical command returns that stored result; a different command returns
-`OperationConflict` without replacing the canonical record. The first result
-remains authoritative even when later state would produce a different result.
-
-`CreatedOperation` in `crates/capture-authority/src/created.rs` contains command
-bytes and a result. Only `CreatedResult::Created` carries `decision_log_id`.
-Negative results carry no first-decision ID. Heads, versions, events, and
-operations therefore cannot reconstruct the source of every first decision.
-For example, a stored `HeadMismatch` records an observed head, but does not
-prove that the head existed at the original decision cut. Later retries do
-not supply the missing historical evidence.
-
-`validate_application` in `crates/capture-authority/src/application.rs` derives
-first operation IDs from retained log/application records and, for native
-schema 4, replays decisions from the beginning using in-memory maps and sets.
-That intact-prefix validator is evidence for existing behavior; it is not a
-checkpoint-baseline validator or a bounded recovery algorithm.
-
-The `Row` comment and `StagingSink` in
-`crates/capture-authority/src/snapshot.rs` reserve operation provenance for
-caller validation and require isolated staging. Rows remain opaque to the
-framing decoder. A framing digest detects byte corruption; it does not establish
-semantic consistency, historical truth, or authority to install a checkpoint.
-
-## Canonical operation provenance
-
-Checkpoint-capable storage must use an explicitly versioned operation envelope
-containing the full first-decision `LogId` and the existing canonical operation
-bytes, including their stored result. Preserve those bytes and result; do not
-recompute a negative result against checkpoint-time heads or binding state.
-The envelope version is a storage capability boundary, distinct from the
-snapshot framing version. The existing CSNP v1 framing and header must not be
-silently extended. Header capability metadata requires a declared framing
-revision or an authenticated, versioned manifest whose consistency with the
-captured header is validated. Distinct storage and framing versions do not
-permit silently changing the v1 format.
-
-The first insertion must atomically record the envelope, application response,
-applied position, and any successful head/version/event changes in the same
-write transaction. Its first-decision ID must identify that insertion's log
-entry. An identical-command retry returns the original result and leaves both
-bytes and first-decision ID unchanged. A conflicting-command retry records its
-own application response as `OperationConflict`; it cannot overwrite the
-canonical envelope or claim to be the first decision.
-
-```mermaid
-flowchart TD
-    A[Committed command and operation ID] --> B{Canonical operation exists?}
-    B -->|No| C[Evaluate once at this log ID]
-    C --> D[Atomically persist first ID, original bytes and result]
-    B -->|Yes| E{Same canonical command bytes?}
-    E -->|Yes| F[Return original result; preserve first ID]
-    E -->|No| G[Return OperationConflict; preserve canonical record]
-```
-
-## Compatibility and promotion
-
-The existing schema 3 to schema 4 migration remains unchanged. Adding provenance
-requires a separately declared storage capability with explicit readers, writers,
-upgrade, and downgrade rules. A schema 4 label alone must not imply provenance
-support. Unknown envelope versions and stores missing required provenance must
-be rejected by checkpoint installation and purge authorization.
-
-An old schema 4 store may backfill first IDs only from its complete, intact,
-validated prefix: replay and validate the original command/results, identify the
-first canonical insertion for each operation, then atomically persist the
-capability and envelopes. A retry's ID or a successful event's ID cannot stand
-in for missing negative provenance. If the prefix has already been removed or
-cannot be validated, refuse promotion and cleanup; do not invent IDs. Readers
-without the new capability must refuse the upgraded format rather than silently
-interpret it as the old operation encoding. Retaining all history forever is
-not the checkpoint design: validated promotion supplies a recoverable baseline
-that permits bounded suffix retention.
-
-## Capture cut and resource ownership
-
-Capture must read the header, heads, versions, events, and operations through
-one owned read transaction. The view must retain the owning store/database
-handle for the entire transaction, cursor, and output lifetime; an owned
-transaction type alone does not prove database lifetime safety. Keep the store
-`Arc` alive so dropping another handle cannot invalidate the view. These are
-ownership requirements, not a claim about unverified library behavior.
-
-The header must describe that same committed, applied cut with its full `LogId`,
-full effective membership including its decision position and node metadata,
-cluster identity, binding identity/revision, and supported storage/envelope
-capabilities. Membership and binding must come from the captured transaction,
-not from a later live read. Never capture beyond the durable committed position.
-A four-table view without this cut and metadata is not independently installable.
-
-Emit rows incrementally in deterministic section/key order. Bound row size,
-section and total bytes, counts, temporary disk consumption, and concurrent
-capture admission. Use checked arithmetic. Do not collect the whole snapshot
-in RAM. Cancellation, disk exhaustion, or limit failure must leave the live
-store unchanged and release task-owned staging resources.
-
-## Isolated validation and trust
-
-Decode only into isolated disk staging. Check framing, limits, section order,
-unique keys, complete header metadata, capability versions, cluster/binding
-compatibility, canonical command/result encoding, envelope IDs at or before the
-cut, and all head/version/event/operation relationships. A successful operation
-must agree with its version, event, and first decision. Negative operations must
-have no success event or side effect attributable to that decision.
-
-These checks still cannot prove the historical predicate behind a negative
-result. Neither a digest nor a first ID proves that a `HeadMismatch`,
-`BindingMismatch`, or `VersionConflict` was justified when first decided.
-Remote installation therefore requires an explicitly authenticated and
-authorized Raft authority for the claimed cluster, cut, and membership, or a
-complete historical proof validated from a trusted baseline. Reject input that
-has neither. Authentication and authorization must precede publication and
-purge; a caller's assertion of trust is insufficient. This document does not
-claim that this remote Raft authentication boundary is implemented.
-
-Replay validation must use disk-backed staging/indexes and bounded buffers,
-including duplicate-operation and version/head tracking. Putting the input on
-disk while retaining replay maps proportional to history does not meet this
-requirement. The existing `validate_application` RAM replay cannot simply be
-reused to claim bounded checkpoint installation.
-
-## Atomic publication and recovery
-
-After isolated validation, publication must have one durable atomic point that
-selects the complete checkpoint generation, its application state, cut,
-membership, binding, capability, and recovery metadata. A pointer switch is
-acceptable only when all referenced bytes are already durable and recovery
-cannot observe a partial generation. Serialize publication with application so
-no committed suffix write is lost between validation and the switch.
-
-Preserve local node identity and persisted vote/hard state. Remote checkpoint
-metadata cannot overwrite either. Validate cluster identity, binding, full cut
-identity, and membership before replacing state. Reject rollback behind the
-local durable application/checkpoint floor; an equal index with a different
-term/leader identity is a conflict, not an equivalent cut. Retain valid local
-suffix and commit state consistently with the installed baseline, rather than
-resetting them from remote metadata.
-
-On reopen, select only a fully published generation and verify its recoverable
-baseline before replay. The suffix starts exactly at cut.index + 1, with checked
-index arithmetic; the complete cut ID must agree wherever the boundary is
-retained. Validate contiguous suffix indices, entry identities, committed and
-applied bounds, typed application responses, canonical retry behavior, and
-membership transitions relative to checkpoint state. Reject gaps, conflicting
-boundary IDs, stale membership, or a suffix belonging to another cluster or
-binding. Recovery must begin from checkpoint heads and canonical operations,
-not from empty state or the assumption that the retained log starts at zero.
-
-Purge is authorized only after the published checkpoint can be reopened and
-validated as a recovery baseline. Delete only entries/results covered by that
-exact cut, retaining everything needed for the continuous suffix and local
-Raft durability. Re-check the selected generation and cut immediately before
-cleanup. Persist a recoverable purge boundary so interrupted cleanup is
-idempotent. Never delete the only evidence before provenance promotion or
-checkpoint durability succeeds.
-
-```mermaid
-flowchart TD
-    A[One owned transaction and store lifetime] --> B[Stream captured cut to isolated disk staging]
-    B --> C[Validate framing, semantics, provenance and authorized source]
-    C --> D[Make complete generation durable]
-    D --> E[Single atomic publication point]
-    E --> F[Reopen baseline and validate continuous suffix]
-    F --> G[Authorize purge only through that exact cut]
-    G --> H[Idempotent cleanup; preserve recoverable baseline and suffix]
-    B -->|Failure| I[Discard staging; live generation remains usable]
-    C -->|Failure| I
-    D -->|Before publication crash| I
-```
-
-A crash before publication leaves the previous generation recoverable. A crash
-after publication must recover the new complete baseline, including when purge
-has not begun or is only partly complete. Orphan staging is never selected as
-live state. Old generations may be reclaimed only after no active view needs
-them and the selected generation remains recoverable.
-
-## Behavioral verification requirements
-
-The following are required behavior checks, not reports of tests run or gates
-passed. Existing symbols above anchor the semantics; future checkpoint paths
-must demonstrate the additional requirements directly.
-
-| Behavior | Required observation |
-|---|---|
-| First negative result | Original command/result and first full ID survive capture, install, and reopen; later state does not re-evaluate it. |
-| Identical and conflicting retries | `apply_created` semantics hold across the checkpoint boundary: original result/ID retained; conflict cannot overwrite the canonical record. |
-| Old schema 4 promotion | An intact prefix validates and supplies first IDs; missing/corrupt prefix refuses promotion and purge; schema 3 to 4 behavior stays unchanged. |
-| Snapshot cut under concurrent writes | Header, membership, binding, and all rows describe one transaction cut; later writes remain in the suffix. |
-| View lifetime | Dropping other store handles during row emission cannot invalidate the retained transaction or database. |
-| Framing versus authority | Valid digest with forged negative history is insufficient; unauthenticated or unauthorized remote input is rejected. |
-| Cross-record corruption | Altered command, first ID, event, version, head, membership, or binding fails isolated validation before publication. |
-| Recovery boundary | Exact cut plus continuous suffix recovers; gaps, conflicting full IDs, cross-cluster input, and rollback are rejected. |
-| Local Raft state | Installation preserves local identity, vote/hard state, and compatible durable suffix/commit information. |
-| Resource exhaustion | Oversize rows, arithmetic overflow, disk exhaustion, cancellation, and large histories fail safely within declared memory/disk bounds. |
-| Crash publication | Crashes before and after the atomic point select a complete old or new generation, never a mixture. |
-| Interrupted purge | Reopen succeeds before cleanup is authorized; each interruption resumes without deleting the recoverable baseline or required suffix. |
-
-Documentation path and whitespace guards only validate document hygiene. They
-cannot establish these production behaviors or the remote trust boundary.