test(capture): specify atomic populated native migration

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

@@ -459,6 +459,286 @@ mod tests {
             node_id: 0,
         }
     }
+    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(),
+            );
+        }
+        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(),
+            );
+        }
+        snapshot
+    }
+    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"),
+            })
+            .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 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,
@@ -535,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);