use std::fs;
use std::io::Write;
use std::path::Path;
use armdb::armour::{Db, TypedMigration};
use armdb::{
CollectionMeta, Config, CounterTargets, DbError, FormatAction, FormatMigration,
FormatMigrationPolicy, IndexTargets, MigrateAction, NoHook, OnMissingTarget, Projected,
RapiraCodec, TtlPolicy, Upsert, serialize_entry,
};
use armour_core::GetType;
use rapira::Rapira;
use tempfile::tempdir;
use zerocopy::{FromBytes, Immutable, IntoBytes};
const NAME: &str = "format_policy_rows";
#[derive(Clone, Copy, FromBytes, GetType, Immutable, IntoBytes)]
#[repr(C)]
struct V1 {
value: u32,
}
impl CollectionMeta for V1 {
type SelfId = [u8; 8];
const NAME: &'static str = NAME;
const VERSION: u16 = 1;
}
#[derive(Clone, Copy, Debug, Eq, FromBytes, GetType, Immutable, IntoBytes, PartialEq)]
#[repr(C)]
struct V2 {
value: u64,
}
const DROP_MANAGED_NAME: &str = "format_drop_managed_rows";
const DROP_MANAGED_INDEX: &str = "format_drop_managed_index";
const DROP_MANAGED_STATS: &str = "format_drop_managed_stats";
#[derive(Clone, Debug, PartialEq, Rapira, GetType)]
struct DropManagedV1 {
weight: u64,
}
impl CollectionMeta for DropManagedV1 {
type SelfId = [u8; 8];
const NAME: &'static str = DROP_MANAGED_NAME;
const VERSION: u16 = 1;
}
#[derive(Clone, Debug, PartialEq, Rapira, GetType)]
struct DropManagedV2 {
weight: u64,
}
fn drop_managed_row(_: &[u8], _: &[u8], _: &mut Vec<u8>) -> armdb::DbResult<FormatAction> {
Ok(FormatAction::Drop)
}
const DROP_MANAGED: FormatMigration = FormatMigration::raw(1, drop_managed_row);
impl CollectionMeta for DropManagedV2 {
type SelfId = [u8; 8];
const NAME: &'static str = DROP_MANAGED_NAME;
const VERSION: u16 = 2;
const FORMAT_MIGRATIONS: &'static [FormatMigration] = &[DROP_MANAGED];
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, FromBytes, GetType, Immutable, IntoBytes)]
#[repr(C)]
struct DropManagedIndex;
impl CollectionMeta for DropManagedIndex {
type SelfId = [u8; 8];
const NAME: &'static str = DROP_MANAGED_INDEX;
}
#[derive(Clone, Debug, PartialEq, Rapira, GetType)]
struct DropManagedStats {
weight: u64,
}
impl CollectionMeta for DropManagedStats {
type SelfId = [u8; 8];
const NAME: &'static str = DROP_MANAGED_STATS;
}
fn drop_managed_hook_v1(
index: &std::sync::Arc<armdb::zero_tree!(DropManagedIndex)>,
stats: &std::sync::Arc<armdb::typed_map!(DropManagedStats, RapiraCodec)>,
) -> Projected<[u8; 8], DropManagedV1> {
let index_for_projection = std::sync::Arc::clone(index);
let stats_for_counter = std::sync::Arc::clone(stats);
Projected::<[u8; 8], DropManagedV1>::new(
IndexTargets::new().with_index(index),
move |key, _row, sink| sink.row(&index_for_projection, *key, DropManagedIndex),
)
.and_counters(
CounterTargets::new().with_counter(
stats,
"weight",
OnMissingTarget::Create,
|row: &DropManagedStats| row.weight,
|_old, weight| Upsert::Put(DropManagedStats { weight }),
),
move |_key, row, sink| sink.add(&stats_for_counter, [0; 8], row.weight),
)
}
fn drop_managed_hook_v2(
index: &std::sync::Arc<armdb::zero_tree!(DropManagedIndex)>,
stats: &std::sync::Arc<armdb::typed_map!(DropManagedStats, RapiraCodec)>,
) -> Projected<[u8; 8], DropManagedV2> {
let index_for_projection = std::sync::Arc::clone(index);
let stats_for_counter = std::sync::Arc::clone(stats);
Projected::<[u8; 8], DropManagedV2>::new(
IndexTargets::new().with_index(index),
move |key, _row, sink| sink.row(&index_for_projection, *key, DropManagedIndex),
)
.and_counters(
CounterTargets::new().with_counter(
stats,
"weight",
OnMissingTarget::Create,
|row: &DropManagedStats| row.weight,
|_old, weight| Upsert::Put(DropManagedStats { weight }),
),
move |_key, row, sink| sink.add(&stats_for_counter, [0; 8], row.weight),
)
}
const TTL_FORMAT_NAME: &str = "format_ttl_final_value";
#[derive(Clone, Debug, PartialEq, Rapira, GetType)]
struct TtlFormatV1 {
deadline: u64,
}
impl CollectionMeta for TtlFormatV1 {
type SelfId = [u8; 8];
const NAME: &'static str = TTL_FORMAT_NAME;
const VERSION: u16 = 1;
}
#[derive(Clone, Debug, PartialEq, Rapira, GetType)]
struct TtlFormatV2 {
expires_at_ms: u64,
}
const TTL_FORMAT: FormatMigration = armdb::format_step!(
1,
TtlFormatV1 => TtlFormatV2,
RapiraCodec,
|old| Ok::<_, DbError>(Some(TtlFormatV2 {
expires_at_ms: old.deadline + 100,
}))
);
impl CollectionMeta for TtlFormatV2 {
type SelfId = [u8; 8];
const NAME: &'static str = TTL_FORMAT_NAME;
const VERSION: u16 = 2;
const FORMAT_MIGRATIONS: &'static [FormatMigration] = &[TTL_FORMAT];
const TTL: Option<TtlPolicy<Self>> = Some(TtlPolicy {
revision: 1,
deadline: |row, _| Some(row.expires_at_ms),
});
}
fn widen(_: &[u8], old: &[u8], out: &mut Vec<u8>) -> armdb::DbResult<FormatAction> {
out.extend_from_slice(
&(u32::from_ne_bytes(old.try_into().expect("width")) as u64).to_ne_bytes(),
);
Ok(FormatAction::Write)
}
const WIDEN: FormatMigration = FormatMigration::raw(1, 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, Eq, FromBytes, GetType, Immutable, IntoBytes, PartialEq)]
#[repr(C)]
struct FailingV2 {
value: u64,
}
fn fail_format_step(_key: &[u8], _old: &[u8], _out: &mut Vec<u8>) -> armdb::DbResult<FormatAction> {
Err(DbError::FormatMismatch("injected format failure".into()))
}
const FAIL_FORMAT: FormatMigration = FormatMigration::raw(1, fail_format_step);
impl CollectionMeta for FailingV2 {
type SelfId = [u8; 8];
const NAME: &'static str = NAME;
const VERSION: u16 = 2;
const FORMAT_MIGRATIONS: &'static [FormatMigration] = &[FAIL_FORMAT];
}
fn seed(root: &std::path::Path) {
let db = Db::open_test(root).unwrap();
let tree = db
.open_zero_tree::<V1, { std::mem::size_of::<V1>() }, _>(Config::test(), NoHook, &[])
.unwrap();
tree.put(&1u64.to_be_bytes(), &V1 { value: 1 }).unwrap();
db.close().unwrap();
}
fn seed_empty(root: &std::path::Path) {
let db = Db::open_test(root).unwrap();
let _tree = db
.open_zero_tree::<V1, { std::mem::size_of::<V1>() }, _>(Config::test(), NoHook, &[])
.unwrap();
db.close().unwrap();
}
fn snapshot_tree(root: &Path) -> Vec<(String, bool, u64, Vec<u8>)> {
fn visit(base: &Path, path: &Path, entries: &mut Vec<(String, bool, u64, Vec<u8>)>) {
let mut children = fs::read_dir(path)
.expect("snapshot directory")
.collect::<Result<Vec<_>, _>>()
.expect("snapshot entries");
children.sort_by_key(fs::DirEntry::file_name);
for child in children {
let child_path = child.path();
let relative = child_path
.strip_prefix(base)
.expect("snapshot relative path")
.display()
.to_string();
let file_type = child.file_type().expect("snapshot file type");
if file_type.is_dir() {
entries.push((relative, true, 0, Vec::new()));
visit(base, &child_path, entries);
} else {
let bytes = fs::read(&child_path).expect("snapshot file bytes");
entries.push((relative, false, bytes.len() as u64, bytes));
}
}
}
let mut entries = Vec::new();
visit(root, root, &mut entries);
entries
}
fn snapshot_collection(root: &Path, name: &str) -> Vec<(String, bool, u64, Vec<u8>)> {
snapshot_tree(&root.join(name))
}
fn active_data_file(root: &Path, name: &str) -> std::path::PathBuf {
let mut files = Vec::new();
for shard in fs::read_dir(root.join(name)).expect("collection directory") {
let shard = shard.expect("shard entry");
if !shard.file_type().expect("shard file type").is_dir()
|| !shard.file_name().to_string_lossy().starts_with("shard_")
{
continue;
}
for entry in fs::read_dir(shard.path()).expect("shard directory") {
let path = entry.expect("data entry").path();
if path.extension().is_some_and(|ext| ext == "data") {
files.push(path);
}
}
}
files.sort();
files.pop().expect("active data file")
}
fn data_files(root: &Path, name: &str) -> Vec<std::path::PathBuf> {
let mut files = Vec::new();
for shard in fs::read_dir(root.join(name)).expect("collection directory") {
let shard = shard.expect("shard entry");
if !shard.file_type().expect("shard file type").is_dir()
|| !shard.file_name().to_string_lossy().starts_with("shard_")
{
continue;
}
for entry in fs::read_dir(shard.path()).expect("shard directory") {
let path = entry.expect("data entry").path();
if path.extension().is_some_and(|ext| ext == "data") {
files.push(path);
}
}
}
files.sort();
files
}
#[test]
fn format_migration_refuse_bitcask_writes_nothing() {
let dir = tempdir().unwrap();
seed(dir.path());
let before = snapshot_collection(dir.path(), NAME);
let db = Db::open_test(dir.path()).unwrap();
let mut config = Config::test();
config.format_migration = FormatMigrationPolicy::Refuse;
let err = match db.open_zero_tree::<V2, { std::mem::size_of::<V2>() }, _>(config, NoHook, &[]) {
Ok(_) => panic!("refuse opened"),
Err(err) => err,
};
assert!(matches!(
err,
DbError::FormatMigrationRequired {
collection: NAME,
from: 1,
to: 2
}
));
assert_eq!(snapshot_collection(dir.path(), NAME), before);
}
#[test]
fn format_migration_explicit_then_open_is_clean() {
let dir = tempdir().unwrap();
seed(dir.path());
let db = Db::open_test(dir.path()).unwrap();
let report = db.migrate_format::<V2>(&Config::test()).unwrap();
assert_eq!(
(
report.collection,
report.from,
report.to,
report.rows_migrated
),
(NAME, 1, 2, 1)
);
let mut config = Config::test();
config.format_migration = FormatMigrationPolicy::Refuse;
let tree = db
.open_zero_tree::<V2, { std::mem::size_of::<V2>() }, _>(config, NoHook, &[])
.unwrap();
assert_eq!(tree.get(&1u64.to_be_bytes()), Some(V2 { value: 1 }));
}
#[test]
fn format_migration_plain_direct_io_output_is_page_aligned_and_append_safe() {
let dir = tempdir().expect("tempdir");
seed(dir.path());
let config = Config::small_values().shard_count(2).hints(true).build();
let db = Db::open_test(dir.path()).expect("reopen direct-io db");
let tree = db
.open_zero_tree::<V2, { std::mem::size_of::<V2>() }, _>(config.clone(), NoHook, &[])
.expect("migrate with direct-io layout");
for path in data_files(dir.path(), NAME) {
let len = fs::metadata(&path).expect("target data metadata").len();
assert_eq!(
len % 4096,
0,
"migration output is not page-aligned: {} has {len} bytes",
path.display()
);
}
tree.put(&2u64.to_be_bytes(), &V2 { value: 2 })
.expect("append after direct-io migration");
tree.flush().expect("flush appended row");
drop(tree);
db.close().expect("close direct-io db");
drop(db);
let db = Db::open_test(dir.path()).expect("reopen appended db");
let tree = db
.open_zero_tree::<V2, { std::mem::size_of::<V2>() }, _>(config, NoHook, &[])
.expect("reopen migrated direct-io tree");
assert_eq!(tree.get(&1u64.to_be_bytes()), Some(V2 { value: 1 }));
assert_eq!(tree.get(&2u64.to_be_bytes()), Some(V2 { value: 2 }));
}
#[test]
fn format_migration_explicit_rejects_live_collection() {
let dir = tempdir().unwrap();
let db = Db::open_test(dir.path()).unwrap();
let _tree = db
.open_zero_tree::<V1, { std::mem::size_of::<V1>() }, _>(Config::test(), NoHook, &[])
.unwrap();
assert!(matches!(
db.migrate_format::<V2>(&Config::test()),
Err(DbError::DuplicateCollection(NAME))
));
}
#[test]
fn format_migration_invalid_config_writes_nothing() {
let dir = tempdir().unwrap();
seed(dir.path());
let before = snapshot_collection(dir.path(), NAME);
let db = Db::open_test(dir.path()).unwrap();
let mut invalid = Config::test();
invalid.shard_count = 0;
assert!(matches!(
db.migrate_format::<V2>(&invalid),
Err(DbError::Config(_))
));
assert_eq!(snapshot_collection(dir.path(), NAME), before);
}
#[test]
fn format_migration_auto_rejects_a_missing_configured_bitcask_shard_without_writes() {
let dir = tempdir().unwrap();
seed_empty(dir.path());
fs::remove_dir_all(dir.path().join(NAME).join("shard_001")).unwrap();
let before = snapshot_collection(dir.path(), NAME);
let db = Db::open_test(dir.path()).unwrap();
let error = match db.open_zero_tree::<V2, { std::mem::size_of::<V2>() }, _>(
Config::test(),
NoHook,
&[],
) {
Err(error) => error,
Ok(_) => panic!("a missing configured shard must fail closed"),
};
assert!(matches!(error, DbError::FormatMismatch(_)), "{error}");
assert_eq!(snapshot_collection(dir.path(), NAME), before);
assert!(!db.db_info().collections[NAME].projections_dirty);
}
#[test]
fn format_migration_explicit_rejects_a_missing_configured_bitcask_shard_without_writes() {
let dir = tempdir().unwrap();
seed_empty(dir.path());
fs::remove_dir_all(dir.path().join(NAME).join("shard_001")).unwrap();
let before = snapshot_collection(dir.path(), NAME);
let db = Db::open_test(dir.path()).unwrap();
let error = db
.migrate_format::<V2>(&Config::test())
.expect_err("a missing configured shard must fail closed");
assert!(matches!(error, DbError::FormatMismatch(_)), "{error}");
assert_eq!(snapshot_collection(dir.path(), NAME), before);
assert!(!db.db_info().collections[NAME].projections_dirty);
}
#[test]
fn format_migration_explicit_failure_leaves_projections_dirty_durable() {
let dir = tempdir().expect("tempdir");
seed(dir.path());
let db = Db::open_test(dir.path()).expect("reopen db");
let err = db
.migrate_format::<FailingV2>(&Config::test())
.expect_err("format callback must fail");
assert!(matches!(err, DbError::FormatStepFailed { .. }), "{err}");
assert!(
db.db_info().collections[NAME].projections_dirty,
"explicit migration must durably dirty projections before backend work"
);
drop(db);
let reopened = Db::open_test(dir.path()).expect("reopen after failure");
assert!(
reopened.db_info().collections[NAME].projections_dirty,
"dirty evidence must survive a failed explicit migration"
);
}
#[test]
fn format_migration_accepts_recovery_valid_torn_active_tail() {
let dir = tempdir().expect("tempdir");
seed(dir.path());
let mut torn = serialize_entry(77, &2u64.to_be_bytes(), &7u32.to_ne_bytes(), false);
torn.truncate(torn.len() - 3);
let mut active = fs::OpenOptions::new()
.append(true)
.open(active_data_file(dir.path(), NAME))
.expect("open active data file");
active.write_all(&torn).expect("append torn entry");
active.sync_all().expect("sync torn tail");
let db = Db::open_test(dir.path()).expect("reopen db");
let tree = db
.open_zero_tree::<V2, { std::mem::size_of::<V2>() }, _>(Config::test(), NoHook, &[])
.expect("migration must match recovery's torn-tail contract");
assert_eq!(tree.get(&1u64.to_be_bytes()), Some(V2 { value: 1 }));
assert!(tree.get(&2u64.to_be_bytes()).is_none());
}
#[test]
fn format_migration_discards_torn_mig_marker_and_rewrites_the_recoverable_source() {
let dir = tempdir().expect("tempdir");
seed(dir.path());
let migration = dir.path().join(NAME).join("shard_000.mig");
fs::create_dir(&migration).expect("create interrupted migration output");
fs::write(migration.join("format.v"), b"AFV1\x02").expect("write legacy torn direct marker");
let db = Db::open_test(dir.path()).expect("reopen db");
let tree = db
.open_zero_tree::<V2, { std::mem::size_of::<V2>() }, _>(Config::test(), NoHook, &[])
.expect("recoverable source must replace torn migration output");
assert_eq!(tree.get(&1u64.to_be_bytes()), Some(V2 { value: 1 }));
assert!(!migration.exists());
}
#[test]
fn format_migration_sweeps_validated_stale_temps_without_rewrite() {
let dir = tempdir().expect("tempdir");
seed(dir.path());
let db = Db::open_test(dir.path()).expect("reopen migration db");
let tree = db
.open_zero_tree::<V2, { std::mem::size_of::<V2>() }, _>(Config::test(), NoHook, &[])
.expect("migrate to target");
drop(tree);
db.close().expect("close migrated db");
drop(db);
let collection = dir.path().join(NAME);
fs::write(collection.join(".migration.tmp"), b"stale").expect("stale collection temp");
let staging = collection.join(".format-stage-000");
fs::create_dir(&staging).expect("stale staging directory");
fs::write(staging.join("partial"), b"stale").expect("stale staging payload");
let shard = collection.join("shard_000");
for name in ["format.v.tmp", "format.gsn.tmp", "format.file-id.tmp"] {
fs::write(shard.join(name), b"stale").expect("stale shard marker temp");
}
let db = Db::open_test(dir.path()).expect("reopen target db");
let _tree = db
.open_zero_tree::<V2, { std::mem::size_of::<V2>() }, _>(Config::test(), NoHook, &[])
.expect("target open sweeps admitted stale temps");
assert!(!collection.join(".migration.tmp").exists());
assert!(!staging.exists());
for name in ["format.v.tmp", "format.gsn.tmp", "format.file-id.tmp"] {
assert!(!shard.join(name).exists(), "stale temp survived: {name}");
}
}
#[test]
fn format_migration_rejects_unexpected_staging_identity_without_deleting_it() {
let dir = tempdir().expect("tempdir");
seed(dir.path());
let db = Db::open_test(dir.path()).expect("reopen migration db");
let tree = db
.open_zero_tree::<V2, { std::mem::size_of::<V2>() }, _>(Config::test(), NoHook, &[])
.expect("migrate to target");
drop(tree);
db.close().expect("close migrated db");
drop(db);
let staging = dir.path().join(NAME).join(".format-stage-000");
fs::write(&staging, b"do not delete").expect("unexpected staging file");
let db = Db::open_test(dir.path()).expect("reopen target db");
let error = match db.open_zero_tree::<V2, { std::mem::size_of::<V2>() }, _>(
Config::test(),
NoHook,
&[],
) {
Ok(_) => panic!("unexpected staging identity must fail closed"),
Err(error) => error,
};
assert!(matches!(error, DbError::FormatMismatch(_)), "{error}");
assert_eq!(
fs::read(staging).expect("identity preserved"),
b"do not delete"
);
}
const VAR_TREE_NAME: &str = "format_var_tree_rows";
#[derive(Clone, Debug, PartialEq, Rapira, GetType)]
struct VarTreeV1 {
payload: Vec<u8>,
}
impl CollectionMeta for VarTreeV1 {
type SelfId = [u8; 8];
const NAME: &'static str = VAR_TREE_NAME;
const VERSION: u16 = 1;
}
#[derive(Clone, Debug, PartialEq, Rapira, GetType)]
struct VarTreeV2 {
payload: Vec<u8>,
generation: u16,
}
const VAR_TREE_FORMAT: FormatMigration = armdb::format_step!(
1,
VarTreeV1 => VarTreeV2,
RapiraCodec,
|old| Ok::<_, DbError>(Some(VarTreeV2 {
payload: old.payload,
generation: 2,
}))
);
impl CollectionMeta for VarTreeV2 {
type SelfId = [u8; 8];
const NAME: &'static str = VAR_TREE_NAME;
const VERSION: u16 = 2;
const FORMAT_MIGRATIONS: &'static [FormatMigration] = &[VAR_TREE_FORMAT];
}
#[test]
fn format_migration_var_typed_tree_large_value() {
let dir = tempdir().expect("tempdir");
let key = 11u64.to_be_bytes();
let payload = vec![0x5a; 32 * 1024 + 17];
let db = Db::open_test(dir.path()).expect("open v1 db");
let tree = db
.open_var_typed_tree::<VarTreeV1, RapiraCodec, _>(Config::test(), NoHook, &[])
.expect("open v1 var tree");
tree.put(
&key,
&VarTreeV1 {
payload: payload.clone(),
},
)
.expect("seed large var-tree row");
drop(tree);
db.close().expect("close v1 db");
drop(db);
let db = Db::open_test(dir.path()).expect("reopen v2 db");
let tree = db
.open_var_typed_tree::<VarTreeV2, RapiraCodec, _>(Config::test(), NoHook, &[])
.expect("migrate var tree");
assert_eq!(
tree.get(&key),
Some(VarTreeV2 {
payload,
generation: 2,
})
);
}
#[test]
fn format_migration_skips_8_byte_padding_gap() {
let dir = tempdir().expect("tempdir");
let mut config = Config::test();
config.shard_count = 1;
config.sync_batch_size = 1; config.direct_io = true;
let first = vec![0x5a; 4088 - 16 - 8 - 4];
let rows = [
(1u64, first),
(2u64, vec![0x6b; 100]),
(3u64, vec![0x7c; 200]),
];
let db = Db::open_test(dir.path()).expect("open v1 db");
let tree = db
.open_var_typed_tree::<VarTreeV1, RapiraCodec, _>(config.clone(), NoHook, &[])
.expect("open v1 var tree");
for (key, payload) in &rows {
tree.put(
&key.to_be_bytes(),
&VarTreeV1 {
payload: payload.clone(),
},
)
.expect("seed row");
}
drop(tree);
db.close().expect("close v1 db");
drop(db);
let files = data_files(dir.path(), VAR_TREE_NAME);
let bytes = fs::read(&files[0]).expect("read data file");
assert_eq!(bytes[4088..4096], [0u8; 8], "8-byte padding gap");
assert_eq!(u64::from_ne_bytes(bytes[4096..4104].try_into().unwrap()), 2);
let db = Db::open_test(dir.path()).expect("reopen v2 db");
let tree = db
.open_var_typed_tree::<VarTreeV2, RapiraCodec, _>(config, NoHook, &[])
.expect("migrate across 8-byte gap");
for (key, payload) in rows {
assert_eq!(
tree.get(&key.to_be_bytes()),
Some(VarTreeV2 {
payload,
generation: 2,
})
);
}
}
const VAR_MAP_NAME: &str = "format_var_map_rows";
#[derive(Clone, Debug, PartialEq, Rapira, GetType)]
struct VarMapV1 {
payload: Vec<u8>,
}
impl CollectionMeta for VarMapV1 {
type SelfId = [u8; 8];
const NAME: &'static str = VAR_MAP_NAME;
const VERSION: u16 = 1;
}
#[derive(Clone, Debug, PartialEq, Rapira, GetType)]
struct VarMapV2 {
payload: Vec<u8>,
generation: u16,
}
const VAR_MAP_FORMAT: FormatMigration = armdb::format_step!(
1,
VarMapV1 => VarMapV2,
RapiraCodec,
|old| Ok::<_, DbError>(Some(VarMapV2 {
payload: old.payload,
generation: 2,
}))
);
impl CollectionMeta for VarMapV2 {
type SelfId = [u8; 8];
const NAME: &'static str = VAR_MAP_NAME;
const VERSION: u16 = 2;
const FORMAT_MIGRATIONS: &'static [FormatMigration] = &[VAR_MAP_FORMAT];
}
#[test]
fn format_migration_var_typed_map_large_value() {
let dir = tempdir().expect("tempdir");
let key = 12u64.to_be_bytes();
let payload = vec![0xa5; 32 * 1024 + 31];
let db = Db::open_test(dir.path()).expect("open v1 db");
let map = db
.open_var_typed_map::<VarMapV1, RapiraCodec, _>(Config::test(), NoHook, &[])
.expect("open v1 var map");
map.put(
&key,
&VarMapV1 {
payload: payload.clone(),
},
)
.expect("seed large var-map row");
drop(map);
db.close().expect("close v1 db");
drop(db);
let db = Db::open_test(dir.path()).expect("reopen v2 db");
let map = db
.open_var_typed_map::<VarMapV2, RapiraCodec, _>(Config::test(), NoHook, &[])
.expect("migrate var map");
assert_eq!(
map.get(&key),
Some(VarMapV2 {
payload,
generation: 2,
})
);
}
const MIXED_NAME: &str = "format_mixed_rows";
#[derive(Clone, Debug, PartialEq, Rapira, GetType)]
struct MixedV1 {
text: String,
}
impl CollectionMeta for MixedV1 {
type SelfId = [u8; 8];
const NAME: &'static str = MIXED_NAME;
const VERSION: u16 = 1;
}
#[derive(Clone, Debug, PartialEq, Rapira, GetType)]
struct MixedV3 {
text: String,
format_generation: u16,
value_generation: u16,
}
const MIXED_FORMAT: FormatMigration = armdb::format_step!(
1,
MixedV1 => MixedV3,
RapiraCodec,
|old| Ok::<_, DbError>(Some(MixedV3 {
text: format!("{}:format", old.text),
format_generation: 2,
value_generation: 0,
}))
);
impl CollectionMeta for MixedV3 {
type SelfId = [u8; 8];
const NAME: &'static str = MIXED_NAME;
const VERSION: u16 = 3;
const FORMAT_MIGRATIONS: &'static [FormatMigration] = &[MIXED_FORMAT];
}
fn mixed_value_suffix(_key: &[u8; 8], old: &MixedV3) -> MigrateAction<MixedV3> {
MigrateAction::Update(MixedV3 {
text: format!("{}:value", old.text),
format_generation: old.format_generation,
value_generation: 3,
})
}
#[test]
fn format_migration_mixed_format_then_value() {
let dir = tempdir().expect("tempdir");
let key = 13u64.to_be_bytes();
let db = Db::open_test(dir.path()).expect("open v1 db");
let tree = db
.open_typed_tree::<MixedV1, RapiraCodec, _>(Config::test(), NoHook, &[])
.expect("open mixed v1");
tree.put(
&key,
MixedV1 {
text: "seed".to_owned(),
},
)
.expect("seed mixed row");
drop(tree);
db.close().expect("close v1 db");
drop(db);
let migrations: &[TypedMigration<[u8; 8], MixedV3>] = &[(2, mixed_value_suffix)];
let db = Db::open_test(dir.path()).expect("reopen v3 db");
let tree = db
.open_typed_tree::<MixedV3, RapiraCodec, _>(Config::test(), NoHook, migrations)
.expect("run mixed migration");
assert_eq!(
tree.get(&key).map(|value| value.clone_value()),
Some(MixedV3 {
text: "seed:format:value".to_owned(),
format_generation: 2,
value_generation: 3,
})
);
}
#[test]
fn format_migration_explicit_completed_prefix_reports_installed_version() {
let dir = tempdir().expect("tempdir");
let db = Db::open_test(dir.path()).expect("open v1 db");
let tree = db
.open_typed_tree::<MixedV1, RapiraCodec, _>(Config::test(), NoHook, &[])
.expect("open mixed v1");
drop(tree);
db.close().expect("close v1 db");
drop(db);
let db = Db::open_test(dir.path()).expect("reopen explicit db");
let first = db
.migrate_format::<MixedV3>(&Config::test())
.expect("install format prefix");
assert_eq!((first.from, first.to), (1, 2));
let second = db
.migrate_format::<MixedV3>(&Config::test())
.expect("completed prefix is an empty explicit run");
assert_eq!((second.from, second.to), (2, 2));
assert_eq!(second.shards, 0);
}
const RETAINED_INVALID_NAME: &str = "format_retained_invalid_chain";
#[derive(Clone, Debug, PartialEq, Rapira, GetType)]
struct RetainedTargetV2 {
value: u64,
}
impl CollectionMeta for RetainedTargetV2 {
type SelfId = [u8; 8];
const NAME: &'static str = RETAINED_INVALID_NAME;
const VERSION: u16 = 2;
}
#[derive(Clone, Debug, PartialEq, Rapira)]
struct RetainedInvalidV2 {
value: u64,
}
impl GetType for RetainedInvalidV2 {
const TYPE: armour_core::Typ = RetainedTargetV2::TYPE;
}
fn retained_invalid_step(
_key: &[u8],
old: &[u8],
out: &mut Vec<u8>,
) -> armdb::DbResult<FormatAction> {
out.extend_from_slice(old);
Ok(FormatAction::Write)
}
const RETAINED_INVALID_STEP: FormatMigration =
FormatMigration::raw_checked(1, &u32::TYPE, &u16::TYPE, retained_invalid_step);
impl CollectionMeta for RetainedInvalidV2 {
type SelfId = [u8; 8];
const NAME: &'static str = RETAINED_INVALID_NAME;
const VERSION: u16 = 2;
const FORMAT_MIGRATIONS: &'static [FormatMigration] = &[RETAINED_INVALID_STEP];
}
#[test]
fn format_migration_retained_invalid_chain_fails_without_pending_tx() {
assert_eq!(RetainedTargetV2::TYPE.h(), RetainedInvalidV2::TYPE.h());
let dir = tempdir().expect("tempdir");
let db = Db::open_test(dir.path()).expect("open target db");
let tree = db
.open_typed_tree::<RetainedTargetV2, RapiraCodec, _>(Config::test(), NoHook, &[])
.expect("open target collection");
drop(tree);
db.close().expect("close target db");
drop(db);
let db = Db::open_test(dir.path()).expect("reopen target db");
let error = match db.open_typed_tree::<RetainedInvalidV2, RapiraCodec, _>(
Config::test(),
NoHook,
&[],
) {
Ok(_) => panic!("invalid retained chain must fail on a clean target-version open"),
Err(error) => error,
};
assert!(
matches!(error, DbError::FormatTypMismatch { .. }),
"{error}"
);
}
const NONCONTIGUOUS_NAME: &str = "format_noncontiguous_explicit";
#[derive(Clone, Debug, PartialEq, Rapira, GetType)]
struct NonContiguousV1 {
value: u64,
}
impl CollectionMeta for NonContiguousV1 {
type SelfId = [u8; 8];
const NAME: &'static str = NONCONTIGUOUS_NAME;
const VERSION: u16 = 1;
}
#[derive(Clone, Debug, PartialEq, Rapira, GetType)]
struct NonContiguousV4 {
value: u64,
}
const NONCONTIGUOUS_FORMAT_1: FormatMigration = armdb::format_step!(
1,
NonContiguousV1 => NonContiguousV4,
RapiraCodec,
|old| Ok::<_, DbError>(Some(NonContiguousV4 { value: old.value }))
);
const NONCONTIGUOUS_FORMAT_3: FormatMigration = armdb::format_step!(
3,
NonContiguousV4 => NonContiguousV4,
RapiraCodec,
|old| Ok::<_, DbError>(Some(old))
);
impl CollectionMeta for NonContiguousV4 {
type SelfId = [u8; 8];
const NAME: &'static str = NONCONTIGUOUS_NAME;
const VERSION: u16 = 4;
const FORMAT_MIGRATIONS: &'static [FormatMigration] =
&[NONCONTIGUOUS_FORMAT_1, NONCONTIGUOUS_FORMAT_3];
}
#[test]
fn explicit_format_migration_rejects_format_after_value_gap_without_writes() {
let dir = tempdir().expect("tempdir");
{
let db = Db::open_test(dir.path()).expect("open v1 db");
let tree = db
.open_typed_tree::<NonContiguousV1, RapiraCodec, _>(Config::test(), NoHook, &[])
.expect("open v1 collection");
tree.put(&31u64.to_be_bytes(), NonContiguousV1 { value: 31 })
.expect("seed v1 row");
db.close().expect("close v1 db");
}
let before = snapshot_collection(dir.path(), NONCONTIGUOUS_NAME);
let db = Db::open_test(dir.path()).expect("reopen for explicit migration");
let error = db
.migrate_format::<NonContiguousV4>(&Config::test())
.expect_err("format step 3 after the value step at version 2 must be rejected");
assert!(matches!(
error,
DbError::FormatStepAfterValueStep {
collection: NONCONTIGUOUS_NAME,
version: 3,
}
));
assert_eq!(
snapshot_collection(dir.path(), NONCONTIGUOUS_NAME),
before,
"the full declaration must be validated before dirty markers or shard output"
);
assert!(!db.db_info().collections[NONCONTIGUOUS_NAME].projections_dirty);
}
const WRONG_TARGET_NAME: &str = "format_wrong_target_rows";
#[derive(Clone, Debug, PartialEq, Rapira, GetType)]
struct WrongTargetV1 {
value: u32,
}
impl CollectionMeta for WrongTargetV1 {
type SelfId = [u8; 8];
const NAME: &'static str = WRONG_TARGET_NAME;
const VERSION: u16 = 1;
}
#[derive(Clone, Debug, PartialEq, Rapira, GetType)]
struct WrongIntermediate {
value: u64,
}
#[derive(Clone, Debug, PartialEq, Rapira, GetType)]
struct CheckedFinalTarget {
value: String,
}
const WRONG_FINAL_FORMAT: FormatMigration = armdb::format_step!(
1,
WrongTargetV1 => WrongIntermediate,
RapiraCodec,
|old| Ok::<_, DbError>(Some(WrongIntermediate {
value: u64::from(old.value),
}))
);
impl CollectionMeta for CheckedFinalTarget {
type SelfId = [u8; 8];
const NAME: &'static str = WRONG_TARGET_NAME;
const VERSION: u16 = 2;
const FORMAT_MIGRATIONS: &'static [FormatMigration] = &[WRONG_FINAL_FORMAT];
}
#[test]
fn format_migration_checked_final_target_mismatch_writes_nothing() {
let dir = tempdir().expect("tempdir");
let db = Db::open_test(dir.path()).expect("open v1 db");
let tree = db
.open_typed_tree::<WrongTargetV1, RapiraCodec, _>(Config::test(), NoHook, &[])
.expect("open wrong-target v1");
tree.put(&14u64.to_be_bytes(), WrongTargetV1 { value: 14 })
.expect("seed wrong-target row");
drop(tree);
db.close().expect("close v1 db");
drop(db);
let before = snapshot_collection(dir.path(), WRONG_TARGET_NAME);
let db = Db::open_test(dir.path()).expect("reopen target db");
let err = db
.open_typed_tree::<CheckedFinalTarget, RapiraCodec, _>(Config::test(), NoHook, &[])
.err()
.expect("checked final target mismatch must fail");
assert!(matches!(
err,
DbError::FormatTypMismatch {
collection: WRONG_TARGET_NAME,
from: 2,
..
}
));
assert_eq!(
snapshot_collection(dir.path(), WRONG_TARGET_NAME),
before,
"plan validation must precede every collection write"
);
}
#[cfg(feature = "fjall")]
#[test]
fn format_migration_lsm_is_unsupported() {
let dir = tempdir().unwrap();
let db = Db::open_test(dir.path()).unwrap();
let tree = db
.open_lsm_tree::<V1, armdb::ZerocopyCodec, _>(
None,
armdb::lsm::LsmConfig::default(),
NoHook,
&[],
)
.unwrap();
tree.put(&15u64.to_be_bytes(), &V1 { value: 15 }).unwrap();
drop(tree);
db.close().unwrap();
drop(db);
let db = Db::open_test(dir.path()).unwrap();
let lsm_root = dir.path().join("lsm");
let before = snapshot_tree(&lsm_root);
assert!(matches!(
db.open_lsm_tree::<V2, armdb::ZerocopyCodec, _>(
None,
armdb::lsm::LsmConfig::default(),
NoHook,
&[],
),
Err(DbError::FormatMigrationUnsupported {
collection: NAME,
backend: "lsm"
})
));
assert_eq!(
snapshot_tree(&lsm_root),
before,
"LSM refusal must happen before opening or writing the fjall database"
);
}
#[test]
fn format_migration_drop_reconciles_projection_and_count() {
let dir = tempdir().expect("tempdir");
let key = 17u64.to_be_bytes();
{
let db = Db::open_test(dir.path()).expect("open v1 db");
let index = db
.open_zero_tree::<DropManagedIndex, { std::mem::size_of::<DropManagedIndex>() }, _>(
Config::test(),
NoHook,
&[],
)
.expect("open index");
let stats = db
.open_typed_map::<DropManagedStats, RapiraCodec, _>(Config::test(), NoHook, &[])
.expect("open stats");
let source = db
.open_typed_map::<DropManagedV1, RapiraCodec, _>(
Config::test(),
drop_managed_hook_v1(&index, &stats),
&[],
)
.expect("open source v1");
source
.put(&key, DropManagedV1 { weight: 7 })
.expect("seed source");
assert!(index.contains(&key), "v1 projection exists");
assert_eq!(stats.get(&[0; 8]).map(|stats| stats.weight), Some(7));
db.close().expect("close v1 db");
}
let db = Db::open_test(dir.path()).expect("open v2 db");
let index = db
.open_zero_tree::<DropManagedIndex, { std::mem::size_of::<DropManagedIndex>() }, _>(
Config::test(),
NoHook,
&[],
)
.expect("reopen index");
let stats = db
.open_typed_map::<DropManagedStats, RapiraCodec, _>(Config::test(), NoHook, &[])
.expect("reopen stats");
let source = db
.open_typed_map::<DropManagedV2, RapiraCodec, _>(
Config::test(),
drop_managed_hook_v2(&index, &stats),
&[],
)
.expect("open source v2");
assert!(source.get(&key).is_none(), "format Drop removed source row");
assert!(
!index.contains(&key),
"reconciliation removed stale projection"
);
assert_eq!(
stats.get(&[0; 8]).map(|stats| stats.weight),
Some(0),
"recount rebuilt the aggregate from the dropped source"
);
}
#[test]
fn format_migration_rebuilds_ttl_from_final_value() {
let dir = tempdir().expect("tempdir");
let key = 23u64.to_be_bytes();
{
let db = Db::open_test(dir.path()).expect("open v1 db");
let tree = db
.open_typed_tree::<TtlFormatV1, RapiraCodec, _>(Config::test(), NoHook, &[])
.expect("open v1 tree");
tree.put(&key, TtlFormatV1 { deadline: 400 })
.expect("seed ttl source");
db.close().expect("close v1 db");
}
let db = Db::open_test(dir.path()).expect("open v2 db");
let tree = db
.open_typed_tree::<TtlFormatV2, RapiraCodec, _>(Config::test(), NoHook, &[])
.expect("open v2 tree");
assert_eq!(tree.get(&key).map(|row| row.expires_at_ms), Some(500));
assert_eq!(
tree.ttl_debug_deadline(&key),
Some(500),
"TTL bootstrap observes the migrated, not legacy, field"
);
}