#![allow(dead_code)]
use std::{
collections::VecDeque,
fs::{self, File, OpenOptions},
os::unix::fs::FileExt,
path::Path,
sync::Mutex,
time::Instant,
};
#[allow(unused_imports)]
use crate::format_migration::marker::read_shard_version;
use crate::{
DbError, DbResult, FixedConfig, FormatMigrationReport,
fixed::{
shard::{FixedHeader, HEADER_SIZE, read_header, write_header},
slot,
},
format_migration::{
bitcask::BackendMigrationOutcome,
marker::{
ConfiguredShard, ShardSwapState, configured_shards, swap_prepared_shard,
write_shard_version,
},
plan::{MigrationPlan, TransformBuffers, TransformOutcome},
},
io::direct,
};
pub(crate) fn migrate_fixed(
collection_dir: &Path,
collection: &'static str,
source_version: u16,
plan: &MigrationPlan,
config: &FixedConfig,
key_len: usize,
target_value_len: u16,
) -> DbResult<BackendMigrationOutcome> {
let started = Instant::now();
let shards = configured_shards(
collection_dir,
config.shard_count,
source_version,
plan.format_to_version,
plan.physical_floor,
)?;
let workers = shards
.len()
.min(std::thread::available_parallelism().map_or(1, std::num::NonZero::get));
let queue = Mutex::new(VecDeque::from(shards));
let results = Mutex::new(Vec::new());
std::thread::scope(|scope| -> DbResult<()> {
let mut handles = Vec::with_capacity(workers);
for _ in 0..workers {
handles.push(scope.spawn(|| -> DbResult<()> {
while let Some(ConfiguredShard {
id: shard_id,
name,
state,
}) = queue.lock().expect("migration queue poisoned").pop_front()
{
let result = rewrite_shard(
collection_dir,
&name,
shard_id,
state,
plan,
key_len,
target_value_len,
)?;
results
.lock()
.expect("migration results poisoned")
.push((shard_id, result));
}
Ok(())
}));
}
for handle in handles {
handle.join().map_err(|_| DbError::RecoveryPanic)??;
}
Ok(())
})?;
let mut report = FormatMigrationReport::empty(collection, source_version);
report.to = plan.format_to_version;
report.warnings.extend(plan.warnings.iter().cloned());
let mut results = results.into_inner().expect("migration results poisoned");
results.sort_by_key(|(shard, _)| *shard);
report.shards = results.len();
let mut torn_slots = 0_u64;
for (_, (migrated, dropped, torn, before, after)) in results {
report.rows_migrated += migrated;
report.rows_dropped += dropped;
torn_slots += torn;
report.bytes_before += before;
report.bytes_after += after;
}
if torn_slots != 0 {
report.warnings.push(format!(
"recovered {torn_slots} torn FixedStore occupied slot(s) as free"
));
}
report.elapsed = started.elapsed();
Ok(BackendMigrationOutcome {
report,
gsn_high: 0,
shard_gsn_high: vec![0; config.shard_count],
})
}
#[allow(clippy::too_many_arguments)]
fn rewrite_shard(
collection_dir: &Path,
name: &str,
shard_id: usize,
state: ShardSwapState,
plan: &MigrationPlan,
key_len: usize,
target_value_len: u16,
) -> DbResult<(u64, u64, u64, u64, u64)> {
let source = collection_dir.join(name);
let version = match state {
ShardSwapState::AlreadyTarget => return Ok((0, 0, 0, 0, 0)),
ShardSwapState::DeleteOld => {
fs::remove_dir_all(collection_dir.join(format!("{name}.old")))?;
let migration = collection_dir.join(format!("{name}.mig"));
if migration.exists() {
fs::remove_dir_all(migration)?;
}
direct::sync_dir(collection_dir)?;
return Ok((0, 0, 0, 0, 0));
}
ShardSwapState::DeleteMig => {
fs::remove_dir_all(collection_dir.join(format!("{name}.mig")))?;
direct::sync_dir(collection_dir)?;
return Ok((0, 0, 0, 0, 0));
}
ShardSwapState::DeleteOldThenRewrite(version) => {
fs::remove_dir_all(collection_dir.join(format!("{name}.old")))?;
let migration = collection_dir.join(format!("{name}.mig"));
if migration.exists() {
fs::remove_dir_all(migration)?;
}
direct::sync_dir(collection_dir)?;
version
}
ShardSwapState::FinishSwap => {
swap_prepared_shard(collection_dir, name)?;
return Ok((0, 0, 0, 0, 0));
}
ShardSwapState::DeleteMigThenRewrite(version) => {
fs::remove_dir_all(collection_dir.join(format!("{name}.mig")))?;
direct::sync_dir(collection_dir)?;
version
}
ShardSwapState::RestoreOldThenRewrite(version) => {
let migration = collection_dir.join(format!("{name}.mig"));
if migration.exists() {
fs::remove_dir_all(migration)?;
direct::sync_dir(collection_dir)?;
}
fs::rename(collection_dir.join(format!("{name}.old")), &source)?;
direct::sync_dir(collection_dir)?;
version
}
ShardSwapState::RewriteFrom(version) => version,
};
let source_path = source.join("fixed.data");
let source_file = File::open(&source_path)?;
let before = source_file.metadata()?.len();
let mut header_bytes = [0_u8; HEADER_SIZE as usize];
source_file.read_exact_at(&mut header_bytes, 0)?;
let source_header = read_header(&header_bytes)?;
validate_source_header(source_header, shard_id, key_len)?;
let target_slot_size = u16::try_from(slot::slot_size(key_len, target_value_len as usize))
.map_err(|_| {
DbError::Config("slot_size exceeds u16::MAX (key_len + value_len + header too large)")
})?;
let expected = HEADER_SIZE + source_header.slot_count as u64 * source_header.slot_size as u64;
if before < expected {
return Err(DbError::FormatMismatch(format!(
"fixed.data: file truncated: expected >= {expected} bytes, got {before}"
)));
}
let migration = collection_dir.join(format!("{name}.mig"));
fs::create_dir(&migration)?;
let mut cleanup = FixedMigrationDirGuard::new(&migration);
let target_path = migration.join("fixed.data");
let target_file = OpenOptions::new()
.create_new(true)
.read(true)
.write(true)
.open(&target_path)?;
let target_header = FixedHeader {
slot_size: target_slot_size,
slot_count: source_header.slot_count,
key_len: source_header.key_len,
value_len: target_value_len,
shard_id: source_header.shard_id,
};
target_file.set_len(HEADER_SIZE + source_header.slot_count as u64 * target_slot_size as u64)?;
let defined_header = write_header(target_header);
header_bytes[..18].copy_from_slice(&defined_header[..18]);
target_file.write_all_at(&header_bytes, 0)?;
let mut source_slot = vec![0_u8; source_header.slot_size as usize];
let mut target_slot = vec![0_u8; target_slot_size as usize];
let mut buffers = TransformBuffers::default();
let mut migrated = 0;
let mut dropped = 0;
let mut torn = 0;
for slot_id in 0..source_header.slot_count {
let source_offset = HEADER_SIZE + slot_id as u64 * source_header.slot_size as u64;
source_file.read_exact_at(&mut source_slot, source_offset)?;
let meta = slot::meta_of(&source_slot);
let target_offset = HEADER_SIZE + slot_id as u64 * target_slot_size as u64;
if slot::status_of(meta) != slot::STATUS_OCCUPIED {
target_file.write_all_at(&meta.to_le_bytes(), target_offset)?;
continue;
}
let Some((_, key, value)) =
slot::read_slot(&source_slot, key_len, source_header.value_len as usize)
else {
let free = slot::with_status(meta, slot::STATUS_FREE);
target_file.write_all_at(&free.to_le_bytes(), target_offset)?;
torn += 1;
tracing::warn!(
collection = plan.collection,
shard = shard_id,
slot = slot_id,
offset = source_offset,
"recovered torn FixedStore occupied slot as free during format migration"
);
continue;
};
match plan.transform_from_version(version, key, value, &mut buffers)? {
TransformOutcome::Drop => {
dropped += 1;
let free = slot::with_status(meta, slot::STATUS_FREE);
target_file.write_all_at(&free.to_le_bytes(), target_offset)?;
}
TransformOutcome::Keep(value) | TransformOutcome::Write(value) => {
if value.len() != target_value_len as usize {
return Err(DbError::FormatMismatch(format!(
"fixed format migration wrote {} bytes for target width {target_value_len}",
value.len()
)));
}
slot::serialize_slot(&mut target_slot, meta, key, value);
target_file.write_all_at(&target_slot, target_offset)?;
migrated += 1;
}
}
}
target_file.sync_all()?;
write_shard_version(&migration, plan.format_to_version)?;
hit_failpoint(AFTER_FORMAT_VERSION);
direct::sync_dir(&migration)?;
swap_prepared_shard(collection_dir, name)?;
cleanup.disarm();
Ok((
migrated,
dropped,
torn,
before,
HEADER_SIZE + source_header.slot_count as u64 * target_slot_size as u64,
))
}
struct FixedMigrationDirGuard<'a> {
path: &'a Path,
armed: bool,
}
impl<'a> FixedMigrationDirGuard<'a> {
fn new(path: &'a Path) -> Self {
Self { path, armed: true }
}
fn disarm(&mut self) {
self.armed = false;
}
}
impl Drop for FixedMigrationDirGuard<'_> {
fn drop(&mut self) {
if !self.armed {
return;
}
match fs::remove_dir_all(self.path) {
Ok(()) => {
if let Some(parent) = self.path.parent()
&& let Err(error) = direct::sync_dir(parent)
{
tracing::warn!(path = ?self.path, %error, "fixed migration cleanup sync failed");
}
}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(error) => {
tracing::warn!(path = ?self.path, %error, "fixed migration cleanup failed");
}
}
}
}
#[cfg(any(test, feature = "test-faults"))]
const AFTER_FORMAT_VERSION: u8 = crate::format_migration::failpoints::AFTER_FORMAT_VERSION;
#[cfg(not(any(test, feature = "test-faults")))]
const AFTER_FORMAT_VERSION: u8 = 0;
#[cfg(any(test, feature = "test-faults"))]
fn hit_failpoint(stage: u8) {
crate::format_migration::failpoints::hit(stage);
}
#[cfg(not(any(test, feature = "test-faults")))]
fn hit_failpoint(_: u8) {}
fn validate_source_header(header: FixedHeader, shard_id: usize, key_len: usize) -> DbResult<()> {
if header.key_len as usize != key_len {
return Err(DbError::FormatMismatch(format!(
"fixed.data: key_len mismatch: stored {}, expected {key_len}",
header.key_len
)));
}
if header.shard_id as usize != shard_id {
return Err(DbError::FormatMismatch(format!(
"fixed.data: shard_id mismatch: stored {}, expected {shard_id}",
header.shard_id
)));
}
if header.slot_size as usize != slot::slot_size(key_len, header.value_len as usize) {
return Err(DbError::FormatMismatch(
"fixed.data: invalid slot_size".into(),
));
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{FormatAction, FormatMigration};
use tempfile::tempdir;
fn widen(_key: &[u8], old: &[u8], out: &mut Vec<u8>) -> 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);
fn drop_first(key: &[u8], _old: &[u8], _out: &mut Vec<u8>) -> DbResult<FormatAction> {
if key == 1u64.to_be_bytes() {
Ok(FormatAction::Drop)
} else {
Ok(FormatAction::Keep)
}
}
fn wrong_width(_key: &[u8], _old: &[u8], out: &mut Vec<u8>) -> DbResult<FormatAction> {
out.extend_from_slice(&[0; 7]);
Ok(FormatAction::Write)
}
const DROP_FIRST: FormatMigration = FormatMigration::raw(1, drop_first);
const WRONG_WIDTH: FormatMigration = FormatMigration::raw(1, wrong_width);
fn plan() -> MigrationPlan {
plan_with(&WIDEN)
}
fn plan_with(step: &'static FormatMigration) -> MigrationPlan {
MigrationPlan {
collection: "fixed-physical",
stored_version: 1,
physical_floor: 1,
unchanged_runs: vec![],
replay_history: vec![],
crossed: vec![],
target_version: 2,
format_to_version: 2,
format_steps: vec![step],
accepted_fingerprints: vec![],
warnings: vec![],
}
}
fn fixture() -> tempfile::TempDir {
let root = tempdir().expect("tempdir");
let shard = root.path().join("shard_000");
fs::create_dir(&shard).expect("shard");
let header = FixedHeader {
slot_size: slot::slot_size(8, 4) as u16,
slot_count: 3,
key_len: 8,
value_len: 4,
shard_id: 0,
};
let file = OpenOptions::new()
.create_new(true)
.read(true)
.write(true)
.open(shard.join("fixed.data"))
.expect("data");
file.set_len(HEADER_SIZE + header.slot_count as u64 * header.slot_size as u64)
.expect("size");
file.write_all_at(&write_header(header), 0).expect("header");
let meta = slot::pack_meta(slot::STATUS_OCCUPIED, 17);
let mut row = vec![0; header.slot_size as usize];
slot::serialize_slot(&mut row, meta, &1u64.to_be_bytes(), &9u32.to_ne_bytes());
file.write_all_at(&row, HEADER_SIZE).expect("row");
let deleted = slot::pack_meta(slot::STATUS_DELETED, 11);
file.write_all_at(
&deleted.to_le_bytes(),
HEADER_SIZE + header.slot_size as u64,
)
.expect("deleted meta");
file.sync_all().expect("sync fixture");
root
}
fn config() -> FixedConfig {
FixedConfig {
shard_count: 1,
..FixedConfig::test()
}
}
#[test]
fn format_migration_fixed_rewrites_slots_without_changing_meta() {
let root = fixture();
let outcome = migrate_fixed(root.path(), "fixed-physical", 1, &plan(), &config(), 8, 8)
.expect("rewrite fixed data");
assert_eq!(outcome.report.rows_migrated, 1);
assert_eq!(outcome.report.rows_dropped, 0);
let bytes = fs::read(root.path().join("shard_000/fixed.data")).expect("output");
let header = read_header(&bytes[..HEADER_SIZE as usize]).expect("output header");
assert_eq!(header.value_len, 8);
assert_eq!(header.slot_size as usize, slot::slot_size(8, 8));
let offset = HEADER_SIZE as usize;
let (meta, key, value) = slot::read_slot(&bytes[offset..], 8, 8).expect("surviving row");
assert_eq!(meta, slot::pack_meta(slot::STATUS_OCCUPIED, 17));
assert_eq!(key, 1u64.to_be_bytes());
assert_eq!(value, 9u64.to_ne_bytes());
let deleted_offset = offset + header.slot_size as usize;
assert_eq!(
slot::meta_of(&bytes[deleted_offset..]),
slot::pack_meta(slot::STATUS_DELETED, 11)
);
}
#[test]
fn format_migration_fixed_drop_frees_same_slot_for_reuse() {
let root = fixture();
migrate_fixed(
root.path(),
"fixed-physical",
1,
&plan_with(&DROP_FIRST),
&config(),
8,
8,
)
.expect("drop rewrite");
let mut shard = crate::fixed::shard::FixedShardInner::open(
root.path().join("shard_000"),
0,
8,
8,
&config(),
)
.expect("open rewritten shard");
shard
.recover_slots(
crate::fixed::shard::RecoverMode::Dirty,
4096,
|_k, _v, _slot| None,
)
.expect("recover rewritten shard");
assert_eq!(shard.alloc_slot().expect("reuse dropped slot"), 0);
assert_eq!(
shard.versions[0],
slot::pack_meta(slot::STATUS_FREE, 17),
"Drop must retain the slot version while releasing allocation"
);
}
#[test]
fn format_migration_fixed_recovers_torn_occupied_slot_as_free() {
let root = fixture();
let path = root.path().join("shard_000/fixed.data");
let file = OpenOptions::new().write(true).open(&path).expect("source");
file.write_all_at(&[0xff], HEADER_SIZE + slot::SLOT_HEADER_SIZE as u64 + 8)
.expect("corrupt value");
file.sync_all().expect("persist corruption");
let outcome = migrate_fixed(root.path(), "fixed-physical", 1, &plan(), &config(), 8, 8)
.expect("dirty recovery frees a torn occupied slot");
assert_eq!(outcome.report.rows_migrated, 0);
assert!(
outcome
.report
.warnings
.iter()
.any(|warning| warning.contains("1 torn FixedStore occupied slot")),
"{:?}",
outcome.report.warnings
);
let bytes = fs::read(path).expect("rewritten source");
let meta = slot::meta_of(&bytes[HEADER_SIZE as usize..]);
assert_eq!(slot::status_of(meta), slot::STATUS_FREE);
assert_eq!(slot::version_of(meta), 17);
}
#[test]
fn format_migration_fixed_error_cleans_partial_output_and_retries() {
let root = fixture();
let result = migrate_fixed(
root.path(),
"fixed-physical",
1,
&plan_with(&WRONG_WIDTH),
&config(),
8,
8,
);
assert!(
matches!(result, Err(DbError::FormatMismatch(message)) if message.contains("wrote 7 bytes"))
);
assert!(root.path().join("shard_000/fixed.data").exists());
assert!(!root.path().join("shard_000.mig").exists());
migrate_fixed(root.path(), "fixed-physical", 1, &plan(), &config(), 8, 8)
.expect("retry after callback error");
}
#[test]
fn format_migration_fixed_preserves_reserved_header_bytes() {
let root = fixture();
let path = root.path().join("shard_000/fixed.data");
let file = OpenOptions::new().write(true).open(&path).expect("source");
file.write_all_at(&[0xa5, 0x5a, 0x11, 0xee], 18)
.expect("seed reserved header bytes");
file.sync_all().expect("persist reserved bytes");
migrate_fixed(root.path(), "fixed-physical", 1, &plan(), &config(), 8, 8)
.expect("rewrite fixed data");
let bytes = fs::read(path).expect("rewritten source");
assert_eq!(&bytes[18..22], &[0xa5, 0x5a, 0x11, 0xee]);
let header = read_header(&bytes[..HEADER_SIZE as usize]).expect("rewritten header");
assert_eq!(header.value_len, 8);
}
#[test]
fn format_migration_fixed_recovers_interrupted_swap_layouts() {
let root = fixture();
let shard = root.path().join("shard_000");
let old = root.path().join("shard_000.old");
let saved_old = root.path().join("saved-v1");
fs::create_dir(&saved_old).expect("saved source directory");
fs::copy(shard.join("fixed.data"), saved_old.join("fixed.data"))
.expect("save source before rewrite");
migrate_fixed(root.path(), "fixed-physical", 1, &plan(), &config(), 8, 8)
.expect("first rewrite");
fs::rename(root.path().join("saved-v1"), &old).expect("install old source");
fs::rename(&shard, root.path().join("shard_000.mig"))
.expect("simulate crash after first swap rename");
migrate_fixed(root.path(), "fixed-physical", 1, &plan(), &config(), 8, 8)
.expect("finish prepared swap");
assert!(shard.join("fixed.data").exists());
assert!(!root.path().join("shard_000.mig").exists());
assert!(!old.exists());
assert_eq!(read_shard_version(&shard, 1).expect("version"), 2);
}
}