use powdb_storage::catalog::Catalog;
use powdb_storage::types::{ColumnDef, Schema, TypeId, Value};
fn tmp(tag: &str) -> std::path::PathBuf {
use std::sync::atomic::{AtomicU64, Ordering};
static CTR: AtomicU64 = AtomicU64::new(0);
let uniq = CTR.fetch_add(1, Ordering::Relaxed);
let p = std::env::temp_dir().join(format!(
"powdb_sync_bootstrap_{tag}_{}_{}_{}",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos(),
uniq
));
let _ = std::fs::remove_dir_all(&p);
p
}
fn schema_t() -> Schema {
Schema {
table_name: "T".into(),
columns: vec![ColumnDef {
name: "id".into(),
type_id: TypeId::Int,
required: true,
position: 0,
}],
}
}
fn schema_u() -> Schema {
Schema {
table_name: "U".into(),
columns: vec![ColumnDef {
name: "id".into(),
type_id: TypeId::Int,
required: true,
position: 0,
}],
}
}
fn insert_range(cat: &mut Catalog, start: i64, end: i64) {
for i in start..end {
cat.insert("T", &vec![Value::Int(i)]).unwrap();
}
cat.commit_autocommit().unwrap();
cat.sync_wal().unwrap();
}
#[test]
fn bootstrap_rejects_post_snapshot_ddl_tail_without_publishing_cursor() {
let primary = tmp("ddl_tail_primary");
let mut cat = Catalog::create(&primary).unwrap();
cat.create_table(schema_t()).unwrap();
insert_range(&mut cat, 0, 2);
powdb_sync::open_or_create_identity(&primary).unwrap();
let backup = tmp("ddl_tail_backup");
let manifest = powdb_backup::full_backup(&mut cat, &backup).unwrap();
let snapshot_lsn = manifest.source_lsn;
cat.create_table(schema_u()).unwrap();
powdb_sync::checkpoint_preserving_retained_segments_if_enabled(&mut cat).unwrap();
let remote_lsn = cat.max_lsn();
assert!(remote_lsn > snapshot_lsn);
let replica = tmp("ddl_tail_replica");
let err = powdb_backup::bootstrap_replica_from_full_backup(
&mut cat,
&backup,
&replica,
"replica-ddl-tail",
)
.unwrap_err();
assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput);
assert!(
err.to_string()
.contains("DDL retained units are not supported"),
"bootstrap must reject unsupported DDL tail before restore/cursor publication, got: {err}"
);
assert!(!replica.exists());
assert!(powdb_sync::read_replica_cursors(&primary)
.unwrap()
.iter()
.all(|cursor| cursor.replica_id != "replica-ddl-tail"));
}
#[test]
fn bootstrap_replica_from_full_backup_pins_cursor_and_validates_tail() {
let primary = tmp("primary");
let mut cat = Catalog::create(&primary).unwrap();
cat.create_table(schema_t()).unwrap();
insert_range(&mut cat, 0, 10);
let identity = powdb_sync::open_or_create_identity(&primary).unwrap();
let backup = tmp("backup");
let manifest = powdb_backup::full_backup(&mut cat, &backup).unwrap();
let snapshot_lsn = manifest.source_lsn;
assert!(manifest.sync.is_some());
insert_range(&mut cat, 10, 15);
let remote_lsn = cat.max_lsn();
assert!(remote_lsn > snapshot_lsn);
let replica = tmp("replica");
let summary =
powdb_backup::bootstrap_replica_from_full_backup(&mut cat, &backup, &replica, "replica-a")
.unwrap();
assert_eq!(summary.replica_id, "replica-a");
assert_eq!(summary.snapshot_lsn, snapshot_lsn);
assert_eq!(summary.remote_lsn, remote_lsn);
assert_eq!(summary.retained_tail_start_lsn, Some(snapshot_lsn + 1));
assert_eq!(summary.retained_tail_end_lsn, Some(remote_lsn));
assert_eq!(
summary.retained_units_available as u64,
remote_lsn - snapshot_lsn
);
let cursors = powdb_sync::read_replica_cursors(&primary).unwrap();
let cursor = cursors
.iter()
.find(|cursor| cursor.replica_id == "replica-a")
.expect("bootstrap should publish an active primary-side cursor");
assert!(cursor.active);
assert_eq!(cursor.applied_lsn, snapshot_lsn);
assert_eq!(powdb_sync::read_identity(&replica).unwrap(), identity);
let restored = Catalog::open(&replica).unwrap();
assert_eq!(
restored.scan("T").unwrap().count(),
10,
"bootstrap restores the snapshot; retained-tail apply is covered by sync_apply"
);
}
#[test]
fn bootstrap_rejects_legacy_backup_without_sync_metadata() {
let primary = tmp("legacy_primary");
let mut cat = Catalog::create(&primary).unwrap();
cat.create_table(schema_t()).unwrap();
insert_range(&mut cat, 0, 1);
let backup = tmp("legacy_backup");
let manifest = powdb_backup::full_backup(&mut cat, &backup).unwrap();
assert!(manifest.sync.is_none());
let replica = tmp("legacy_replica");
let err = powdb_backup::bootstrap_replica_from_full_backup(
&mut cat,
&backup,
&replica,
"replica-legacy",
)
.unwrap_err();
assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput);
assert!(err.to_string().contains("sync-enabled"));
}
#[test]
fn bootstrap_rejects_missing_retained_tail_without_publishing_cursor() {
let primary = tmp("missing_tail_primary");
let mut cat = Catalog::create(&primary).unwrap();
cat.create_table(schema_t()).unwrap();
insert_range(&mut cat, 0, 2);
powdb_sync::open_or_create_identity(&primary).unwrap();
let backup = tmp("missing_tail_backup");
let manifest = powdb_backup::full_backup(&mut cat, &backup).unwrap();
insert_range(&mut cat, 2, 3);
powdb_sync::checkpoint_preserving_retained_segments_if_enabled(&mut cat).unwrap();
let remote_lsn = cat.max_lsn();
assert!(remote_lsn > manifest.source_lsn);
for file in powdb_sync::list_segment_files(&powdb_sync::retained_segments_dir(&primary))
.unwrap()
.into_iter()
.filter(|file| file.end_lsn > manifest.source_lsn)
{
std::fs::remove_file(file.path).unwrap();
}
let replica = tmp("missing_tail_replica");
let err = powdb_backup::bootstrap_replica_from_full_backup(
&mut cat,
&backup,
&replica,
"replica-missing-tail",
)
.unwrap_err();
assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
assert!(err.to_string().contains("missing required LSN"));
assert!(powdb_sync::read_replica_cursors(&primary)
.unwrap()
.iter()
.all(|cursor| cursor.replica_id != "replica-missing-tail"));
assert_eq!(cat.max_lsn(), remote_lsn);
}
#[test]
fn bootstrap_rejects_primary_identity_mismatch() {
let source = tmp("source");
let mut source_cat = Catalog::create(&source).unwrap();
source_cat.create_table(schema_t()).unwrap();
insert_range(&mut source_cat, 0, 1);
powdb_sync::open_or_create_identity(&source).unwrap();
let backup = tmp("mismatch_backup");
let manifest = powdb_backup::full_backup(&mut source_cat, &backup).unwrap();
drop(source_cat);
let other_primary = tmp("other_primary");
let mut other_cat = Catalog::create(&other_primary).unwrap();
other_cat.create_table(schema_t()).unwrap();
insert_range(&mut other_cat, 0, 1);
powdb_sync::open_or_create_identity(&other_primary).unwrap();
let replica = tmp("mismatch_replica");
let err = powdb_backup::bootstrap_replica_from_full_backup(
&mut other_cat,
&backup,
&replica,
"replica-mismatch",
)
.unwrap_err();
assert!(manifest.source_lsn > 0);
assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput);
assert!(err.to_string().contains("does not match"));
assert!(powdb_sync::read_replica_cursors(&other_primary)
.unwrap()
.is_empty());
}
#[test]
fn bootstrap_rejects_existing_active_cursor_without_clobbering_it() {
let primary = tmp("active_cursor_primary");
let mut cat = Catalog::create(&primary).unwrap();
cat.create_table(schema_t()).unwrap();
insert_range(&mut cat, 0, 2);
powdb_sync::open_or_create_identity(&primary).unwrap();
let backup = tmp("active_cursor_backup");
let manifest = powdb_backup::full_backup(&mut cat, &backup).unwrap();
insert_range(&mut cat, 2, 4);
powdb_sync::checkpoint_preserving_retained_segments_if_enabled(&mut cat).unwrap();
let existing_applied_lsn = cat.max_lsn();
powdb_sync::upsert_replica_cursor(
&primary,
powdb_sync::ReplicaCursor::active("replica-existing", existing_applied_lsn),
)
.unwrap();
let replica = tmp("active_cursor_replica");
let err = powdb_backup::bootstrap_replica_from_full_backup(
&mut cat,
&backup,
&replica,
"replica-existing",
)
.unwrap_err();
assert_eq!(err.kind(), std::io::ErrorKind::AlreadyExists);
assert!(!replica.exists());
let cursors = powdb_sync::read_replica_cursors(&primary).unwrap();
let cursor = cursors
.iter()
.find(|cursor| cursor.replica_id == "replica-existing")
.unwrap();
assert!(cursor.active);
assert_eq!(cursor.applied_lsn, existing_applied_lsn);
assert!(manifest.source_lsn < existing_applied_lsn);
}
#[test]
fn bootstrap_cleans_restored_replica_when_cursor_publication_fails() {
let primary = tmp("cleanup_primary");
let mut cat = Catalog::create(&primary).unwrap();
cat.create_table(schema_t()).unwrap();
insert_range(&mut cat, 0, 2);
powdb_sync::open_or_create_identity(&primary).unwrap();
let backup = tmp("cleanup_backup");
let manifest = powdb_backup::full_backup(&mut cat, &backup).unwrap();
insert_range(&mut cat, 2, 4);
powdb_sync::checkpoint_preserving_retained_segments_if_enabled(&mut cat).unwrap();
let replica = tmp("cleanup_replica");
let err = powdb_backup::bootstrap_replica_from_full_backup(
&mut cat,
&backup,
&replica,
"replica/invalid",
)
.unwrap_err();
assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput);
assert!(
err.to_string().contains("replica id contains unsupported"),
"invalid replica id should fail during cursor publication, got: {err}"
);
assert!(
!replica.exists(),
"bootstrap must remove the restored replica directory when cursor publication fails"
);
assert!(powdb_sync::read_replica_cursors(&primary)
.unwrap()
.is_empty());
assert!(manifest.source_lsn < cat.max_lsn());
}