#![cfg(feature = "test-faults")]
use std::io::Write;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use armdb::armour::Db;
use armdb::{
Codec, CollectionMeta, Config, DbError, FormatAction, FormatMigration, MigrateAction, NoHook,
RapiraCodec, TxConfig, serialize_entry,
};
use armour_core::GetType;
use tempfile::tempdir;
use zerocopy::{FromBytes, Immutable, IntoBytes};
const NAME: &str = "format_replay_write";
static WIDEN_CALLS: AtomicUsize = AtomicUsize::new(0);
#[derive(Clone, Copy, Debug, FromBytes, GetType, Immutable, IntoBytes, PartialEq, Eq)]
#[repr(C)]
struct V1 {
value: u32,
}
#[derive(
Clone, Copy, Debug, FromBytes, GetType, Immutable, IntoBytes, PartialEq, Eq, rapira::Rapira,
)]
#[repr(C)]
struct FloorRow {
value: u64,
}
impl CollectionMeta for FloorRow {
type SelfId = [u8; 8];
const NAME: &'static str = "format_floor_family_matrix";
}
impl CollectionMeta for V1 {
type SelfId = [u8; 8];
const NAME: &'static str = NAME;
const VERSION: u16 = 1;
}
#[derive(Clone, Copy, Debug, FromBytes, GetType, Immutable, IntoBytes, PartialEq, Eq)]
#[repr(C)]
struct V2 {
value: u64,
}
fn widen(_key: &[u8], old: &[u8], out: &mut Vec<u8>) -> armdb::DbResult<FormatAction> {
WIDEN_CALLS.fetch_add(1, Ordering::SeqCst);
let value = u32::from_ne_bytes(old.try_into().expect("V1 width"));
out.extend_from_slice(&(value as u64).to_ne_bytes());
Ok(FormatAction::Write)
}
const WIDEN: FormatMigration = FormatMigration::raw_checked(1, &V1::TYPE, &V2::TYPE, widen);
impl CollectionMeta for V2 {
type SelfId = [u8; 8];
const NAME: &'static str = NAME;
const VERSION: u16 = 2;
const FORMAT_MIGRATIONS: &'static [FormatMigration] = &[WIDEN];
}
#[derive(Clone, Copy, Debug, FromBytes, GetType, Immutable, IntoBytes, PartialEq, Eq)]
#[repr(C)]
struct V2Keep {
value: u32,
}
fn keep(_key: &[u8], _old: &[u8], _out: &mut Vec<u8>) -> armdb::DbResult<FormatAction> {
Ok(FormatAction::Keep)
}
const KEEP: FormatMigration = FormatMigration::raw_checked(1, &V1::TYPE, &V2Keep::TYPE, keep);
impl CollectionMeta for V2Keep {
type SelfId = [u8; 8];
const NAME: &'static str = NAME;
const VERSION: u16 = 2;
const FORMAT_MIGRATIONS: &'static [FormatMigration] = &[KEEP];
}
#[derive(Clone, Copy, Debug, FromBytes, GetType, Immutable, IntoBytes, PartialEq, Eq)]
#[repr(C)]
struct V2Drop {
value: u64,
}
const MIXED_SUFFIX_NAME: &str = "format_replay_mixed_suffix";
#[derive(Clone, Debug, GetType, PartialEq, Eq, rapira::Rapira)]
struct MixedSuffixV1 {
value: u32,
}
impl CollectionMeta for MixedSuffixV1 {
type SelfId = [u8; 8];
const NAME: &'static str = MIXED_SUFFIX_NAME;
const VERSION: u16 = 1;
}
#[derive(Clone, Debug, GetType, PartialEq, Eq, rapira::Rapira)]
struct MixedSuffixV3 {
value: u64,
generation: u16,
}
fn widen_mixed_suffix(_key: &[u8], old: &[u8], out: &mut Vec<u8>) -> armdb::DbResult<FormatAction> {
let old: MixedSuffixV1 = RapiraCodec.decode_from(old)?;
RapiraCodec.encode_to(
&MixedSuffixV3 {
value: u64::from(old.value),
generation: 2,
},
out,
)?;
Ok(FormatAction::Write)
}
const WIDEN_MIXED_SUFFIX: FormatMigration = FormatMigration::raw_checked(
1,
&MixedSuffixV1::TYPE,
&MixedSuffixV3::TYPE,
widen_mixed_suffix,
);
impl CollectionMeta for MixedSuffixV3 {
type SelfId = [u8; 8];
const NAME: &'static str = MIXED_SUFFIX_NAME;
const VERSION: u16 = 3;
const FORMAT_MIGRATIONS: &'static [FormatMigration] = &[WIDEN_MIXED_SUFFIX];
}
fn finish_mixed_suffix(_key: &[u8; 8], old: &MixedSuffixV3) -> MigrateAction<MixedSuffixV3> {
MigrateAction::Update(MixedSuffixV3 {
value: old.value,
generation: 3,
})
}
fn drop_pending(_key: &[u8], _old: &[u8], _out: &mut Vec<u8>) -> armdb::DbResult<FormatAction> {
Ok(FormatAction::Drop)
}
const DROP: FormatMigration =
FormatMigration::raw_checked(1, &V1::TYPE, &V2Drop::TYPE, drop_pending);
impl CollectionMeta for V2Drop {
type SelfId = [u8; 8];
const NAME: &'static str = NAME;
const VERSION: u16 = 2;
const FORMAT_MIGRATIONS: &'static [FormatMigration] = &[DROP];
}
const SEQUENTIAL_NAME: &str = "format_sequential_high_water";
#[derive(Clone, Copy, Debug, FromBytes, GetType, Immutable, IntoBytes, PartialEq, Eq)]
#[repr(C)]
struct SequentialV1 {
first: u64,
}
impl CollectionMeta for SequentialV1 {
type SelfId = [u8; 8];
const NAME: &'static str = SEQUENTIAL_NAME;
const VERSION: u16 = 1;
}
#[derive(Clone, Copy, Debug, FromBytes, GetType, Immutable, IntoBytes, PartialEq, Eq)]
#[repr(C)]
struct SequentialV2 {
second: u64,
}
#[derive(Clone, Copy, Debug, FromBytes, GetType, Immutable, IntoBytes, PartialEq, Eq)]
#[repr(C)]
struct SequentialV3 {
third: u64,
}
fn sequential_keep(_key: &[u8], _old: &[u8], _out: &mut Vec<u8>) -> armdb::DbResult<FormatAction> {
Ok(FormatAction::Keep)
}
const SEQUENTIAL_1_TO_2: FormatMigration =
FormatMigration::raw_checked(1, &SequentialV1::TYPE, &SequentialV2::TYPE, sequential_keep);
const SEQUENTIAL_2_TO_3: FormatMigration =
FormatMigration::raw_checked(2, &SequentialV2::TYPE, &SequentialV3::TYPE, sequential_keep);
impl CollectionMeta for SequentialV2 {
type SelfId = [u8; 8];
const NAME: &'static str = SEQUENTIAL_NAME;
const VERSION: u16 = 2;
const FORMAT_MIGRATIONS: &'static [FormatMigration] = &[SEQUENTIAL_1_TO_2];
}
impl CollectionMeta for SequentialV3 {
type SelfId = [u8; 8];
const NAME: &'static str = SEQUENTIAL_NAME;
const VERSION: u16 = 3;
const FORMAT_MIGRATIONS: &'static [FormatMigration] = &[SEQUENTIAL_1_TO_2, SEQUENTIAL_2_TO_3];
}
const CODEC_ONLY_NAME: &str = "format_replay_codec_only";
const HISTORICAL_AMBIGUITY_NAME: &str = "format_replay_historical_ambiguity";
const RAW_HISTORY_NAME: &str = "format_replay_raw_history";
#[derive(Clone, Copy, Debug, FromBytes, GetType, Immutable, IntoBytes, PartialEq, Eq)]
#[repr(C)]
struct CodecOnlyV1(u64);
impl CollectionMeta for CodecOnlyV1 {
type SelfId = [u8; 8];
const NAME: &'static str = CODEC_ONLY_NAME;
const VERSION: u16 = 1;
}
#[derive(Clone, Copy, Debug, FromBytes, Immutable, IntoBytes, PartialEq, Eq)]
#[repr(C)]
struct CodecOnlyV2(u64);
impl GetType for CodecOnlyV2 {
const TYPE: armour_core::Typ = CodecOnlyV1::TYPE;
}
fn change_codec_bytes(_key: &[u8], old: &[u8], out: &mut Vec<u8>) -> armdb::DbResult<FormatAction> {
let value = u64::from_ne_bytes(old.try_into().expect("CodecOnlyV1 width"));
out.extend_from_slice(&(value + 1_000).to_ne_bytes());
Ok(FormatAction::Write)
}
const CHANGE_CODEC_BYTES: FormatMigration = FormatMigration::raw_checked(
1,
&CodecOnlyV1::TYPE,
&CodecOnlyV2::TYPE,
change_codec_bytes,
);
impl CollectionMeta for CodecOnlyV2 {
type SelfId = [u8; 8];
const NAME: &'static str = CODEC_ONLY_NAME;
const VERSION: u16 = 2;
const FORMAT_MIGRATIONS: &'static [FormatMigration] = &[CHANGE_CODEC_BYTES];
}
#[derive(Clone, Copy, Debug, FromBytes, GetType, Immutable, IntoBytes, PartialEq, Eq)]
#[repr(C)]
struct RawHistoryV1(u32);
impl CollectionMeta for RawHistoryV1 {
type SelfId = [u8; 8];
const NAME: &'static str = RAW_HISTORY_NAME;
const VERSION: u16 = 1;
}
#[derive(Clone, Copy, Debug, FromBytes, GetType, Immutable, IntoBytes, PartialEq, Eq)]
#[repr(C)]
struct RawHistoryV2(u64);
fn widen_raw_history(_key: &[u8], old: &[u8], out: &mut Vec<u8>) -> armdb::DbResult<FormatAction> {
let value = u32::from_ne_bytes(old.try_into().expect("RawHistoryV1 width"));
out.extend_from_slice(&u64::from(value).to_ne_bytes());
Ok(FormatAction::Write)
}
const WIDEN_RAW_HISTORY: FormatMigration = FormatMigration::raw(1, widen_raw_history);
impl CollectionMeta for RawHistoryV2 {
type SelfId = [u8; 8];
const NAME: &'static str = RAW_HISTORY_NAME;
const VERSION: u16 = 2;
const FORMAT_MIGRATIONS: &'static [FormatMigration] = &[WIDEN_RAW_HISTORY];
}
#[derive(Clone, Copy, Debug, FromBytes, GetType, Immutable, IntoBytes, PartialEq, Eq)]
#[repr(C)]
struct RawHistoryInvalidV3 {
high: u64,
low: u64,
}
impl CollectionMeta for RawHistoryInvalidV3 {
type SelfId = [u8; 8];
const NAME: &'static str = RAW_HISTORY_NAME;
const VERSION: u16 = 3;
const FORMAT_MIGRATIONS: &'static [FormatMigration] = &[WIDEN_RAW_HISTORY];
}
#[derive(Clone, Copy, Debug, FromBytes, GetType, Immutable, IntoBytes, PartialEq, Eq)]
#[repr(C)]
struct HistoricalA1(u32);
impl CollectionMeta for HistoricalA1 {
type SelfId = [u8; 8];
const NAME: &'static str = HISTORICAL_AMBIGUITY_NAME;
const VERSION: u16 = 1;
}
#[derive(Clone, Copy, Debug, FromBytes, GetType, Immutable, IntoBytes, PartialEq, Eq)]
#[repr(C)]
struct HistoricalB2(u64);
fn historical_a_to_b(_key: &[u8], old: &[u8], out: &mut Vec<u8>) -> armdb::DbResult<FormatAction> {
let value = u32::from_ne_bytes(old.try_into().expect("HistoricalA1 width"));
out.extend_from_slice(&(value as u64).to_ne_bytes());
Ok(FormatAction::Write)
}
const HISTORICAL_A_TO_B: FormatMigration = FormatMigration::raw_checked(
1,
&HistoricalA1::TYPE,
&HistoricalB2::TYPE,
historical_a_to_b,
);
impl CollectionMeta for HistoricalB2 {
type SelfId = [u8; 8];
const NAME: &'static str = HISTORICAL_AMBIGUITY_NAME;
const VERSION: u16 = 2;
const FORMAT_MIGRATIONS: &'static [FormatMigration] = &[HISTORICAL_A_TO_B];
}
#[derive(Clone, Copy, Debug, FromBytes, Immutable, IntoBytes, PartialEq, Eq)]
#[repr(C)]
struct HistoricalA3(u32);
impl GetType for HistoricalA3 {
const TYPE: armour_core::Typ = HistoricalA1::TYPE;
}
fn historical_b_to_a(_key: &[u8], old: &[u8], out: &mut Vec<u8>) -> armdb::DbResult<FormatAction> {
let value = u64::from_ne_bytes(old.try_into().expect("HistoricalB2 width"));
out.extend_from_slice(&(value as u32).to_ne_bytes());
Ok(FormatAction::Write)
}
const HISTORICAL_B_TO_A: FormatMigration = FormatMigration::raw_checked(
2,
&HistoricalB2::TYPE,
&HistoricalA3::TYPE,
historical_b_to_a,
);
impl CollectionMeta for HistoricalA3 {
type SelfId = [u8; 8];
const NAME: &'static str = HISTORICAL_AMBIGUITY_NAME;
const VERSION: u16 = 3;
const FORMAT_MIGRATIONS: &'static [FormatMigration] = &[HISTORICAL_A_TO_B, HISTORICAL_B_TO_A];
}
const MIXED_A_NAME: &str = "format_replay_mixed_a";
const MIXED_B_NAME: &str = "format_replay_mixed_b";
#[derive(Clone, Copy, Debug, FromBytes, GetType, Immutable, IntoBytes, PartialEq, Eq)]
#[repr(C)]
struct MixedA1 {
value: u32,
}
impl CollectionMeta for MixedA1 {
type SelfId = [u8; 8];
const NAME: &'static str = MIXED_A_NAME;
const VERSION: u16 = 1;
}
#[derive(Clone, Copy, Debug, FromBytes, GetType, Immutable, IntoBytes, PartialEq, Eq)]
#[repr(C)]
struct MixedA2 {
value: u64,
}
fn widen_mixed_a(_key: &[u8], old: &[u8], out: &mut Vec<u8>) -> armdb::DbResult<FormatAction> {
let value = u32::from_ne_bytes(old.try_into().expect("MixedA1 width"));
out.extend_from_slice(&(value as u64).to_ne_bytes());
Ok(FormatAction::Write)
}
const WIDEN_MIXED_A: FormatMigration =
FormatMigration::raw_checked(1, &MixedA1::TYPE, &MixedA2::TYPE, widen_mixed_a);
impl CollectionMeta for MixedA2 {
type SelfId = [u8; 8];
const NAME: &'static str = MIXED_A_NAME;
const VERSION: u16 = 2;
const FORMAT_MIGRATIONS: &'static [FormatMigration] = &[WIDEN_MIXED_A];
}
#[derive(Clone, Copy, Debug, FromBytes, GetType, Immutable, IntoBytes, PartialEq, Eq)]
#[repr(C)]
struct MixedB1 {
value: u32,
}
impl CollectionMeta for MixedB1 {
type SelfId = [u8; 8];
const NAME: &'static str = MIXED_B_NAME;
const VERSION: u16 = 1;
}
#[derive(Clone, Copy, Debug, FromBytes, GetType, Immutable, IntoBytes, PartialEq, Eq)]
#[repr(C)]
struct MixedB2 {
value: u64,
}
fn widen_mixed_b(_key: &[u8], old: &[u8], out: &mut Vec<u8>) -> armdb::DbResult<FormatAction> {
let value = u32::from_ne_bytes(old.try_into().expect("MixedB1 width"));
out.extend_from_slice(&(value as u64).to_ne_bytes());
Ok(FormatAction::Write)
}
const WIDEN_MIXED_B: FormatMigration =
FormatMigration::raw_checked(1, &MixedB1::TYPE, &MixedB2::TYPE, widen_mixed_b);
impl CollectionMeta for MixedB2 {
type SelfId = [u8; 8];
const NAME: &'static str = MIXED_B_NAME;
const VERSION: u16 = 2;
const FORMAT_MIGRATIONS: &'static [FormatMigration] = &[WIDEN_MIXED_B];
}
const UNKNOWN_NAME: &str = "format_replay_unknown";
#[derive(Clone, Copy, Debug, FromBytes, GetType, Immutable, IntoBytes, PartialEq, Eq)]
#[repr(C)]
struct LoggedUnknownV1 {
value: i32,
}
impl CollectionMeta for LoggedUnknownV1 {
type SelfId = [u8; 8];
const NAME: &'static str = UNKNOWN_NAME;
const VERSION: u16 = 1;
}
#[derive(Clone, Copy, Debug, FromBytes, GetType, Immutable, IntoBytes, PartialEq, Eq)]
#[repr(C)]
struct ExpectedV1 {
value: u32,
}
#[derive(Clone, Copy, Debug, FromBytes, GetType, Immutable, IntoBytes, PartialEq, Eq)]
#[repr(C)]
struct ExpectedV2 {
value: u64,
}
fn widen_expected(_key: &[u8], old: &[u8], out: &mut Vec<u8>) -> armdb::DbResult<FormatAction> {
let value = u32::from_ne_bytes(old.try_into().expect("ExpectedV1 width"));
out.extend_from_slice(&(value as u64).to_ne_bytes());
Ok(FormatAction::Write)
}
const WIDEN_EXPECTED: FormatMigration =
FormatMigration::raw_checked(1, &ExpectedV1::TYPE, &ExpectedV2::TYPE, widen_expected);
impl CollectionMeta for ExpectedV2 {
type SelfId = [u8; 8];
const NAME: &'static str = UNKNOWN_NAME;
const VERSION: u16 = 2;
const FORMAT_MIGRATIONS: &'static [FormatMigration] = &[WIDEN_EXPECTED];
}
fn snapshot_txlog(root: &Path) -> Vec<(String, Vec<u8>)> {
let txlog = root.join("txlog");
let mut entries = std::fs::read_dir(&txlog)
.expect("txlog directory")
.collect::<Result<Vec<_>, _>>()
.expect("txlog entries");
entries.sort_by_key(std::fs::DirEntry::file_name);
entries
.into_iter()
.map(|entry| {
assert!(entry.file_type().expect("txlog file type").is_file());
(
entry.file_name().to_string_lossy().into_owned(),
std::fs::read(entry.path()).expect("txlog file bytes"),
)
})
.collect()
}
fn snapshot_database(root: &Path) -> Vec<(PathBuf, Vec<u8>)> {
fn visit(root: &Path, path: &Path, files: &mut Vec<(PathBuf, Vec<u8>)>) {
for entry in std::fs::read_dir(path)
.expect("database directory")
.collect::<Result<Vec<_>, _>>()
.expect("database entries")
{
let path = entry.path();
if path.is_dir() {
visit(root, &path, files);
} else {
files.push((
path.strip_prefix(root)
.expect("path below database root")
.to_owned(),
std::fs::read(&path).expect("database file bytes"),
));
}
}
}
let mut files = Vec::new();
visit(root, root, &mut files);
files.sort_by(|left, right| left.0.cmp(&right.0));
files
}
fn assert_database_unchanged(root: &Path, before: &[(PathBuf, Vec<u8>)], context: &str) {
let after = snapshot_database(root);
if after == before {
return;
}
let mut changed = before
.iter()
.filter_map(|(path, bytes)| {
(after
.iter()
.find(|(candidate, _)| candidate == path)
.map(|(_, bytes)| bytes)
!= Some(bytes))
.then_some(path.clone())
})
.collect::<Vec<_>>();
changed.extend(
after
.iter()
.filter(|(path, _)| !before.iter().any(|(candidate, _)| candidate == path))
.map(|(path, _)| path.clone()),
);
changed.sort();
changed.dedup();
panic!("{context}; changed files: {changed:?}");
}
fn only_data_file(root: &Path) -> PathBuf {
only_data_file_named(root, NAME)
}
fn only_data_file_named(root: &Path, name: &str) -> PathBuf {
fn visit(path: &Path, found: &mut Vec<PathBuf>) {
for entry in std::fs::read_dir(path).expect("read collection directory") {
let path = entry.expect("directory entry").path();
if path.is_dir() {
visit(&path, found);
} else if path
.extension()
.is_some_and(|extension| extension == "data")
{
found.push(path);
}
}
}
let mut files = Vec::new();
visit(&root.join(name), &mut files);
assert_eq!(files.len(), 1, "one-shard test fixture");
files.pop().expect("data file")
}
fn codec_only_pending_fixture() -> tempfile::TempDir {
let dir = tempdir().unwrap();
{
let db = Arc::new(Db::open_test(dir.path()).unwrap());
let tree = db
.open_zero_tree::<CodecOnlyV1, 8, _>(Config::test(), NoHook, &[])
.unwrap();
tree.put(&1u64.to_be_bytes(), &CodecOnlyV1(7)).unwrap();
Db::tx_debug_skip_apply(true);
db.tx()
.put(&tree, &2u64.to_be_bytes(), &CodecOnlyV1(9))
.unwrap()
.durable()
.commit()
.unwrap();
Db::tx_debug_skip_apply(false);
drop(tree);
db.close().unwrap();
}
assert_eq!(CodecOnlyV1::TYPE.h(), CodecOnlyV2::TYPE.h());
dir
}
fn raw_history_pending_v1_fixture() -> tempfile::TempDir {
let dir = tempdir().expect("tempdir");
{
let db = Arc::new(Db::open_test(dir.path()).expect("open raw V1 db"));
let tree = db
.open_zero_tree::<RawHistoryV1, 4, _>(Config::test(), NoHook, &[])
.expect("open raw V1 tree");
tree.put(&1u64.to_be_bytes(), &RawHistoryV1(1))
.expect("seed raw V1 row");
Db::tx_debug_skip_apply(true);
db.tx()
.put(&tree, &2u64.to_be_bytes(), &RawHistoryV1(2))
.expect("stage raw V1 pending row")
.durable()
.commit()
.expect("commit raw V1 pending row");
Db::tx_debug_skip_apply(false);
db.close().expect("close raw V1 db");
}
dir
}
fn raw_history_pending_v2_fixture() -> tempfile::TempDir {
let dir = tempdir().expect("tempdir");
{
let db = Arc::new(Db::open_test(dir.path()).expect("open raw V1 db"));
let tree = db
.open_zero_tree::<RawHistoryV1, 4, _>(Config::test(), NoHook, &[])
.expect("open raw V1 tree");
tree.put(&1u64.to_be_bytes(), &RawHistoryV1(1))
.expect("seed raw V1 row");
drop(tree);
db.close().expect("close raw V1 db");
}
{
let db = Arc::new(Db::open_test(dir.path()).expect("open migration db"));
db.migrate_format::<RawHistoryV2>(&Config::test())
.expect("install raw V2 format without pending history");
let tree = db
.open_zero_tree::<RawHistoryV2, 8, _>(Config::test(), NoHook, &[])
.expect("open raw V2 tree");
Db::tx_debug_skip_apply(true);
db.tx()
.put(&tree, &2u64.to_be_bytes(), &RawHistoryV2(2))
.expect("stage current-format pending row")
.durable()
.commit()
.expect("commit current-format pending row");
Db::tx_debug_skip_apply(false);
db.close().expect("close raw V2 db");
}
dir
}
fn add_recoverable_torn_raw_history_migration_output(root: &Path) {
let migration = root.join(RAW_HISTORY_NAME).join("shard_000.mig");
std::fs::create_dir(&migration).expect("create torn migration output");
std::fs::write(migration.join("format.v"), b"AFV1\x02")
.expect("write malformed migration marker");
}
#[test]
fn bitcask_data_enumeration_propagates_an_individual_entry_error_without_swap() {
let dir = tempdir().expect("tempdir");
{
let db = Arc::new(Db::open_test(dir.path()).expect("open raw V1 db"));
let tree = db
.open_zero_tree::<RawHistoryV1, 4, _>(Config::test(), NoHook, &[])
.expect("open raw V1 tree");
tree.put(&1u64.to_be_bytes(), &RawHistoryV1(41))
.expect("seed raw V1 row");
drop(tree);
db.close().expect("close raw V1 db");
}
let source = dir.path().join(RAW_HISTORY_NAME).join("shard_000");
let before = snapshot_database(&source);
armdb::format_migration::failpoints::fail_next_read_dir_entry(&source);
let db = Arc::new(Db::open_test(dir.path()).expect("open migration db"));
let error = match db.open_zero_tree::<RawHistoryV2, 8, _>(Config::test(), NoHook, &[]) {
Ok(_) => panic!("an individual directory-entry failure must abort migration"),
Err(error) => error,
};
assert!(
matches!(error, DbError::Io(ref error) if error.to_string().contains("read_dir entry")),
"{error}"
);
assert_eq!(
snapshot_database(&source),
before,
"source shard changed after enumeration failure"
);
assert!(!dir.path().join(format!("{RAW_HISTORY_NAME}.old")).exists());
assert!(!dir.path().join(format!("{RAW_HISTORY_NAME}.mig")).exists());
}
fn historical_ambiguity_at_v2_fixture() -> tempfile::TempDir {
let dir = tempdir().expect("tempdir");
let key = 3u64.to_be_bytes();
{
let db = Arc::new(Db::open_test(dir.path()).expect("open historical V1 db"));
let tree = db
.open_zero_tree::<HistoricalA1, 4, _>(Config::test(), NoHook, &[])
.expect("open historical V1 tree");
tree.put(&1u64.to_be_bytes(), &HistoricalA1(1))
.expect("seed physical V1 row");
Db::tx_debug_skip_apply(true);
db.tx()
.put(&tree, &key, &HistoricalA1(3))
.expect("stage historical A operation")
.durable()
.commit()
.expect("commit historical A operation");
Db::tx_debug_skip_apply(false);
db.close().expect("close historical V1 db");
}
{
let db = Db::open_test(dir.path()).expect("reopen historical V1 db");
db.migrate_format::<HistoricalB2>(&Config::test())
.expect("install unambiguous V2 physical format");
assert_eq!(
db.db_info().collections[HISTORICAL_AMBIGUITY_NAME].version,
2
);
assert_eq!(db.pending_transactions().len(), 1);
db.close().expect("close stored V2 fixture");
}
assert_eq!(HistoricalA1::TYPE.h(), HistoricalA3::TYPE.h());
assert_ne!(HistoricalA1::TYPE.h(), HistoricalB2::TYPE.h());
dir
}
#[test]
fn historical_replay_ambiguity_refuses_before_physical_rewrite() {
let dir = historical_ambiguity_at_v2_fixture();
let db = Arc::new(Db::open_test(dir.path()).expect("open stored V2 db"));
let before = snapshot_database(dir.path());
let error = match db.open_zero_tree::<HistoricalA3, 4, _>(Config::test(), NoHook, &[]) {
Ok(_) => panic!("A -> B -> A replay chain must be ambiguous"),
Err(error) => error,
};
assert!(
matches!(error, DbError::TxFormatAmbiguous { .. }),
"{error}"
);
assert_eq!(
db.db_info().collections[HISTORICAL_AMBIGUITY_NAME].version,
2,
"ambiguity admission must precede the physical rewrite"
);
assert_database_unchanged(
dir.path(),
&before,
"fail-closed ambiguity mutated physical or transaction bytes",
);
}
#[test]
fn retained_raw_history_rejects_a_current_target_pending_operation() {
let dir = raw_history_pending_v2_fixture();
let before = snapshot_database(dir.path());
let db = Arc::new(Db::open_test(dir.path()).expect("reopen raw V2 db"));
let error = match db.open_zero_tree::<RawHistoryV2, 8, _>(Config::test(), NoHook, &[]) {
Ok(_) => panic!("raw history has no version qualifier even at the target fingerprint"),
Err(error) => error,
};
assert!(
matches!(error, DbError::TxFormatAmbiguous { .. }),
"{error}"
);
assert_database_unchanged(
dir.path(),
&before,
"raw target ambiguity must fail before applying logged bytes",
);
}
#[test]
fn managed_raw_history_ambiguity_preserves_recoverable_torn_migration_output() {
let dir = raw_history_pending_v1_fixture();
add_recoverable_torn_raw_history_migration_output(dir.path());
let before = snapshot_database(dir.path());
let db = Arc::new(Db::open_test(dir.path()).expect("reopen raw V1 db"));
let error = match db.open_zero_tree::<RawHistoryV2, 8, _>(Config::test(), NoHook, &[]) {
Ok(_) => panic!("raw-history replay must be rejected before torn-output recovery"),
Err(error) => error,
};
assert!(
matches!(error, DbError::TxFormatAmbiguous { .. }),
"{error}"
);
assert_database_unchanged(
dir.path(),
&before,
"managed ambiguity recovery mutated a recoverable torn migration output",
);
}
#[test]
fn discard_policy_does_not_mutate_txlog_before_invalid_chain_rejection() {
let dir = raw_history_pending_v1_fixture();
let before = snapshot_database(dir.path());
let db = Arc::new(
Db::open_with_tx_config(
dir.path(),
TxConfig {
discard_pending_on_schema_mismatch: true,
..TxConfig::default()
},
)
.expect("open invalid-chain db"),
);
let error = match db.open_zero_tree::<RawHistoryInvalidV3, 16, _>(Config::test(), NoHook, &[]) {
Ok(_) => panic!("an incomplete format chain must be rejected"),
Err(error) => error,
};
assert!(matches!(
error,
DbError::SchemaMismatch {
kind: armdb::SchemaMismatchKind::MissingStep { from: 2 },
..
}
));
assert_database_unchanged(
dir.path(),
&before,
"invalid-chain rejection mutated database or txlog bytes",
);
}
#[test]
fn discard_policy_does_not_mutate_txlog_before_refuse_policy_rejection() {
let dir = raw_history_pending_v1_fixture();
let before = snapshot_database(dir.path());
let db = Arc::new(
Db::open_with_tx_config(
dir.path(),
TxConfig {
discard_pending_on_schema_mismatch: true,
..TxConfig::default()
},
)
.expect("open refuse db"),
);
let mut config = Config::test();
config.format_migration = armdb::FormatMigrationPolicy::Refuse;
let error = match db.open_zero_tree::<RawHistoryV2, 8, _>(config, NoHook, &[]) {
Ok(_) => panic!("Refuse policy must reject the migration"),
Err(error) => error,
};
assert!(matches!(error, DbError::FormatMigrationRequired { .. }));
assert_database_unchanged(
dir.path(),
&before,
"Refuse policy mutated database or txlog bytes",
);
}
#[test]
fn explicit_migration_bypasses_refuse_policy() {
let dir = tempdir().expect("tempdir");
{
let db = Arc::new(Db::open_test(dir.path()).expect("open raw V1 db"));
let tree = db
.open_zero_tree::<RawHistoryV1, 4, _>(Config::test(), NoHook, &[])
.expect("open raw V1 tree");
tree.put(&1u64.to_be_bytes(), &RawHistoryV1(41))
.expect("seed raw V1 row");
db.close().expect("close raw V1 db");
}
let db = Db::open_test(dir.path()).expect("open explicit refuse db");
let mut config = Config::test();
config.format_migration = armdb::FormatMigrationPolicy::Refuse;
let report = db
.migrate_format::<RawHistoryV2>(&config)
.expect("explicit executor must bypass the managed-open Refuse policy");
assert_eq!((report.from, report.to), (1, 2));
}
#[test]
fn explicit_migration_rejects_historical_ambiguity_before_physical_rewrite() {
let dir = historical_ambiguity_at_v2_fixture();
let db = Db::open_test(dir.path()).expect("open stored V2 db");
let before = snapshot_database(dir.path());
let error = match db.migrate_format::<HistoricalA3>(&Config::test()) {
Ok(_) => panic!("ambiguous history must be admitted before explicit rewrite"),
Err(error) => error,
};
assert!(
matches!(error, DbError::TxFormatAmbiguous { .. }),
"{error}"
);
assert_database_unchanged(dir.path(), &before, "explicit ambiguity rewrote storage");
}
#[test]
fn explicit_raw_history_ambiguity_preserves_recoverable_torn_migration_output() {
let dir = raw_history_pending_v1_fixture();
add_recoverable_torn_raw_history_migration_output(dir.path());
let before = snapshot_database(dir.path());
let db = Db::open_test(dir.path()).expect("reopen raw V1 db");
let error = match db.migrate_format::<RawHistoryV2>(&Config::test()) {
Ok(_) => panic!("raw-history replay must be rejected before torn-output recovery"),
Err(error) => error,
};
assert!(
matches!(error, DbError::TxFormatAmbiguous { .. }),
"{error}"
);
assert_database_unchanged(
dir.path(),
&before,
"explicit ambiguity recovery mutated a recoverable torn migration output",
);
}
#[test]
fn codec_only_explicit_migration_rejects_ambiguity_before_physical_rewrite() {
let dir = codec_only_pending_fixture();
let db = Arc::new(Db::open_test(dir.path()).unwrap());
let before = snapshot_database(dir.path());
let error = match db.migrate_format::<CodecOnlyV2>(&Config::test()) {
Ok(_) => panic!("same-fingerprint old bytes must be rejected before rewrite"),
Err(error) => error,
};
assert!(
error.to_string().contains("ambiguous"),
"codec-only replay refusal must explain the unqualified tx bytes: {error}"
);
assert_eq!(db.db_info().collections[CODEC_ONLY_NAME].version, 1);
assert_database_unchanged(dir.path(), &before, "codec ambiguity rewrote storage");
}
#[test]
fn codec_only_format_migration_durably_drains_ambiguous_pending_bytes_when_configured() {
let dir = codec_only_pending_fixture();
let tx_config = TxConfig {
discard_pending_on_schema_mismatch: true,
..TxConfig::default()
};
{
let db = Arc::new(Db::open_with_tx_config(dir.path(), tx_config).unwrap());
let tree = db
.open_zero_tree::<CodecOnlyV2, 8, _>(Config::test(), NoHook, &[])
.expect("explicit discard policy drains ambiguous records before migration");
assert_eq!(tree.get(&1u64.to_be_bytes()), Some(CodecOnlyV2(1_007)));
assert_eq!(
tree.get(&2u64.to_be_bytes()),
None,
"old-codec pending bytes must never be applied directly"
);
assert!(db.pending_transactions().is_empty());
drop(tree);
db.close().unwrap();
}
let db = Arc::new(Db::open_with_tx_config(dir.path(), tx_config).unwrap());
let tree = db
.open_zero_tree::<CodecOnlyV2, 8, _>(Config::test(), NoHook, &[])
.expect("drain must survive restart");
assert_eq!(tree.get(&1u64.to_be_bytes()), Some(CodecOnlyV2(1_007)));
assert_eq!(tree.get(&2u64.to_be_bytes()), None);
assert!(db.pending_transactions().is_empty());
}
#[test]
fn format_migration_replay_write_transforms_a_pending_put() {
WIDEN_CALLS.store(0, Ordering::SeqCst);
let dir = tempdir().expect("tempdir");
let key = 7u64.to_be_bytes();
{
let db = Arc::new(Db::open_test(dir.path()).expect("open V1 db"));
let tree = db
.open_zero_tree::<V1, { std::mem::size_of::<V1>() }, _>(Config::test(), NoHook, &[])
.expect("open V1 tree");
Db::tx_debug_skip_apply(true);
db.tx()
.put(&tree, &key, &V1 { value: 7 })
.expect("stage put")
.durable()
.commit()
.expect("commit into txlog");
Db::tx_debug_skip_apply(false);
assert!(tree.get(&key).is_none(), "apply is suppressed");
db.close().expect("close V1 db");
}
let db = Db::open_test(dir.path()).expect("reopen V2 db");
let tree = db
.open_zero_tree::<V2, { std::mem::size_of::<V2>() }, _>(Config::test(), NoHook, &[])
.expect("pending V1 Put is transformed during replay");
assert_eq!(tree.get(&key), Some(V2 { value: 7 }));
assert_eq!(
WIDEN_CALLS.load(Ordering::SeqCst),
1,
"replay transforms once"
);
}
#[test]
fn format_migration_replay_delete_skips_the_value_transformer() {
let dir = tempdir().expect("tempdir");
let key = 8u64.to_be_bytes();
{
let db = Arc::new(Db::open_test(dir.path()).expect("open V1 db"));
let tree = db
.open_zero_tree::<V1, { std::mem::size_of::<V1>() }, _>(Config::test(), NoHook, &[])
.expect("open V1 tree");
tree.put(&key, &V1 { value: 8 }).expect("seed row");
Db::tx_debug_skip_apply(true);
db.tx()
.delete(&tree, &key)
.expect("stage delete")
.durable()
.commit()
.expect("commit delete into txlog");
Db::tx_debug_skip_apply(false);
db.close().expect("close V1 db");
}
let db = Db::open_test(dir.path()).expect("reopen V2 db");
let tree = db
.open_zero_tree::<V2, { std::mem::size_of::<V2>() }, _>(Config::test(), NoHook, &[])
.expect("Delete never enters the V1 value transformer");
assert!(tree.get(&key).is_none());
}
#[test]
fn explicit_format_migration_replays_pending_old_put_and_delete() {
let dir = tempdir().expect("tempdir");
let put_key = 80u64.to_be_bytes();
let delete_key = 81u64.to_be_bytes();
{
let db = Arc::new(Db::open_test(dir.path()).expect("open V1 db"));
let tree = db
.open_zero_tree::<V1, { std::mem::size_of::<V1>() }, _>(Config::test(), NoHook, &[])
.expect("open V1 tree");
tree.put(&delete_key, &V1 { value: 81 })
.expect("seed row to delete");
Db::tx_debug_skip_apply(true);
db.tx()
.put(&tree, &put_key, &V1 { value: 80 })
.expect("stage old Put")
.delete(&tree, &delete_key)
.expect("stage old Delete")
.durable()
.commit()
.expect("commit old operations");
Db::tx_debug_skip_apply(false);
db.close().expect("close V1 db");
}
let logged = snapshot_txlog(dir.path());
let db = Db::open_test(dir.path()).expect("reopen for explicit migration");
db.migrate_format::<V2>(&Config::test())
.expect("install V2 physical format without replay");
assert_eq!(
snapshot_txlog(dir.path()),
logged,
"migration rewrote txlog"
);
let tree = db
.open_zero_tree::<V2, { std::mem::size_of::<V2>() }, _>(Config::test(), NoHook, &[])
.expect("reconstruct replay-only V1 to V2 plan");
assert_eq!(tree.get(&put_key), Some(V2 { value: 80 }));
assert!(tree.get(&delete_key).is_none());
assert_eq!(snapshot_txlog(dir.path()), logged, "replay rewrote txlog");
}
#[test]
fn explicit_format_migration_preserves_replay_plan_across_value_suffix() {
let dir = tempdir().expect("tempdir");
let key = 82u64.to_be_bytes();
{
let db = Arc::new(Db::open_test(dir.path()).expect("open mixed V1 db"));
let tree = db
.open_typed_tree::<MixedSuffixV1, RapiraCodec, _>(Config::test(), NoHook, &[])
.expect("open mixed V1 tree");
Db::tx_debug_skip_apply(true);
db.tx()
.put(&tree, &key, &MixedSuffixV1 { value: 82 })
.expect("stage old mixed Put")
.durable()
.commit()
.expect("commit old mixed Put");
Db::tx_debug_skip_apply(false);
db.close().expect("close mixed V1 db");
}
let migrations = &[(
2,
finish_mixed_suffix as fn(&[u8; 8], &MixedSuffixV3) -> MigrateAction<MixedSuffixV3>,
)];
let db = Db::open_test(dir.path()).expect("reopen for mixed explicit migration");
db.migrate_format::<MixedSuffixV3>(&Config::test())
.expect("install only the format prefix");
let tree = db
.open_typed_tree::<MixedSuffixV3, RapiraCodec, _>(Config::test(), NoHook, migrations)
.expect("replay old bytes before running the typed value suffix");
assert_eq!(
tree.get(&key).map(|value| value.clone_value()),
Some(MixedSuffixV3 {
value: 82,
generation: 3,
})
);
}
#[test]
fn format_migration_replay_keep_reuses_logged_bytes() {
let dir = tempdir().unwrap();
let key = 9u64.to_be_bytes();
{
let db = Arc::new(Db::open_test(dir.path()).unwrap());
let tree = db
.open_zero_tree::<V1, { std::mem::size_of::<V1>() }, _>(Config::test(), NoHook, &[])
.unwrap();
Db::tx_debug_skip_apply(true);
db.tx()
.put(&tree, &key, &V1 { value: 9 })
.unwrap()
.durable()
.commit()
.unwrap();
Db::tx_debug_skip_apply(false);
db.close().unwrap();
}
let db = Db::open_test(dir.path()).unwrap();
let tree = db
.open_zero_tree::<V2Keep, { std::mem::size_of::<V2Keep>() }, _>(Config::test(), NoHook, &[])
.unwrap();
assert_eq!(tree.get(&key), Some(V2Keep { value: 9 }));
}
#[test]
fn format_migration_replay_drop_skips_the_pending_put() {
let dir = tempdir().unwrap();
let key = 10u64.to_be_bytes();
{
let db = Arc::new(Db::open_test(dir.path()).unwrap());
let tree = db
.open_zero_tree::<V1, { std::mem::size_of::<V1>() }, _>(Config::test(), NoHook, &[])
.unwrap();
Db::tx_debug_skip_apply(true);
db.tx()
.put(&tree, &key, &V1 { value: 10 })
.unwrap()
.durable()
.commit()
.unwrap();
Db::tx_debug_skip_apply(false);
db.close().unwrap();
}
let db = Db::open_test(dir.path()).unwrap();
db.migrate_format::<V2Drop>(&Config::test()).unwrap();
let tree = db
.open_zero_tree::<V2Drop, { std::mem::size_of::<V2Drop>() }, _>(Config::test(), NoHook, &[])
.unwrap();
assert!(tree.get(&key).is_none());
}
#[test]
fn format_migration_write_calls_transform_once_per_physical_row() {
let dir = tempdir().expect("tempdir");
let key = 11u64.to_be_bytes();
WIDEN_CALLS.store(0, Ordering::SeqCst);
{
let db = Db::open_test(dir.path()).expect("open V1 db");
let tree = db
.open_zero_tree::<V1, { std::mem::size_of::<V1>() }, _>(Config::test(), NoHook, &[])
.expect("open V1 tree");
tree.put(&key, &V1 { value: 11 }).expect("seed V1 row");
db.close().expect("close V1 db");
}
let db = Db::open_test(dir.path()).expect("reopen V2 db");
let tree = db
.open_zero_tree::<V2, { std::mem::size_of::<V2>() }, _>(Config::test(), NoHook, &[])
.expect("migrate physical row");
assert_eq!(tree.get(&key), Some(V2 { value: 11 }));
assert_eq!(
WIDEN_CALLS.load(Ordering::SeqCst),
1,
"a non-idempotent format callback runs once per physical row"
);
}
#[test]
fn format_migration_replay_already_applied_put_skips_transform() {
let dir = tempdir().unwrap();
let key = 12u64.to_be_bytes();
WIDEN_CALLS.store(0, Ordering::SeqCst);
{
let db = Arc::new(Db::open_test(dir.path()).unwrap());
let tree = db
.open_zero_tree::<V1, { std::mem::size_of::<V1>() }, _>(Config::test(), NoHook, &[])
.unwrap();
db.tx()
.put(&tree, &key, &V1 { value: 12 })
.unwrap()
.durable()
.commit()
.unwrap();
db.close().unwrap();
}
let db = Db::open_test(dir.path()).unwrap();
let tree = db
.open_zero_tree::<V2, { std::mem::size_of::<V2>() }, _>(Config::test(), NoHook, &[])
.unwrap();
assert_eq!(tree.get(&key), Some(V2 { value: 12 }));
assert_eq!(
WIDEN_CALLS.load(Ordering::SeqCst),
1,
"only the physical migration transforms"
);
}
#[test]
fn format_migration_replay_mixed_record_waits_for_each_collection_open() {
let dir = tempdir().expect("tempdir");
let a_key = 13u64.to_be_bytes();
let b_key = 14u64.to_be_bytes();
{
let db = Arc::new(Db::open_test(dir.path()).expect("open source db"));
let a = db
.open_zero_tree::<MixedA1, { std::mem::size_of::<MixedA1>() }, _>(
Config::test(),
NoHook,
&[],
)
.expect("open source A");
let b = db
.open_zero_tree::<MixedB1, { std::mem::size_of::<MixedB1>() }, _>(
Config::test(),
NoHook,
&[],
)
.expect("open source B");
Db::tx_debug_skip_apply(true);
db.tx()
.put(&a, &a_key, &MixedA1 { value: 13 })
.expect("stage A")
.put(&b, &b_key, &MixedB1 { value: 14 })
.expect("stage B")
.durable()
.commit()
.expect("commit mixed record");
Db::tx_debug_skip_apply(false);
assert!(a.get(&a_key).is_none(), "A apply is suppressed");
assert!(b.get(&b_key).is_none(), "B apply is suppressed");
db.close().expect("close source db");
}
let logged = snapshot_txlog(dir.path());
let db = Db::open_test(dir.path()).expect("reopen target db");
let a = db
.open_zero_tree::<MixedA2, { std::mem::size_of::<MixedA2>() }, _>(
Config::test(),
NoHook,
&[],
)
.expect("migrate and replay A");
assert_eq!(a.get(&a_key), Some(MixedA2 { value: 13 }));
assert_eq!(db.db_info().collections[MIXED_A_NAME].version, 2);
assert_eq!(db.db_info().collections[MIXED_B_NAME].version, 1);
assert_eq!(
db.pending_transactions()
.into_iter()
.map(|record| record.collections)
.collect::<Vec<_>>(),
vec![vec![MIXED_A_NAME.to_owned(), MIXED_B_NAME.to_owned()]],
"B keeps the mixed record pending after only A opens"
);
assert_eq!(
snapshot_txlog(dir.path()),
logged,
"A replay rewrote the log"
);
let b = db
.open_zero_tree::<MixedB2, { std::mem::size_of::<MixedB2>() }, _>(
Config::test(),
NoHook,
&[],
)
.expect("migrate and replay B");
assert_eq!(b.get(&b_key), Some(MixedB2 { value: 14 }));
assert_eq!(db.db_info().collections[MIXED_B_NAME].version, 2);
assert_eq!(
db.pending_transactions().len(),
1,
"an applied record remains visible until checkpoint retirement"
);
assert_eq!(
snapshot_txlog(dir.path()),
logged,
"B replay rewrote the log"
);
drop(a);
drop(b);
db.close().expect("checkpoint the fully replayed record");
drop(db);
let reopened = Db::open_test(dir.path()).expect("reopen checkpointed db");
assert!(
reopened.pending_transactions().is_empty(),
"the record becomes checkpoint-eligible only after B opens"
);
}
#[test]
fn format_migration_replay_unknown_logged_fingerprint_fails_closed() {
assert_ne!(LoggedUnknownV1::TYPE.h(), ExpectedV1::TYPE.h());
let dir = tempdir().expect("tempdir");
let key = 15u64.to_be_bytes();
{
let db = Arc::new(Db::open_test(dir.path()).expect("open logged-schema db"));
let tree = db
.open_zero_tree::<LoggedUnknownV1, { std::mem::size_of::<LoggedUnknownV1>() }, _>(
Config::test(),
NoHook,
&[],
)
.expect("open logged-schema tree");
Db::tx_debug_skip_apply(true);
db.tx()
.put(&tree, &key, &LoggedUnknownV1 { value: 15 })
.expect("stage unknown-schema put")
.durable()
.commit()
.expect("commit unknown-schema record");
Db::tx_debug_skip_apply(false);
db.db_info_update(|info| {
let stored = info
.collections
.get_mut(UNKNOWN_NAME)
.expect("stored collection metadata");
stored.typ_hash = ExpectedV1::TYPE.h();
})
.expect("install the declared source fingerprint fixture");
db.close().expect("close logged-schema db");
}
let logged = snapshot_txlog(dir.path());
let db = Db::open_test(dir.path()).expect("reopen target db");
let err = match db.open_zero_tree::<ExpectedV2, { std::mem::size_of::<ExpectedV2>() }, _>(
Config::test(),
NoHook,
&[],
) {
Err(err) => err,
Ok(_) => panic!("an unknown logged fingerprint must fail closed"),
};
assert!(
matches!(
err,
DbError::TxSchemaMismatch {
collection: UNKNOWN_NAME,
stored,
logged,
..
} if stored == ExpectedV2::TYPE.h() && logged == LoggedUnknownV1::TYPE.h()
),
"{err}"
);
assert_eq!(
db.db_info().collections[UNKNOWN_NAME].version,
2,
"the nonempty format plan ran before replay rejected the record"
);
assert_eq!(db.pending_transactions().len(), 1);
assert_eq!(
snapshot_txlog(dir.path()),
logged,
"failed replay rewrote the log"
);
}
#[test]
fn format_migration_replay_unknown_delete_fingerprint_fails_closed() {
assert_ne!(LoggedUnknownV1::TYPE.h(), ExpectedV1::TYPE.h());
let dir = tempdir().expect("tempdir");
let key = 16u64.to_be_bytes();
{
let db = Arc::new(Db::open_test(dir.path()).expect("open logged-schema db"));
let tree = db
.open_zero_tree::<LoggedUnknownV1, { std::mem::size_of::<LoggedUnknownV1>() }, _>(
Config::test(),
NoHook,
&[],
)
.expect("open logged-schema tree");
tree.put(&key, &LoggedUnknownV1 { value: 16 })
.expect("seed row");
Db::tx_debug_skip_apply(true);
db.tx()
.delete(&tree, &key)
.expect("stage unknown-schema delete")
.durable()
.commit()
.expect("commit unknown-schema delete");
Db::tx_debug_skip_apply(false);
db.db_info_update(|info| {
info.collections
.get_mut(UNKNOWN_NAME)
.expect("stored collection metadata")
.typ_hash = ExpectedV1::TYPE.h();
})
.expect("install declared source fingerprint fixture");
db.close().expect("close logged-schema db");
}
let db = Db::open_test(dir.path()).expect("reopen target db");
let err = match db.open_zero_tree::<ExpectedV2, { std::mem::size_of::<ExpectedV2>() }, _>(
Config::test(),
NoHook,
&[],
) {
Err(err) => err,
Ok(_) => panic!("an unknown Delete fingerprint must fail closed"),
};
assert!(
matches!(
err,
DbError::TxSchemaMismatch {
collection: UNKNOWN_NAME,
stored,
logged,
..
} if stored == ExpectedV2::TYPE.h() && logged == LoggedUnknownV1::TYPE.h()
),
"{err}"
);
assert_eq!(db.pending_transactions().len(), 1);
}
#[test]
fn sequential_format_migrations_preserve_prior_shard_high_water() {
let dir = tempdir().expect("tempdir");
let key = 50u64.to_be_bytes();
let mut config = Config::test();
config.shard_count = 1;
{
let db = Arc::new(Db::open_test(dir.path()).expect("open V1 db"));
let tree = db
.open_zero_tree::<SequentialV1, { std::mem::size_of::<SequentialV1>() }, _>(
config.clone(),
NoHook,
&[],
)
.expect("open V1 tree");
for value in 1..=49 {
tree.put(&key, &SequentialV1 { first: value })
.expect("advance source GSN");
}
Db::tx_debug_skip_apply(true);
db.tx()
.put(&tree, &key, &SequentialV1 { first: 50 })
.expect("stage GSN 50 put")
.durable()
.commit()
.expect("persist GSN 50 put");
Db::tx_debug_skip_apply(false);
db.close().expect("close V1 db");
}
let tombstone = serialize_entry(100, &key, &[], true);
let mut active = std::fs::OpenOptions::new()
.append(true)
.open(only_data_file_named(dir.path(), SEQUENTIAL_NAME))
.expect("open active source file");
active
.write_all(&tombstone)
.expect("append GSN 100 tombstone");
active.sync_all().expect("sync GSN 100 tombstone");
let db = Db::open_test(dir.path()).expect("reopen migration db");
db.migrate_format::<SequentialV2>(&config)
.expect("run V1 to V2 explicit migration");
db.migrate_format::<SequentialV3>(&config)
.expect("run V2 to V3 explicit migration");
assert_eq!(
db.db_info().collections[SEQUENTIAL_NAME].shard_gsn_high,
vec![100],
"the second rewrite must retain the first rewrite's dropped tombstone floor"
);
let tree = db
.open_zero_tree::<SequentialV3, { std::mem::size_of::<SequentialV3>() }, _>(
config,
NoHook,
&[],
)
.expect("open V3 tree");
assert!(
tree.get(&key).is_none(),
"pending GSN 50 must not resurrect after the GSN 100 tombstone"
);
}
#[test]
fn format_migration_shard_replay_uses_persisted_floor() {
let dir = tempdir().expect("tempdir");
let key = 61u64.to_be_bytes();
let recovered_key = 63u64.to_be_bytes();
let mut config = Config::test();
config.shard_count = 1;
{
let db = Arc::new(Db::open_test(dir.path()).expect("open V1 db"));
let tree = db
.open_zero_tree::<V1, { std::mem::size_of::<V1>() }, _>(config.clone(), NoHook, &[])
.expect("open V1 tree");
Db::tx_debug_skip_apply(true);
db.tx()
.put(&tree, &key, &V1 { value: 61 })
.expect("stage put")
.durable()
.commit()
.expect("persist pending transaction");
Db::tx_debug_skip_apply(false);
db.db_info_update(|info| {
let collection = info.collections.get_mut(NAME).expect("collection metadata");
collection.gsn_high = 0;
collection.shard_gsn_high = vec![0];
})
.expect("persist recovered floor");
db.close().expect("close V1 db");
}
let raw = serialize_entry(
99,
&recovered_key,
&V1 { value: 63 }.value.to_ne_bytes(),
false,
);
let mut file = std::fs::OpenOptions::new()
.append(true)
.open(only_data_file(dir.path()))
.expect("open recovered shard data");
file.write_all(&raw).expect("append raw recovered entry");
file.sync_all().expect("sync raw recovered entry");
let db = Db::open_test(dir.path()).expect("reopen V2 db");
let tree = db
.open_zero_tree::<V2, { std::mem::size_of::<V2>() }, _>(config, NoHook, &[])
.expect("open migrated V2 tree");
assert!(
tree.get(&key).is_none(),
"raw recovered floor makes the lower pending GSN durable"
);
assert_eq!(tree.get(&recovered_key), Some(V2 { value: 63 }));
}
#[test]
fn format_migration_engine_gsn_uses_persisted_floor() {
let dir = tempdir().expect("tempdir");
let key = 62u64.to_be_bytes();
{
let db = Db::open_test(dir.path()).expect("open V1 db");
let _tree = db
.open_zero_tree::<V1, { std::mem::size_of::<V1>() }, _>(Config::test(), NoHook, &[])
.expect("open V1 tree");
db.db_info_update(|info| {
let collection = info.collections.get_mut(NAME).expect("collection metadata");
collection.gsn_high = 62;
collection.shard_gsn_high = vec![62];
})
.expect("persist floor");
db.close().expect("close V1 db");
}
{
let db = Arc::new(Db::open_test(dir.path()).expect("reopen V1 db"));
let tree = db
.open_zero_tree::<V1, { std::mem::size_of::<V1>() }, _>(Config::test(), NoHook, &[])
.expect("open V1 tree with floor");
Db::tx_debug_skip_apply(true);
db.tx()
.put(&tree, &key, &V1 { value: 62 })
.expect("stage put")
.durable()
.commit()
.expect("persist pending transaction");
Db::tx_debug_skip_apply(false);
db.close().expect("close V1 db");
}
let db = Db::open_test(dir.path()).expect("reopen V2 db");
let tree = db
.open_zero_tree::<V2, { std::mem::size_of::<V2>() }, _>(Config::test(), NoHook, &[])
.expect("open migrated V2 tree");
assert_eq!(tree.get(&key), Some(V2 { value: 62 }));
}
macro_rules! persisted_floor_family_case {
($name:ident, $open:expr) => {
#[test]
fn $name() {
let dir = tempdir().expect("tempdir");
let key = 71u64.to_be_bytes();
let open = $open;
{
let db = Arc::new(Db::open_test(dir.path()).expect("open db"));
let _tree = open(&db);
db.db_info_update(|info| {
let collection = info
.collections
.get_mut(FloorRow::NAME)
.expect("collection metadata");
collection.gsn_high = 70;
collection.shard_gsn_high = vec![70];
})
.expect("persist high-water floor");
db.close().expect("close initial db");
}
{
let db = Arc::new(Db::open_test(dir.path()).expect("reopen db"));
let tree = open(&db);
Db::tx_debug_skip_apply(true);
db.tx()
.put(&tree, &key, &FloorRow { value: 71 })
.expect("stage floor-after put")
.durable()
.commit()
.expect("persist pending transaction");
Db::tx_debug_skip_apply(false);
db.close().expect("close pending db");
}
let db = Arc::new(Db::open_test(dir.path()).expect("replay db"));
let tree = open(&db);
assert!(
tree.get(&key).is_some(),
"a GSN reserved after the persisted floor must replay"
);
}
};
}
persisted_floor_family_case!(format_migration_floor_typed_tree, |db: &Arc<Db>| db
.open_typed_tree::<FloorRow, armdb::RapiraCodec, _>(Config::test(), NoHook, &[])
.expect("open typed tree"));
persisted_floor_family_case!(format_migration_floor_typed_map, |db: &Arc<Db>| db
.open_typed_map::<FloorRow, armdb::RapiraCodec, _>(Config::test(), NoHook, &[])
.expect("open typed map"));
persisted_floor_family_case!(
format_migration_floor_zero_tree_raw_const_carrier,
|db: &Arc<Db>| db
.open_zero_tree::<FloorRow, { std::mem::size_of::<FloorRow>() }, _>(
Config::test(),
NoHook,
&[],
)
.expect("open zero tree")
);
persisted_floor_family_case!(
format_migration_floor_zero_map_raw_const_carrier,
|db: &Arc<Db>| db
.open_zero_map::<FloorRow, { std::mem::size_of::<FloorRow>() }, _>(
Config::test(),
NoHook,
&[],
)
.expect("open zero map")
);
persisted_floor_family_case!(
format_migration_floor_var_typed_tree_raw_var_carrier,
|db: &Arc<Db>| db
.open_var_typed_tree::<FloorRow, armdb::RapiraCodec, _>(Config::test(), NoHook, &[])
.expect("open var typed tree")
);
persisted_floor_family_case!(
format_migration_floor_var_typed_map_raw_var_carrier,
|db: &Arc<Db>| db
.open_var_typed_map::<FloorRow, armdb::RapiraCodec, _>(Config::test(), NoHook, &[])
.expect("open var typed map")
);