use std::ffi::OsString;
use std::fs::{self, File, OpenOptions};
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use crate::franken_sync::Connection;
use crate::franken_sync::compat::{OpenFlags, open_with_flags};
use chrono::Utc;
use fsqlite_types::SqliteValue;
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use crate::cli::{
DoctorMigrateSchemaApplyArgs, DoctorMigrateSchemaArgs, DoctorMigrateSchemaCommand,
DoctorMigrateSchemaPlanArgs, DoctorMigrateSchemaRecoverArgs, DoctorMigrateSchemaUndoArgs,
};
use crate::config;
use crate::error::{BeadsError, Result};
use crate::output::OutputContext;
use crate::storage::schema::{
CURRENT_SCHEMA_VERSION, REVIEWED_LEGACY_V15_RETIRED_INDEXES,
REVIEWED_MIGRATION_SOURCE_VERSIONS, ReviewedSchemaMigrationEffects,
apply_reviewed_legacy_v15_core_conversion_in_transaction,
attest_reviewed_migration_source_core_tables, is_reviewed_migration_managed_index,
run_reviewed_schema_migration_steps_in_transaction, runtime_schema_compatible,
};
use crate::sync::{DatabaseFamilyWriteLock, DatabaseTargetAuthorityState};
const PLAN_SCHEMA: &str = "br.doctor.schema_migration.plan.v1";
const PREPARED_SCHEMA: &str = "br.doctor.schema_migration.prepared.v1";
const COMMIT_READY_SCHEMA: &str = "br.doctor.schema_migration.commit_ready.v1";
const APPLIED_SCHEMA: &str = "br.doctor.schema_migration.applied.v1";
const FAILED_SCHEMA: &str = "br.doctor.schema_migration.failed.v1";
const UNDO_SCHEMA: &str = "br.doctor.schema_migration.undo.v1";
const FAMILY_SUFFIXES: &[&str] = &["", "-wal", "-shm", "-journal"];
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
struct RawComponentWitness {
suffix: String,
present: bool,
length: Option<u64>,
sha256: Option<String>,
unix_mode: Option<u32>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
struct RawFamilyWitness {
components: Vec<RawComponentWitness>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
struct LogicalTableWitness {
name: String,
row_count: u64,
rows_sha256: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
struct LogicalDatabaseWitness {
user_version: u32,
integrity_check: String,
schema_sha256: String,
contents_sha256: String,
tables: Vec<LogicalTableWitness>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
struct LegacyTableProjection {
#[serde(flatten)]
table: LogicalTableWitness,
rowid_included: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
struct LegacyV15Projection {
profile: String,
tables: Vec<LegacyTableProjection>,
preserved_schema_sha256: String,
retired_indexes: Vec<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
struct MigrationForecast {
from_version: u32,
to_version: u32,
content_hash_rows_rebuilt: usize,
gate_result_history_created: bool,
#[serde(default, skip_serializing_if = "is_false")]
prerequisites_column_added: bool,
#[serde(default, skip_serializing_if = "is_false")]
dependency_type_key_added: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
legacy_v15_conversion: Option<LegacyV15Projection>,
#[serde(default, skip_serializing_if = "is_false")]
post_migration_maintenance: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct MigrationPlanReceipt {
schema_version: String,
eligible: bool,
database_path: String,
from_version: u32,
to_version: u32,
raw_witness: RawFamilyWitness,
logical_witness: LogicalDatabaseWitness,
forecast: Option<MigrationForecast>,
plan_token: Option<String>,
apply_command: Option<String>,
note: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
struct PreparedMigrationReceipt {
schema_version: String,
run_id: String,
database_path: String,
plan_token: String,
marked_at: String,
forecast: MigrationForecast,
raw_before: RawFamilyWitness,
logical_before: LogicalDatabaseWitness,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
struct CommitReadyMigrationReceipt {
schema_version: String,
run_id: String,
database_path: String,
plan_token: String,
prepared_receipt_sha256: String,
marked_at: String,
forecast: MigrationForecast,
effects: ReviewedSchemaMigrationEffectsReceipt,
raw_before: RawFamilyWitness,
logical_before: LogicalDatabaseWitness,
logical_after: LogicalDatabaseWitness,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct AppliedMigrationReceipt {
schema_version: String,
run_id: String,
database_path: String,
plan_token: String,
prepared_receipt_sha256: String,
marked_at: String,
forecast: MigrationForecast,
effects: ReviewedSchemaMigrationEffectsReceipt,
raw_before: RawFamilyWitness,
logical_before: LogicalDatabaseWitness,
raw_after: Option<RawFamilyWitness>,
logical_after: Option<LogicalDatabaseWitness>,
attested: bool,
attestation_errors: Vec<String>,
undo_command: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct FailedMigrationReceipt {
schema_version: String,
run_id: String,
database_path: String,
plan_token: String,
marked_at: String,
error: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
failed_stage: Option<String>,
raw_before: RawFamilyWitness,
logical_before: LogicalDatabaseWitness,
raw_observed_after_failure: Option<RawFamilyWitness>,
logical_observed_after_failure: Option<LogicalDatabaseWitness>,
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
struct ReviewedSchemaMigrationEffectsReceipt {
from_version: u32,
to_version: u32,
content_hash_rows_rebuilt: usize,
gate_result_history_created: bool,
#[serde(default, skip_serializing_if = "is_false")]
prerequisites_column_added: bool,
#[serde(default, skip_serializing_if = "is_false")]
dependency_type_key_added: bool,
#[serde(default, skip_serializing_if = "is_false")]
legacy_v15_tables_converted: bool,
#[serde(default, skip_serializing_if = "is_false")]
post_migration_maintenance_completed: bool,
}
impl From<ReviewedSchemaMigrationEffects> for ReviewedSchemaMigrationEffectsReceipt {
fn from(value: ReviewedSchemaMigrationEffects) -> Self {
Self {
from_version: value.from_version,
to_version: value.to_version,
content_hash_rows_rebuilt: value.content_hash_rows_rebuilt,
gate_result_history_created: value.gate_result_history_created,
prerequisites_column_added: value.prerequisites_column_added,
dependency_type_key_added: value.dependency_type_key_added,
legacy_v15_tables_converted: false,
post_migration_maintenance_completed: false,
}
}
}
#[allow(clippy::trivially_copy_pass_by_ref)]
const fn is_false(value: &bool) -> bool {
!*value
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct UndoReceipt {
schema_version: String,
run_id: String,
dry_run: bool,
database_path: String,
quarantine_path: String,
applied_receipt_sha256: String,
raw_expected_before: RawFamilyWitness,
logical_expected_before: LogicalDatabaseWitness,
raw_live_before_undo: RawFamilyWitness,
logical_live_before_undo: Option<LogicalDatabaseWitness>,
raw_restored: Option<RawFamilyWitness>,
logical_restored: Option<LogicalDatabaseWitness>,
}
#[derive(Serialize)]
struct PlanTokenMaterial<'a> {
contract: &'static str,
database_path: &'a str,
from_version: u32,
to_version: u32,
logical_witness: &'a LogicalDatabaseWitness,
forecast: &'a MigrationForecast,
}
pub(super) struct MigrationContext {
pub(super) beads_dir: PathBuf,
pub(super) db_path: PathBuf,
pub(super) write_authority: Arc<DatabaseFamilyWriteLock>,
}
pub fn execute(
args: &DoctorMigrateSchemaArgs,
cli: &config::CliOverrides,
_ctx: &OutputContext,
) -> Result<()> {
let migration = resolve_context(cli)?;
match &args.command {
DoctorMigrateSchemaCommand::Recover(recover) => execute_recover(recover, &migration),
DoctorMigrateSchemaCommand::Plan(plan) => execute_plan(plan, &migration),
DoctorMigrateSchemaCommand::Apply(apply) => execute_apply(apply, &migration),
DoctorMigrateSchemaCommand::Undo(undo) => execute_undo(undo, &migration),
DoctorMigrateSchemaCommand::Heal(heal) => super::schema_heal::execute_heal(
heal,
cli,
&migration.beads_dir,
&migration.write_authority,
),
}
}
fn resolve_context(cli: &config::CliOverrides) -> Result<MigrationContext> {
let beads_dir =
config::discover_optional_beads_dir_with_cli(cli)?.ok_or(BeadsError::NotInitialized)?;
let paths = config::resolve_paths(&beads_dir, cli.db.as_ref())?;
let write_authority = if let Some(authority) =
cli.database_family_write_authority_for(&beads_dir, &paths.db_path)
{
authority.verify_database_authority()?;
Arc::clone(authority)
} else {
Arc::new(
crate::sync::blocking_database_family_write_lock_with_timeout(
&beads_dir,
&paths.db_path,
cli.lock_timeout,
)?,
)
};
write_authority.verify_database_authority()?;
Ok(MigrationContext {
beads_dir,
db_path: paths.db_path,
write_authority,
})
}
fn execute_plan(args: &DoctorMigrateSchemaPlanArgs, migration: &MigrationContext) -> Result<()> {
let plan = build_plan(&migration.db_path)?;
emit_plan(&plan, args.json)?;
Ok(())
}
#[derive(Debug, Serialize)]
struct EngineRecoveryReceipt {
schema_version: &'static str,
database_path: String,
backup_path: String,
backup_scope: &'static str,
stage: String,
raw_before: RawFamilyWitness,
raw_after: Option<RawFamilyWitness>,
logical_after: Option<LogicalDatabaseWitness>,
#[serde(skip_serializing_if = "Option::is_none")]
index_corruption: Option<IndexCorruptionFinding>,
error: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
failure_kind: Option<RecoveryFailureKind>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "kebab-case")]
enum RecoveryFailureKind {
TransientContention,
FailClosed,
}
impl RecoveryFailureKind {
fn for_error(stage: &str, error: &BeadsError) -> Self {
if stage == "live-recovery" && live_recovery_contention(error) {
Self::TransientContention
} else {
Self::FailClosed
}
}
const fn retry_hint(self) -> &'static str {
match self {
Self::TransientContention => {
"; transient contention: close other database readers/openers and retry this command"
}
Self::FailClosed => "",
}
}
}
const fn engine_recovery_contention(error: &fsqlite_error::FrankenError) -> bool {
matches!(
error,
fsqlite_error::FrankenError::Busy
| fsqlite_error::FrankenError::BusyRecovery
| fsqlite_error::FrankenError::DatabaseLocked { .. }
)
}
fn live_recovery_contention(error: &BeadsError) -> bool {
match error {
BeadsError::Database(error) => engine_recovery_contention(error),
BeadsError::DatabaseLocked { .. } => true,
BeadsError::WithContext { source, .. } => {
source
.downcast_ref::<BeadsError>()
.is_some_and(live_recovery_contention)
|| source
.downcast_ref::<fsqlite_error::FrankenError>()
.is_some_and(engine_recovery_contention)
}
_ => false,
}
}
#[derive(Debug, Clone, Serialize)]
struct IndexCorruptionFinding {
integrity_check: String,
remediation: &'static str,
index_entries_without_table_rows: bool,
}
impl IndexCorruptionFinding {
const REMEDIATION: &'static str = "br doctor --repair-indexes";
fn summary(&self) -> String {
integrity_summary(&self.integrity_check)
}
const MISSING_ROWS_WARNING: &'static str = "Some index entries point at rows that are \
missing from their table, so rows may have been lost from the table itself; \
rebuilding the indexes drops those entries. Before running it, check issues.jsonl for \
the missing records (the complete pre-recovery family is retained).";
}
fn integrity_reports_index_entries_without_table_rows(integrity_check: &str) -> bool {
integrity_check.lines().any(|line| {
let lower = line.to_ascii_lowercase();
if lower.contains("no matching table row") {
return true;
}
let count_after = |marker: &str| {
lower.find(marker).and_then(|start| {
lower[start + marker.len()..]
.split_whitespace()
.next()
.and_then(|number| number.parse::<u64>().ok())
})
};
matches!(
(count_after(" contains "), count_after(" requires exactly ")),
(Some(entries), Some(required)) if entries > required
)
})
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum PreexistingIndexCorruption {
Refuse,
Admit,
}
fn execute_recover(
args: &DoctorMigrateSchemaRecoverArgs,
migration: &MigrationContext,
) -> Result<()> {
let receipt =
recover_engine_admission_with_lease(migration, false, PreexistingIndexCorruption::Admit)?;
if args.json {
println!("{}", serde_json::to_string_pretty(&receipt)?);
} else {
println!(
"Recovered engine read admission for {}",
receipt.database_path
);
if receipt.backup_scope == RECOVERY_BACKUP_WAL_INDEX_ONLY {
println!(
"Only the WAL index needed rebuilding; the main database and WAL were unchanged. \
Pre-recovery index retained at {}",
receipt.backup_path
);
} else {
println!("Pre-recovery family retained at {}", receipt.backup_path);
}
if let Some(finding) = &receipt.index_corruption {
println!(
"The database already had corrupt indexes ({}). A private rehearsal showed that \
rebuilding them repairs it without changing any row. Run `{}` next; it keeps a \
pre-repair snapshot.",
finding.summary(),
finding.remediation
);
if finding.index_entries_without_table_rows {
println!("{}", IndexCorruptionFinding::MISSING_ROWS_WARNING);
}
} else {
println!("Run `br doctor migrate-schema plan` to review the schema migration.");
}
}
Ok(())
}
const WAL_HEADER_BYTES: u64 = 32;
pub fn missing_wal_index(db_path: &Path) -> Result<bool> {
missing_wal_index_for_engine(
db_path,
crate::franken_sync::wal_index::STARTUP_WAL_INDEX_RECOVERY,
)
}
fn missing_wal_index_for_engine(db_path: &Path, engine_reads_on_disk_index: bool) -> Result<bool> {
if !engine_reads_on_disk_index
|| secure_file_metadata(db_path)?.is_none()
|| secure_file_metadata(&family_component_path(db_path, "-shm"))?.is_some()
{
return Ok(false);
}
Ok(
secure_file_metadata(&family_component_path(db_path, "-wal"))?
.is_some_and(|metadata| metadata.len() >= WAL_HEADER_BYTES),
)
}
pub fn torn_wal_present(db_path: &Path) -> Result<bool> {
if secure_file_metadata(db_path)?.is_none() {
return Ok(false);
}
Ok(
secure_file_metadata(&family_component_path(db_path, "-wal"))?
.is_some_and(|metadata| is_torn_wal_length(metadata.len())),
)
}
const fn is_torn_wal_length(length: u64) -> bool {
length > 0 && length < WAL_HEADER_BYTES
}
#[derive(Debug, Clone)]
pub struct TornWalQuarantine {
pub wal_length: u64,
pub quarantined_paths: Vec<PathBuf>,
}
pub fn quarantine_torn_wal_for_startup(
_beads_dir: &Path,
db_path: &Path,
authority: &Arc<DatabaseFamilyWriteLock>,
) -> Result<Option<TornWalQuarantine>> {
if crate::sync::database_write_authority_sha256(db_path)? != authority.authority_path_sha256() {
return Err(BeadsError::SyncConflict {
message: "Torn WAL quarantine path does not match the held database-family authority"
.to_string(),
});
}
authority.verify_database_authority()?;
if !torn_wal_present(db_path)? {
return Ok(None);
}
let mut lease = crate::sync::DatabaseOpenerLease::register(db_path)?;
let Some(exclusive) = lease.try_exclusive() else {
return Ok(None);
};
let result = (|| -> Result<Option<TornWalQuarantine>> {
authority.verify_database_authority()?;
let wal_path = family_component_path(db_path, "-wal");
let Some(metadata) = secure_file_metadata(&wal_path)? else {
return Ok(None);
};
let wal_length = metadata.len();
if !is_torn_wal_length(wal_length) {
return Ok(None);
}
let recovery_dir = recovery_artifacts_dir(db_path)?;
let quarantined_paths = config::quarantine_database_artifacts(
db_path,
&recovery_dir,
[wal_path, family_component_path(db_path, "-shm")],
"truncated-wal",
)?;
authority.verify_database_authority()?;
Ok(Some(TornWalQuarantine {
wal_length,
quarantined_paths,
}))
})();
let released = lease.release_exclusive(exclusive);
result.and_then(|quarantine| released.map(|()| quarantine))
}
pub fn wal_index_needs_recovery(db_path: &Path) -> Result<bool> {
wal_index_needs_recovery_for_engine(
db_path,
crate::franken_sync::wal_index::STARTUP_WAL_INDEX_RECOVERY,
)
}
fn wal_index_needs_recovery_for_engine(
db_path: &Path,
engine_reads_on_disk_index: bool,
) -> Result<bool> {
Ok(
missing_wal_index_for_engine(db_path, engine_reads_on_disk_index)?
|| crate::franken_sync::wal_index::poisoned_index_present_for_engine(
db_path,
engine_reads_on_disk_index,
)?
|| crate::franken_sync::wal_index::stale_index_present_for_engine(
db_path,
engine_reads_on_disk_index,
)?,
)
}
pub fn recover_wal_index_for_startup(
beads_dir: &Path,
db_path: &Path,
authority: &Arc<DatabaseFamilyWriteLock>,
) -> Result<()> {
if crate::sync::database_write_authority_sha256(db_path)? != authority.authority_path_sha256() {
return Err(BeadsError::SyncConflict {
message: "WAL index recovery path does not match the held database-family authority"
.to_string(),
});
}
authority.verify_database_authority()?;
if !wal_index_needs_recovery(db_path)? {
return Ok(());
}
require_frame_free_engine_recovery(db_path)?;
let raw_before = recovery_family_witness(db_path)?;
let recovery_dir = recovery_artifacts_dir(db_path)?;
if let Some(prior) = prior_failed_recovery(&recovery_dir, &raw_before)? {
return Err(BeadsError::SyncConflict {
message: format!(
"automatic WAL index recovery already failed for this exact database family \
at stage {} ({}); its pre-recovery state is retained at {}. The database was \
not copied again. Run `br doctor migrate-schema recover` to retry explicitly, \
or `br doctor` for diagnosis.",
prior.stage,
prior.error.as_deref().unwrap_or("no recorded error"),
prior.backup_path
),
});
}
let migration = MigrationContext {
beads_dir: beads_dir.to_path_buf(),
db_path: db_path.to_path_buf(),
write_authority: Arc::clone(authority),
};
recover_engine_admission_with_lease(&migration, true, PreexistingIndexCorruption::Refuse)?;
Ok(())
}
#[derive(Debug, Deserialize)]
struct PriorRecoveryFailure {
backup_path: String,
stage: String,
raw_before: RawFamilyWitness,
error: Option<String>,
#[serde(flatten)]
details: serde_json::Map<String, serde_json::Value>,
}
impl PriorRecoveryFailure {
fn is_transient_contention(&self) -> bool {
if self.stage != "live-recovery"
|| self.details.get("schema_version").and_then(serde_json::Value::as_str)
!= Some("br.doctor.schema_migration.recovery.v1")
|| !matches!(
self.details.get("backup_scope").and_then(serde_json::Value::as_str),
Some(RECOVERY_BACKUP_WAL_INDEX_ONLY | RECOVERY_BACKUP_COMPLETE_FAMILY)
)
|| self.details.get("raw_after") != Some(&serde_json::Value::Null)
|| self.details.get("logical_after") != Some(&serde_json::Value::Null)
|| self.details.get("index_corruption").is_some_and(|value| !value.is_null())
{
return false;
}
match self.details.get("failure_kind") {
Some(kind) => kind.as_str() == Some("transient-contention"),
None => matches!(
self.error.as_deref(),
Some(
"Database error: database is busy"
| "Database error: database is busy (recovery in progress)"
)
),
}
}
}
fn prior_failed_recovery(
beads_dir: &Path,
raw_before: &RawFamilyWitness,
) -> Result<Option<PriorRecoveryFailure>> {
let root = migration_runs_root(beads_dir);
let entries = match fs::read_dir(&root) {
Ok(entries) => entries,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(error) => return Err(error.into()),
};
for entry in entries {
let entry = entry?;
if !entry.file_type()?.is_dir() {
continue;
}
let receipt_path = entry.path().join("recovery-failed.json");
let bytes = match fs::read(&receipt_path) {
Ok(bytes) => bytes,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
Err(error) => return Err(error.into()),
};
let Ok(prior) = serde_json::from_slice::<PriorRecoveryFailure>(&bytes) else {
continue;
};
if &prior.raw_before == raw_before && !prior.is_transient_contention() {
return Ok(Some(prior));
}
}
Ok(None)
}
fn recovery_artifacts_dir(db_path: &Path) -> Result<PathBuf> {
let parent = db_path
.parent()
.filter(|parent| !parent.as_os_str().is_empty())
.unwrap_or_else(|| Path::new("."));
Ok(fs::canonicalize(parent)?)
}
fn recover_engine_admission_with_lease(
migration: &MigrationContext,
strict_wal: bool,
policy: PreexistingIndexCorruption,
) -> Result<EngineRecoveryReceipt> {
if crate::sync::database_write_authority_sha256(&migration.db_path)?
!= migration.write_authority.authority_path_sha256()
{
return Err(BeadsError::SyncConflict {
message: "engine recovery path does not match the held database-family authority"
.to_string(),
});
}
migration.write_authority.verify_database_authority()?;
require_frame_free_engine_recovery(&migration.db_path)?;
let recovery = MigrationContext {
beads_dir: recovery_artifacts_dir(&migration.db_path)?,
db_path: migration.db_path.clone(),
write_authority: Arc::clone(&migration.write_authority),
};
let migration = &recovery;
let mut lease = crate::sync::DatabaseOpenerLease::register(&migration.db_path)?;
let exclusive = lease
.try_exclusive()
.ok_or_else(|| BeadsError::SyncConflict {
message:
"engine recovery requires a verified sole opener; close peer br processes and retry"
.to_string(),
})?;
let result = recover_engine_admission(migration, strict_wal, policy);
let released = lease.release_exclusive(exclusive);
result.and_then(|receipt| released.map(|()| receipt))
}
fn recovery_family_witness(path: &Path) -> Result<RawFamilyWitness> {
raw_family_witness_for_suffixes(
path,
std::iter::once("").chain(config::db_sidecar_suffixes().copied()),
)
}
fn require_recovery_payload_unchanged(
before: &RawFamilyWitness,
after: &RawFamilyWitness,
) -> Result<()> {
for suffix in ["", "-wal", "-journal"] {
if component_for_suffix(before, suffix)? != component_for_suffix(after, suffix)? {
return Err(BeadsError::internal(format!(
"engine admission recovery changed protected database component {suffix:?}; \
retained the complete pre-state and recovery evidence"
)));
}
}
Ok(())
}
fn require_private_recovery_payload_unchanged(
before: &RawFamilyWitness,
private: &RawFamilyWitness,
) -> Result<()> {
for suffix in ["", "-wal", "-journal"] {
let expected = component_for_suffix(before, suffix)?;
let observed = component_for_suffix(private, suffix)?;
if expected.present != observed.present
|| expected.length != observed.length
|| expected.sha256 != observed.sha256
{
return Err(BeadsError::internal(format!(
"private recovery changed protected database component {suffix:?}; \\
refusing index publication"
)));
}
}
Ok(())
}
fn require_frame_free_engine_recovery(path: &Path) -> Result<()> {
crate::franken_sync::wal_index::require_frame_free_recovery(path).map_err(|error| {
BeadsError::SyncConflict {
message: error.to_string(),
}
})
}
fn recover_existing_family(
path: &Path,
authority: Option<&DatabaseFamilyWriteLock>,
mut live_stage: Option<&mut String>,
) -> Result<()> {
require_frame_free_engine_recovery(path)?;
if let Some(authority) = authority {
return recover_live_index_from_private_copy(path, authority, live_stage);
}
let metadata = secure_file_metadata(path)?
.ok_or_else(|| BeadsError::internal("recovery database disappeared"))?;
let retained = File::open(path)?;
if !same_file_identity(&metadata, &retained.metadata()?) {
return Err(BeadsError::internal(
"recovery database changed before identity binding",
));
}
let identity = fsqlite_vfs::FileIdentity::from_file(&retained)?
.ok_or_else(|| BeadsError::internal("recovery database identity is unavailable"))?;
if let Some(authority) = authority {
authority.verify_database_authority()?;
}
if let Some(stage) = live_stage.as_deref_mut() {
*stage = "live-recovery".to_string();
}
let conn = Connection::open_existing_with_expected_identity(
path.to_string_lossy().into_owned(),
identity,
)?;
if let Some(stage) = live_stage {
*stage = "live-verification".to_string();
}
close_connection(conn)?;
if !same_file_identity(&retained.metadata()?, &fs::symlink_metadata(path)?) {
return Err(BeadsError::internal(
"recovery database identity changed during open",
));
}
Ok(())
}
fn recover_live_index_from_private_copy(
db_path: &Path,
authority: &DatabaseFamilyWriteLock,
mut live_stage: Option<&mut String>,
) -> Result<()> {
authority.verify_database_authority()?;
let raw_before = recovery_family_witness(db_path)?;
let scratch_root = recovery_artifacts_dir(db_path)?.join(".br_recovery");
fs::create_dir_all(&scratch_root)?;
let scratch = tempfile::Builder::new()
.prefix("index-rehearsal-")
.tempdir_in(&scratch_root)?;
let probe_db = backup_component_path(scratch.path(), db_path, "")?;
let operation = (|| -> Result<()> {
copy_family_to_backup(db_path, scratch.path(), &raw_before)?;
verify_backup_family(db_path, scratch.path(), &raw_before)?;
if let Some(stage) = live_stage.as_deref_mut() {
*stage = "private-recovery".to_string();
}
let probe_before = recovery_family_witness(&probe_db)?;
recover_existing_family(&probe_db, None, None)?;
require_recovery_payload_unchanged(&probe_before, &recovery_family_witness(&probe_db)?)?;
let _logical = logical_witness(&probe_db)?;
publish_recovered_index(
db_path,
&probe_db,
&raw_before,
authority,
live_stage.as_deref_mut(),
)
})();
if let Err(error) = operation {
if live_stage
.as_deref()
.is_some_and(|stage| stage == "live-recovery")
&& live_recovery_contention(&error)
{
return Err(error);
}
let evidence = scratch.keep();
return Err(BeadsError::WithContext {
context: format!("private index rehearsal retained at {}", evidence.display()),
source: Box::new(error),
});
}
Ok(())
}
fn publish_recovered_index(
db_path: &Path,
probe_db: &Path,
expected: &RawFamilyWitness,
authority: &DatabaseFamilyWriteLock,
live_stage: Option<&mut String>,
) -> Result<()> {
let probe_after = recovery_family_witness(probe_db)?;
require_private_recovery_payload_unchanged(expected, &probe_after)?;
if !crate::franken_sync::wal_index::ENGINE_READS_ON_DISK_WAL_INDEX {
authority.verify_database_authority()?;
require_frame_free_engine_recovery(db_path)?;
if &recovery_family_witness(db_path)? != expected {
return Err(BeadsError::internal(
"database family changed before recovery verification",
));
}
return Ok(());
}
let mut index = component_for_suffix(&probe_after, "-shm")?.clone();
if !index.present {
return Err(BeadsError::internal(
"private recovery produced no WAL index to install",
));
}
if let Some(mode) = component_for_suffix(expected, "-shm")?.unix_mode {
index.unix_mode = Some(mode);
}
install_recovery_index(
db_path,
&family_component_path(probe_db, "-shm"),
&index,
expected,
authority,
live_stage,
)
}
const NAMESPACE_COORDINATION_SUFFIXES: [&str; 2] = ["-fsqlite-ns-gate", "-fsqlite-ns-use"];
fn family_matches_outside_namespace_coordination(
expected: &RawFamilyWitness,
observed: &RawFamilyWitness,
) -> bool {
let content =
|component: &&RawComponentWitness| !is_namespace_coordination_suffix(&component.suffix);
expected
.components
.iter()
.filter(content)
.eq(observed.components.iter().filter(content))
}
fn is_namespace_coordination_suffix(suffix: &str) -> bool {
NAMESPACE_COORDINATION_SUFFIXES.contains(&suffix)
}
fn install_recovery_index(
db_path: &Path,
retained_index: &Path,
index: &RawComponentWitness,
expected: &RawFamilyWitness,
authority: &DatabaseFamilyWriteLock,
mut live_stage: Option<&mut String>,
) -> Result<()> {
require_frame_free_engine_recovery(db_path)?;
authority.verify_database_authority()?;
let metadata = secure_file_metadata(db_path)?
.ok_or_else(|| BeadsError::internal("recovery database disappeared"))?;
let retained = File::open(db_path)?;
if !same_file_identity(&metadata, &retained.metadata()?) {
return Err(BeadsError::internal(
"recovery database changed before identity binding",
));
}
let identity = fsqlite_vfs::FileIdentity::from_file(&retained)?
.ok_or_else(|| BeadsError::internal("recovery database identity is unavailable"))?;
if let Some(stage) = live_stage.as_deref_mut() {
*stage = "live-recovery".to_string();
}
crate::franken_sync::wal_index::with_recovery_exclusion(db_path, identity, || -> Result<()> {
if let Some(stage) = live_stage {
*stage = "live-index-install".to_string();
}
authority.verify_database_authority()?;
require_frame_free_engine_recovery(db_path)?;
if !family_matches_outside_namespace_coordination(
expected,
&recovery_family_witness(db_path)?,
) {
return Err(BeadsError::internal(
"database family changed before recovered-index installation; live family untouched",
));
}
verify_retained_component(retained_index, index)?;
install_retained_index(db_path, retained_index, index)?;
authority.verify_database_authority()?;
let after = recovery_family_witness(db_path)?;
for component in &expected.components {
if component.suffix != "-shm"
&& !is_namespace_coordination_suffix(&component.suffix)
&& component_for_suffix(&after, &component.suffix)? != component
{
return Err(BeadsError::internal(format!(
"index installation changed protected component {:?}; retain recovery evidence",
component.suffix
)));
}
}
verify_retained_component(&family_component_path(db_path, "-shm"), index)
})?
}
fn recover_engine_admission(
migration: &MigrationContext,
strict_wal: bool,
policy: PreexistingIndexCorruption,
) -> Result<EngineRecoveryReceipt> {
migration.write_authority.verify_database_authority()?;
refuse_non_regular_component(&migration.db_path)?;
let raw_before = recovery_family_witness(&migration.db_path)?;
let (receipt, run_dir) = if index_only_recovery_applies(&migration.db_path, &raw_before)? {
match recover_index_only(migration, &raw_before)? {
Some(completed) => completed,
None => recover_engine_admission_with_complete_backup(migration, strict_wal, policy)?,
}
} else {
recover_engine_admission_with_complete_backup(migration, strict_wal, policy)?
};
match prune_recovery_runs(
&migration.beads_dir,
&migration.db_path,
&run_dir,
Utc::now(),
) {
Ok(pruned) if !pruned.is_empty() => tracing::info!(
pruned = pruned.len(),
kept_recent = RECOVERY_RUNS_KEPT,
"pruned completed WAL-index recovery runs past the retention window"
),
Ok(_) => {}
Err(error) => tracing::warn!(
%error,
"could not prune old WAL-index recovery runs; they are kept"
),
}
match recovery_failure_receipts(&migration.beads_dir).and_then(|failures| {
crate::franken_sync::wal_index::prune_quarantined_indexes(
&migration.db_path,
RECOVERY_RUNS_KEPT,
std::time::Duration::from_secs(
u64::try_from(RECOVERY_RUN_MIN_AGE_DAYS).unwrap_or(7) * 24 * 60 * 60,
),
std::time::SystemTime::now(),
&failures,
)
.map_err(BeadsError::Io)
}) {
Ok(pruned) if !pruned.is_empty() => tracing::info!(
pruned = pruned.len(),
kept_recent = RECOVERY_RUNS_KEPT,
"pruned completed WAL-index quarantines past the retention window"
),
Ok(_) => {}
Err(error) => tracing::warn!(
%error,
"could not prune old WAL-index quarantines; they are kept"
),
}
Ok(receipt)
}
fn recovery_failure_receipts(beads_dir: &Path) -> Result<Vec<String>> {
let root = migration_runs_root(beads_dir);
let entries = match fs::read_dir(&root) {
Ok(entries) => entries,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
Err(error) => return Err(BeadsError::Io(error)),
};
let mut receipts = Vec::new();
for entry in entries {
let entry = entry.map_err(BeadsError::Io)?;
match fs::read(entry.path().join("recovery-failed.json")) {
Ok(bytes) => receipts.push(String::from_utf8_lossy(&bytes).into_owned()),
Err(error)
if matches!(
error.kind(),
std::io::ErrorKind::NotFound | std::io::ErrorKind::NotADirectory
) => {}
Err(error) => return Err(BeadsError::Io(error)),
}
}
Ok(receipts)
}
const RECOVERY_RUNS_KEPT: usize = 5;
const RECOVERY_RUN_MIN_AGE_DAYS: i64 = 7;
const RECOVERY_BACKUP_COMPLETE_FAMILY: &str = "complete-family";
const RECOVERY_BACKUP_WAL_INDEX_ONLY: &str = "wal-index-only";
const RECOVERY_SUPERSEDED_RECEIPT: &str = "recovery-superseded.json";
const PRUNING_PREFIX: &str = ".pruning-";
fn index_only_recovery_applies(db_path: &Path, raw_before: &RawFamilyWitness) -> Result<bool> {
let wal = component_for_suffix(raw_before, "-wal")?;
let shm = component_for_suffix(raw_before, "-shm")?;
let journal = component_for_suffix(raw_before, "-journal")?;
if !(wal.present && wal.length == Some(WAL_HEADER_BYTES) && shm.present && !journal.present) {
return Ok(false);
}
Ok(crate::franken_sync::wal_index::poisoned_index_present(
db_path,
)?)
}
fn recover_index_only(
migration: &MigrationContext,
raw_before: &RawFamilyWitness,
) -> Result<Option<(EngineRecoveryReceipt, PathBuf)>> {
let db_path = &migration.db_path;
let shm_witness = component_for_suffix(raw_before, "-shm")?.clone();
let run_id = allocate_run_id(&migration.beads_dir)?;
let run_dir = migration_runs_root(&migration.beads_dir).join(run_id);
let before_dir = run_dir.join("recovery-before");
ensure_new_directory(&before_dir)?;
let index_backup = backup_component_path(&before_dir, db_path, "-shm")?;
copy_regular_file_new(&family_component_path(db_path, "-shm"), &index_backup, None)?;
sync_directory(&before_dir)?;
verify_retained_component(&index_backup, &shm_witness)?;
let mut receipt = EngineRecoveryReceipt {
schema_version: "br.doctor.schema_migration.recovery.v1",
database_path: db_path.display().to_string(),
backup_path: before_dir.display().to_string(),
backup_scope: RECOVERY_BACKUP_WAL_INDEX_ONLY,
stage: "live-preflight".to_string(),
raw_before: raw_before.clone(),
raw_after: None,
logical_after: None,
index_corruption: None,
error: None,
failure_kind: None,
};
write_json_new(&run_dir.join("recovery-prepared.json"), &receipt)?;
let operation = (|| -> Result<bool> {
migration.write_authority.verify_database_authority()?;
if recovery_family_witness(db_path)? != receipt.raw_before {
return Err(BeadsError::internal(
"database family changed before WAL index recovery",
));
}
recover_existing_family(
db_path,
Some(&migration.write_authority),
Some(&mut receipt.stage),
)?;
migration.write_authority.verify_database_authority()?;
let raw_after = recovery_family_witness(db_path)?;
require_recovery_payload_unchanged(&receipt.raw_before, &raw_after)?;
receipt.raw_after = Some(raw_after);
let logical_after = logical_witness(db_path)?;
let clean = integrity_check_is_clean(&logical_after.integrity_check);
receipt.logical_after = Some(logical_after);
if clean {
receipt.stage = "complete".to_string();
return Ok(true);
}
receipt.stage = "reinstate-index".to_string();
let installed = receipt
.raw_after
.as_ref()
.ok_or_else(|| BeadsError::internal("recovery omitted the installed index witness"))?;
install_recovery_index(
db_path,
&index_backup,
&shm_witness,
installed,
&migration.write_authority,
None,
)?;
migration.write_authority.verify_database_authority()?;
let reinstated = recovery_family_witness(db_path)?;
for suffix in ["", "-wal", "-shm", "-journal"] {
if component_for_suffix(&reinstated, suffix)?
!= component_for_suffix(&receipt.raw_before, suffix)?
{
return Err(BeadsError::internal(format!(
"reinstating the original WAL index left component {suffix:?} different \
from the pre-recovery state"
)));
}
}
receipt.stage = "superseded".to_string();
Ok(false)
})();
match operation {
Ok(true) => {
write_json_new(&run_dir.join("recovery-complete.json"), &receipt)?;
Ok(Some((receipt, run_dir)))
}
Ok(false) => {
write_json_new(&run_dir.join(RECOVERY_SUPERSEDED_RECEIPT), &receipt)?;
Ok(None)
}
Err(error) => {
let failure_kind = RecoveryFailureKind::for_error(&receipt.stage, &error);
receipt.failure_kind = Some(failure_kind);
receipt.error = Some(error.to_string());
write_json_new(&run_dir.join("recovery-failed.json"), &receipt)?;
Err(BeadsError::WithContext {
context: format!(
"WAL index recovery failed at {}; the original index is retained at {}{}",
receipt.stage,
before_dir.display(),
failure_kind.retry_hint()
),
source: Box::new(error),
})
}
}
}
fn verify_retained_component(copy: &Path, witness: &RawComponentWitness) -> Result<()> {
let metadata = secure_file_metadata(copy)?.ok_or_else(|| {
BeadsError::internal(format!("retained copy is missing: {}", copy.display()))
})?;
let (length, sha256) = hash_regular_file(copy, &metadata)?;
if witness.length != Some(length) || witness.sha256.as_deref() != Some(sha256.as_str()) {
return Err(BeadsError::internal(format!(
"retained copy does not match the pre-recovery witness: {}",
copy.display()
)));
}
Ok(())
}
fn install_retained_index(
db_path: &Path,
retained: &Path,
witness: &RawComponentWitness,
) -> Result<()> {
let live = family_component_path(db_path, "-shm");
let parent = live
.parent()
.ok_or_else(|| BeadsError::internal("WAL index path has no parent directory"))?;
let nonce = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map_or(0, |elapsed| elapsed.as_nanos());
let staged = parent.join(format!(
".{}.reinstate-{}-{nonce}.tmp",
live.file_name()
.map_or_else(|| "shm".into(), |name| name.to_string_lossy()),
std::process::id()
));
copy_regular_file_new(retained, &staged, witness.unix_mode)?;
let installed = verify_retained_component(&staged, witness)
.and_then(|()| fs::rename(&staged, &live).map_err(BeadsError::Io));
if installed.is_err() {
let _ = fs::remove_file(&staged);
}
installed?;
sync_directory(parent)
}
#[allow(clippy::too_many_lines)]
fn prune_recovery_runs(
beads_dir: &Path,
db_path: &Path,
current_run: &Path,
now: chrono::DateTime<Utc>,
) -> Result<Vec<PathBuf>> {
if !cfg!(any(
target_os = "linux",
target_os = "android",
target_vendor = "apple",
windows
)) {
return Ok(Vec::new());
}
let root = migration_runs_root(beads_dir);
let entries = match fs::read_dir(&root) {
Ok(entries) => entries,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
Err(error) => return Err(BeadsError::Io(error)),
};
let Some(root_directory) = open_recovery_pruning_entry(&root, true)? else {
return Ok(Vec::new());
};
let database_path = db_path.display().to_string();
let min_age = chrono::Duration::days(RECOVERY_RUN_MIN_AGE_DAYS);
let mut runs = Vec::new();
let mut staged_runs = Vec::new();
for entry in entries {
let entry = entry.map_err(BeadsError::Io)?;
let file_type = entry.file_type().map_err(BeadsError::Io)?;
if !file_type.is_dir() || file_type.is_symlink() {
continue;
}
let name = entry.file_name().to_string_lossy().into_owned();
let path = entry.path();
if let Some(original_name) = name.strip_prefix(PRUNING_PREFIX) {
let Some(created) = recovery_run_timestamp(original_name) else {
continue;
};
let original = root.join(original_name);
if original == current_run || path == current_run || now - created < min_age {
continue;
}
staged_runs.push((created, path, original));
continue;
}
let Some(created) = recovery_run_timestamp(&name) else {
continue;
};
let Some(prepared) = read_recovery_pruning_receipt(&path.join("recovery-prepared.json"))?
else {
continue;
};
if prepared.database_path != database_path {
continue;
}
runs.push((created, path));
}
runs.sort_by(|left, right| right.0.cmp(&left.0).then_with(|| right.1.cmp(&left.1)));
for (created, path, original) in staged_runs {
if !runs
.get(RECOVERY_RUNS_KEPT - 1)
.is_some_and(|(kept_created, kept_path)| {
(kept_created, kept_path) > (&created, &original)
})
{
continue;
}
let Some(directory) = open_recovery_pruning_entry(&path, true)? else {
continue;
};
if !recovery_pruning_entry_unchanged(&root, &root_directory, true)? {
return Ok(Vec::new());
}
if let Err(error) = remove_verified_recovery_run(&path, &original, db_path, &directory) {
tracing::warn!(run = %path.display(), %error, "could not remove verified staged recovery run");
}
}
let mut pruned = Vec::new();
for (created, path) in runs.into_iter().skip(RECOVERY_RUNS_KEPT) {
if path == current_run || now - created < min_age {
continue;
}
let Some(name) = path.file_name() else {
continue;
};
let mut doomed_name = OsString::from(PRUNING_PREFIX);
doomed_name.push(name);
let doomed = root.join(doomed_name);
let Some(directory) = open_recovery_pruning_entry(&path, true)? else {
continue;
};
if !recovery_pruning_entry_unchanged(&root, &root_directory, true)?
|| recovery_run_status(&path, &path, db_path)? != Some(RecoveryRunStatus::Finished)
|| !recovery_pruning_entry_unchanged(&path, &directory, true)?
{
continue;
}
if let Err(error) = rename_path_no_replace(&path, &doomed) {
tracing::warn!(run = %path.display(), %error, "could not set aside old recovery run");
continue;
}
match remove_verified_recovery_run(&doomed, &path, db_path, &directory) {
Ok(true) => pruned.push(path),
Ok(false) => {
tracing::warn!(run = %doomed.display(), "recovery evidence changed; staged run is retained");
}
Err(error) => {
tracing::warn!(run = %doomed.display(), %error, "could not remove old recovery run");
}
}
}
Ok(pruned)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum RecoveryRunStatus {
Finished,
Retained,
}
fn recovery_run_timestamp(name: &str) -> Option<chrono::DateTime<Utc>> {
let (stamp, _) = name.split_once("Z-")?;
chrono::NaiveDateTime::parse_from_str(stamp, "%Y%m%dT%H%M%S%.f")
.ok()
.map(|naive| naive.and_utc())
}
fn recovery_run_status(
run_dir: &Path,
recorded_dir: &Path,
db_path: &Path,
) -> Result<Option<RecoveryRunStatus>> {
let Some(prepared) = read_recovery_pruning_receipt(&run_dir.join("recovery-prepared.json"))?
else {
return Ok(None);
};
if prepared.database_path != db_path.display().to_string() {
return Ok(None);
}
let retained = Some(RecoveryRunStatus::Retained);
match fs::symlink_metadata(run_dir.join("recovery-failed.json")) {
Ok(_) => return Ok(retained),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(error) => return Err(BeadsError::Io(error)),
}
let Some(completed) = read_recovery_pruning_receipt(&run_dir.join("recovery-complete.json"))?
else {
return Ok(retained);
};
if !recovery_pruning_receipts_agree(&prepared, &completed, recorded_dir) {
return Ok(retained);
}
let index_only = prepared.backup_scope == RECOVERY_BACKUP_WAL_INDEX_ONLY;
let directories: &[&str] = if index_only {
&["recovery-before"]
} else {
&["recovery-before", "recovery-probe"]
};
if !recovery_pruning_directory_contents(
run_dir,
&[
OsString::from("recovery-prepared.json"),
OsString::from("recovery-complete.json"),
],
directories,
)? || !recovery_pruning_backup_matches(
&run_dir.join("recovery-before"),
db_path,
&prepared.raw_before,
index_only,
)? {
return Ok(retained);
}
if !index_only
&& !recovery_pruning_probe_matches(
&run_dir.join("recovery-probe"),
recorded_dir,
db_path,
&prepared.raw_before,
)?
{
return Ok(retained);
}
Ok(Some(RecoveryRunStatus::Finished))
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct RecoveryPruningReceipt {
schema_version: String,
database_path: String,
backup_path: String,
backup_scope: String,
stage: String,
#[serde(deserialize_with = "deserialize_recovery_pruning_raw")]
raw_before: RawFamilyWitness,
#[serde(deserialize_with = "deserialize_optional_recovery_pruning_raw")]
raw_after: Option<RawFamilyWitness>,
#[serde(deserialize_with = "deserialize_optional_recovery_pruning_logical")]
logical_after: Option<LogicalDatabaseWitness>,
error: serde_json::Value,
#[serde(default, deserialize_with = "present_recovery_pruning_field")]
index_corruption: Option<serde_json::Value>,
#[serde(default, deserialize_with = "present_recovery_pruning_field")]
failure_kind: Option<serde_json::Value>,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct RecoveryPruningRaw {
components: Vec<RecoveryPruningComponent>,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct RecoveryPruningComponent {
suffix: String,
present: bool,
#[serde(deserialize_with = "required_recovery_pruning_option")]
length: Option<u64>,
#[serde(deserialize_with = "required_recovery_pruning_option")]
sha256: Option<String>,
#[serde(deserialize_with = "required_recovery_pruning_option")]
unix_mode: Option<u32>,
}
impl From<RecoveryPruningRaw> for RawFamilyWitness {
fn from(raw: RecoveryPruningRaw) -> Self {
Self {
components: raw
.components
.into_iter()
.map(|component| RawComponentWitness {
suffix: component.suffix,
present: component.present,
length: component.length,
sha256: component.sha256,
unix_mode: component.unix_mode,
})
.collect(),
}
}
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct RecoveryPruningLogical {
user_version: u32,
integrity_check: String,
schema_sha256: String,
contents_sha256: String,
tables: Vec<RecoveryPruningTable>,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct RecoveryPruningTable {
name: String,
row_count: u64,
rows_sha256: String,
}
impl From<RecoveryPruningLogical> for LogicalDatabaseWitness {
fn from(logical: RecoveryPruningLogical) -> Self {
Self {
user_version: logical.user_version,
integrity_check: logical.integrity_check,
schema_sha256: logical.schema_sha256,
contents_sha256: logical.contents_sha256,
tables: logical
.tables
.into_iter()
.map(|table| LogicalTableWitness {
name: table.name,
row_count: table.row_count,
rows_sha256: table.rows_sha256,
})
.collect(),
}
}
}
fn required_recovery_pruning_option<'de, D, T>(
deserializer: D,
) -> std::result::Result<Option<T>, D::Error>
where
D: serde::Deserializer<'de>,
T: Deserialize<'de>,
{
Option::<T>::deserialize(deserializer)
}
fn deserialize_recovery_pruning_raw<'de, D>(
deserializer: D,
) -> std::result::Result<RawFamilyWitness, D::Error>
where
D: serde::Deserializer<'de>,
{
RecoveryPruningRaw::deserialize(deserializer).map(Into::into)
}
fn deserialize_optional_recovery_pruning_raw<'de, D>(
deserializer: D,
) -> std::result::Result<Option<RawFamilyWitness>, D::Error>
where
D: serde::Deserializer<'de>,
{
Option::<RecoveryPruningRaw>::deserialize(deserializer).map(|raw| raw.map(Into::into))
}
fn deserialize_optional_recovery_pruning_logical<'de, D>(
deserializer: D,
) -> std::result::Result<Option<LogicalDatabaseWitness>, D::Error>
where
D: serde::Deserializer<'de>,
{
Option::<RecoveryPruningLogical>::deserialize(deserializer)
.map(|logical| logical.map(Into::into))
}
fn present_recovery_pruning_field<'de, D>(
deserializer: D,
) -> std::result::Result<Option<serde_json::Value>, D::Error>
where
D: serde::Deserializer<'de>,
{
serde_json::Value::deserialize(deserializer).map(Some)
}
fn recovery_pruning_receipts_agree(
prepared: &RecoveryPruningReceipt,
completed: &RecoveryPruningReceipt,
recorded_dir: &Path,
) -> bool {
let index_only = prepared.backup_scope == RECOVERY_BACKUP_WAL_INDEX_ONLY;
let expected_preparation = if index_only {
matches!(prepared.stage.as_str(), "live-preflight" | "live-recovery")
} else {
prepared.backup_scope == RECOVERY_BACKUP_COMPLETE_FAMILY
&& prepared.stage == "private-recovery"
};
if !expected_preparation
|| prepared.schema_version != "br.doctor.schema_migration.recovery.v1"
|| completed.schema_version != prepared.schema_version
|| completed.database_path != prepared.database_path
|| prepared.backup_path != recorded_dir.join("recovery-before").display().to_string()
|| completed.backup_path != prepared.backup_path
|| completed.backup_scope != prepared.backup_scope
|| completed.raw_before != prepared.raw_before
|| completed.stage != "complete"
|| prepared.raw_after.is_some()
|| prepared.logical_after.is_some()
|| !prepared.error.is_null()
|| !completed.error.is_null()
|| prepared.failure_kind.is_some()
|| completed.failure_kind.is_some()
|| prepared.index_corruption.is_some()
|| completed.index_corruption.is_some()
|| !recovery_pruning_raw_is_valid(&prepared.raw_before)
{
return false;
}
let Some(after) = &completed.raw_after else {
return false;
};
let Some(logical) = &completed.logical_after else {
return false;
};
if !recovery_pruning_raw_is_valid(after)
|| !integrity_check_is_clean(&logical.integrity_check)
|| !recovery_pruning_hash_is_valid(&logical.schema_sha256)
|| !recovery_pruning_hash_is_valid(&logical.contents_sha256)
|| logical
.tables
.iter()
.any(|table| !recovery_pruning_hash_is_valid(&table.rows_sha256))
|| require_recovery_payload_unchanged(&prepared.raw_before, after).is_err()
|| !prepared
.raw_before
.components
.iter()
.map(|component| &component.suffix)
.eq(after.components.iter().map(|component| &component.suffix))
{
return false;
}
!index_only
|| (component_for_suffix(&prepared.raw_before, "-wal")
.is_ok_and(|wal| wal.present && wal.length == Some(WAL_HEADER_BYTES))
&& component_for_suffix(&prepared.raw_before, "-shm").is_ok_and(|shm| shm.present)
&& component_for_suffix(&prepared.raw_before, "-journal")
.is_ok_and(|journal| !journal.present))
}
fn recovery_pruning_hash_is_valid(value: &str) -> bool {
value.len() == 64
&& value
.bytes()
.all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
}
fn recovery_pruning_raw_is_valid(witness: &RawFamilyWitness) -> bool {
let known: Vec<&str> = std::iter::once("")
.chain(config::db_sidecar_suffixes().copied())
.collect();
let mut seen = std::collections::HashSet::new();
witness.components.len() >= FAMILY_SUFFIXES.len()
&& witness.components.iter().all(|component| {
known.contains(&component.suffix.as_str())
&& seen.insert(component.suffix.as_str())
&& if component.present {
component.length.is_some()
&& component
.sha256
.as_deref()
.is_some_and(recovery_pruning_hash_is_valid)
} else {
component.length.is_none()
&& component.sha256.is_none()
&& component.unix_mode.is_none()
}
})
&& FAMILY_SUFFIXES.iter().all(|suffix| seen.contains(suffix))
&& component_for_suffix(witness, "").is_ok_and(|main| main.present)
}
fn recovery_pruning_identity(file: &File) -> Result<fsqlite_vfs::FileIdentity> {
fsqlite_vfs::FileIdentity::from_file(file)?
.ok_or_else(|| BeadsError::internal("recovery pruning cannot verify filesystem identity"))
}
fn open_recovery_pruning_entry(path: &Path, directory: bool) -> Result<Option<File>> {
let metadata = match fs::symlink_metadata(path) {
Ok(metadata) => metadata,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(error) => return Err(error.into()),
};
if if directory {
!metadata.is_dir()
} else {
!metadata.is_file()
} {
return Ok(None);
}
#[cfg(unix)]
{
use std::os::unix::fs::MetadataExt;
if !directory && metadata.nlink() != 1 {
return Ok(None);
}
}
let mut options = OpenOptions::new();
options.read(true);
#[cfg(unix)]
{
use std::os::unix::fs::OpenOptionsExt;
options.custom_flags(
libc::O_NOFOLLOW
| libc::O_CLOEXEC
| if directory {
libc::O_DIRECTORY
} else {
libc::O_NONBLOCK
},
);
}
#[cfg(windows)]
{
use std::os::windows::fs::OpenOptionsExt;
use windows_sys::Win32::Storage::FileSystem::{
FILE_FLAG_BACKUP_SEMANTICS, FILE_FLAG_OPEN_REPARSE_POINT,
};
options.custom_flags(
FILE_FLAG_OPEN_REPARSE_POINT
| if directory {
FILE_FLAG_BACKUP_SEMANTICS
} else {
0
},
);
}
let file = options.open(path)?;
let opened = file.metadata()?;
if if directory {
!opened.is_dir()
} else {
!opened.is_file()
} {
return Ok(None);
}
#[cfg(unix)]
{
use std::os::unix::fs::MetadataExt;
if !directory && opened.nlink() != 1 {
return Ok(None);
}
}
let _ = recovery_pruning_identity(&file)?;
Ok(Some(file))
}
fn recovery_pruning_entry_unchanged(path: &Path, retained: &File, directory: bool) -> Result<bool> {
let Some(current) = open_recovery_pruning_entry(path, directory)? else {
return Ok(false);
};
Ok(recovery_pruning_identity(¤t)? == recovery_pruning_identity(retained)?)
}
fn read_recovery_pruning_receipt(path: &Path) -> Result<Option<RecoveryPruningReceipt>> {
const LIMIT: u64 = 16 * 1024 * 1024;
let Some(mut file) = open_recovery_pruning_entry(path, false)? else {
return Ok(None);
};
if file.metadata()?.len() > LIMIT {
return Ok(None);
}
let mut bytes = Vec::new();
(&mut file).take(LIMIT + 1).read_to_end(&mut bytes)?;
if u64::try_from(bytes.len()).unwrap_or(u64::MAX) > LIMIT
|| !recovery_pruning_entry_unchanged(path, &file, false)?
{
return Ok(None);
}
Ok(serde_json::from_slice(&bytes).ok())
}
fn recovery_pruning_directory_contents(
dir: &Path,
files: &[OsString],
directories: &[&str],
) -> Result<bool> {
let Some(retained) = open_recovery_pruning_entry(dir, true)? else {
return Ok(false);
};
let mut found = 0;
for entry in fs::read_dir(dir)? {
let entry = entry?;
let kind = entry.file_type()?;
let name = entry.file_name();
if !((kind.is_file() && files.contains(&name))
|| (kind.is_dir() && directories.iter().any(|expected| name == *expected)))
{
return Ok(false);
}
found += 1;
}
Ok(found == files.len() + directories.len()
&& recovery_pruning_entry_unchanged(dir, &retained, true)?)
}
fn recovery_pruning_file_matches(path: &Path, component: &RawComponentWitness) -> Result<bool> {
let Some(mut file) = open_recovery_pruning_entry(path, false)? else {
return Ok(false);
};
if component.length != Some(file.metadata()?.len()) {
return Ok(false);
}
let mut digest = Sha256::new();
let mut buffer = vec![0_u8; 64 * 1024].into_boxed_slice();
loop {
let read = file.read(&mut buffer)?;
if read == 0 {
break;
}
digest.update(&buffer[..read]);
}
Ok(
component.sha256.as_deref() == Some(hex_digest(digest.finalize().as_slice()).as_str())
&& component.length == Some(file.metadata()?.len())
&& recovery_pruning_entry_unchanged(path, &file, false)?,
)
}
fn recovery_pruning_backup_matches(
dir: &Path,
db_path: &Path,
before: &RawFamilyWitness,
index_only: bool,
) -> Result<bool> {
let mut names = Vec::new();
for component in &before.components {
if !component.present || (index_only && component.suffix != "-shm") {
continue;
}
let path = backup_component_path(dir, db_path, &component.suffix)?;
let Some(name) = path.file_name() else {
return Ok(false);
};
names.push(name.to_os_string());
if !recovery_pruning_file_matches(&path, component)? {
return Ok(false);
}
}
recovery_pruning_directory_contents(dir, &names, &[])
}
fn recovery_pruning_probe_matches(
dir: &Path,
recorded_run: &Path,
db_path: &Path,
before: &RawFamilyWitness,
) -> Result<bool> {
let mut names = Vec::new();
for suffix in std::iter::once("").chain(config::db_sidecar_suffixes().copied()) {
let path = backup_component_path(dir, db_path, suffix)?;
let Some(file) = open_recovery_pruning_entry(&path, false)? else {
continue;
};
if let Some(name) = path.file_name() {
names.push(name.to_os_string());
}
drop(file);
}
let quarantine = match recovery_probe_directories(dir, recorded_run, db_path)? {
ProbeDirectories::Empty => None,
ProbeDirectories::FinishedQuarantine(name) => Some(name),
ProbeDirectories::Unrecognized => return Ok(false),
};
let directories: Vec<&str> = quarantine.as_deref().into_iter().collect();
if !recovery_pruning_directory_contents(dir, &names, &directories)? {
return Ok(false);
}
for suffix in ["", "-wal", "-journal"] {
let component = component_for_suffix(before, suffix)?;
let path = backup_component_path(dir, db_path, suffix)?;
if component.present {
if !recovery_pruning_file_matches(&path, component)? {
return Ok(false);
}
} else if fs::symlink_metadata(path).is_ok() {
return Ok(false);
}
}
Ok(true)
}
enum ProbeDirectories {
Empty,
FinishedQuarantine(String),
Unrecognized,
}
fn recovery_probe_directories(
dir: &Path,
recorded_run: &Path,
db_path: &Path,
) -> Result<ProbeDirectories> {
let entries = match fs::read_dir(dir) {
Ok(entries) => entries,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
return Ok(ProbeDirectories::Empty);
}
Err(error) => return Err(BeadsError::Io(error)),
};
let mut directories = Vec::new();
for entry in entries {
let entry = entry?;
if entry.file_type()?.is_dir() {
directories.push(entry.file_name());
}
}
let name = match directories.as_slice() {
[] => return Ok(ProbeDirectories::Empty),
[name] => match name.to_str() {
Some(name) => name.to_string(),
None => return Ok(ProbeDirectories::Unrecognized),
},
_ => return Ok(ProbeDirectories::Unrecognized),
};
let (Some(root), Some(run_name)) = (recorded_run.parent(), recorded_run.file_name()) else {
return Ok(ProbeDirectories::Unrecognized);
};
let Ok(root) = fs::canonicalize(root) else {
return Ok(ProbeDirectories::Unrecognized);
};
let recorded_probe = root.join(run_name).join("recovery-probe");
let probe_db = backup_component_path(dir, db_path, "")?;
let Some(db_name) = probe_db.file_name() else {
return Ok(ProbeDirectories::Unrecognized);
};
if crate::franken_sync::wal_index::is_finished_quarantine(
&dir.join(&name),
&recorded_probe.join(&name),
&recorded_probe.join(db_name).display().to_string(),
)? {
Ok(ProbeDirectories::FinishedQuarantine(name))
} else {
Ok(ProbeDirectories::Unrecognized)
}
}
fn remove_verified_recovery_run(
dir: &Path,
recorded_dir: &Path,
db_path: &Path,
retained: &File,
) -> Result<bool> {
if !recovery_pruning_entry_unchanged(dir, retained, true)?
|| recovery_run_status(dir, recorded_dir, db_path)? != Some(RecoveryRunStatus::Finished)
|| !recovery_pruning_entry_unchanged(dir, retained, true)?
{
return Ok(false);
}
let parent = dir
.parent()
.ok_or_else(|| BeadsError::internal("recovery run has no parent"))?;
sync_directory(parent)?;
fs::remove_dir_all(dir)?;
Ok(true)
}
fn recover_engine_admission_with_complete_backup(
migration: &MigrationContext,
strict_wal: bool,
policy: PreexistingIndexCorruption,
) -> Result<(EngineRecoveryReceipt, PathBuf)> {
migration.write_authority.verify_database_authority()?;
refuse_non_regular_component(&migration.db_path)?;
let raw_before = recovery_family_witness(&migration.db_path)?;
let run_id = allocate_run_id(&migration.beads_dir)?;
let run_dir = migration_runs_root(&migration.beads_dir).join(run_id);
let before_dir = run_dir.join("recovery-before");
ensure_new_directory(&before_dir)?;
copy_family_to_backup(&migration.db_path, &before_dir, &raw_before)?;
verify_backup_family(&migration.db_path, &before_dir, &raw_before)?;
let mut receipt = EngineRecoveryReceipt {
schema_version: "br.doctor.schema_migration.recovery.v1",
database_path: migration.db_path.display().to_string(),
backup_path: before_dir.display().to_string(),
backup_scope: RECOVERY_BACKUP_COMPLETE_FAMILY,
stage: "private-recovery".to_string(),
raw_before,
raw_after: None,
logical_after: None,
index_corruption: None,
error: None,
failure_kind: None,
};
write_json_new(&run_dir.join("recovery-prepared.json"), &receipt)?;
let operation = (|| -> Result<()> {
let probe_dir = run_dir.join("recovery-probe");
ensure_new_directory(&probe_dir)?;
let backup_db = backup_component_path(&before_dir, &migration.db_path, "")?;
copy_family_to_backup(&backup_db, &probe_dir, &receipt.raw_before)?;
verify_backup_family(&backup_db, &probe_dir, &receipt.raw_before)?;
let probe_db = backup_component_path(&probe_dir, &migration.db_path, "")?;
let probe_before = recovery_family_witness(&probe_db)?;
if strict_wal {
crate::storage::sqlite::validate_wal_for_index_recovery(&probe_db)?;
}
recover_existing_family(&probe_db, None, None)?;
require_recovery_payload_unchanged(&probe_before, &recovery_family_witness(&probe_db)?)?;
let expected = logical_witness(&probe_db)?;
if !integrity_check_is_clean(&expected.integrity_check) {
receipt.index_corruption = Some(admit_preexisting_index_damage(
&probe_db,
&run_dir.join("index-rehearsal"),
&expected,
policy,
)?);
}
receipt.stage = "live-preflight".to_string();
migration.write_authority.verify_database_authority()?;
if recovery_family_witness(&migration.db_path)? != receipt.raw_before {
return Err(BeadsError::internal(
"database family changed during recovery rehearsal",
));
}
verify_backup_family(&migration.db_path, &before_dir, &receipt.raw_before)?;
publish_recovered_index(
&migration.db_path,
&probe_db,
&receipt.raw_before,
&migration.write_authority,
Some(&mut receipt.stage),
)?;
migration.write_authority.verify_database_authority()?;
let raw_after = recovery_family_witness(&migration.db_path)?;
require_recovery_payload_unchanged(&receipt.raw_before, &raw_after)?;
receipt.raw_after = Some(raw_after);
receipt.logical_after = Some(logical_witness(&migration.db_path)?);
if receipt.logical_after.as_ref() != Some(&expected) {
return Err(BeadsError::internal(
"live engine recovery differs from the private witness",
));
}
receipt.stage = "complete".to_string();
Ok(())
})();
if let Err(error) = operation {
let failure_kind = RecoveryFailureKind::for_error(&receipt.stage, &error);
receipt.failure_kind = Some(failure_kind);
receipt.error = Some(error.to_string());
write_json_new(&run_dir.join("recovery-failed.json"), &receipt)?;
return Err(BeadsError::WithContext {
context: format!(
"engine recovery failed at {}; complete pre-state retained at {}{}",
receipt.stage,
before_dir.display(),
failure_kind.retry_hint()
),
source: Box::new(error),
});
}
write_json_new(&run_dir.join("recovery-complete.json"), &receipt)?;
Ok((receipt, run_dir))
}
#[allow(clippy::too_many_lines)]
fn build_plan(db_path: &Path) -> Result<MigrationPlanReceipt> {
refuse_non_regular_component(db_path)?;
let logical_witness = logical_witness(db_path)?;
let raw_witness = raw_family_witness(db_path)?;
let target = current_schema_version()?;
let from = logical_witness.user_version;
let database_path = db_path.display().to_string();
let integrity_clean = integrity_check_is_clean(&logical_witness.integrity_check);
if from == target {
if !current_runtime_shape_is_canonical(db_path)? {
return Err(BeadsError::internal(
"database declares the current schema version but its runtime shape is not \
canonical; refusing to issue a migration no-op receipt",
));
}
if !integrity_clean {
if !integrity_check_is_repairable(&logical_witness.integrity_check) {
return Err(BeadsError::internal(format!(
"database declares the current schema version but its integrity diagnostics \
are not eligible for reviewed page-layout maintenance: {:?}",
logical_witness.integrity_check
)));
}
let forecast = MigrationForecast {
from_version: from,
to_version: target,
content_hash_rows_rebuilt: 0,
gate_result_history_created: false,
prerequisites_column_added: false,
dependency_type_key_added: false,
legacy_v15_conversion: None,
post_migration_maintenance: true,
};
let plan_token = compute_plan_token(&database_path, &logical_witness, &forecast)?;
return Ok(MigrationPlanReceipt {
schema_version: PLAN_SCHEMA.to_string(),
eligible: true,
database_path,
from_version: from,
to_version: target,
raw_witness,
logical_witness,
forecast: Some(forecast),
apply_command: Some(format!(
"br doctor migrate-schema apply --plan-token {plan_token}"
)),
plan_token: Some(plan_token),
note: "the current schema has a repairable page-layout diagnostic; apply will \
preserve a complete recovery bundle, checkpoint, rebuild indexes, rewrite \
the database pages, and require a clean fresh-connection integrity check"
.to_string(),
});
}
return Ok(MigrationPlanReceipt {
schema_version: PLAN_SCHEMA.to_string(),
eligible: false,
database_path,
from_version: from,
to_version: target,
raw_witness,
logical_witness,
forecast: None,
plan_token: None,
apply_command: None,
note: "database already has the current canonical schema; no migration is needed"
.to_string(),
});
}
if !REVIEWED_MIGRATION_SOURCE_VERSIONS.contains(&from) {
return Err(BeadsError::internal(format!(
"reviewed schema migration is available only from source schemas 13, 14, 15, 16, 17, and 18 \
to {target}; observed unsupported source version {from}. Run \
`br doctor migrate-schema heal` instead: it rebuilds this database from issues.jsonl, \
keeps database-only issues, and retains the old database as a backup"
)));
}
if !integrity_clean && !integrity_check_is_repairable(&logical_witness.integrity_check) {
return Err(BeadsError::internal(format!(
"schema migration refused because PRAGMA integrity_check returned diagnostics that \
are not eligible for reviewed page-layout maintenance: {:?}",
logical_witness.integrity_check
)));
}
let conn = open_read_only(db_path)?;
require_source_tables(&conn, from)?;
let legacy_v15 = attest_reviewed_migration_source_core_tables(&conn)?;
let legacy_v15_conversion = legacy_v15
.then(|| legacy_v15_projection(&conn, true))
.transpose()?;
let issue_count = query_count(&conn, "SELECT COUNT(*) FROM issues")?;
let gate_result_history_created = !named_table_exists(&conn, "gate_result_history")?;
close_connection(conn)?;
let forecast = MigrationForecast {
from_version: from,
to_version: target,
content_hash_rows_rebuilt: if from == 13 {
usize::try_from(issue_count).map_err(|_| {
BeadsError::internal(format!(
"issue count {issue_count} cannot be represented on this platform"
))
})?
} else {
0
},
gate_result_history_created,
prerequisites_column_added: from < 18,
dependency_type_key_added: !legacy_v15,
legacy_v15_conversion,
post_migration_maintenance: true,
};
let plan_token = compute_plan_token(&database_path, &logical_witness, &forecast)?;
Ok(MigrationPlanReceipt {
schema_version: PLAN_SCHEMA.to_string(),
eligible: true,
database_path,
from_version: from,
to_version: target,
raw_witness,
logical_witness,
forecast: Some(forecast),
apply_command: Some(format!(
"br doctor migrate-schema apply --plan-token {plan_token}"
)),
plan_token: Some(plan_token),
note: if integrity_clean {
"review the forecast and retain this receipt; apply will recompute the complete \
logical witness and refuse semantic drift, then back up the current raw SQLite \
family before migration and mandatory post-migration page maintenance"
.to_string()
} else {
"the supported source schema has a repairable page-layout diagnostic; review the \
forecast and retain this receipt. Apply will recompute the complete logical witness, \
back up the current raw SQLite family, migrate it, run mandatory page-layout \
maintenance, and require a clean fresh-connection integrity check"
.to_string()
},
})
}
fn emit_plan(plan: &MigrationPlanReceipt, json: bool) -> Result<()> {
if json {
println!(
"{}",
serde_json::to_string_pretty(plan).map_err(BeadsError::Json)?
);
return Ok(());
}
if plan.eligible {
println!(
"Reviewed schema migration: {} -> {}",
plan.from_version, plan.to_version
);
println!(
"Plan token: {}",
plan.plan_token.as_deref().unwrap_or_default()
);
println!(
"Apply: {}",
plan.apply_command.as_deref().unwrap_or_default()
);
} else {
println!("{}", plan.note);
}
Ok(())
}
fn execute_apply(args: &DoctorMigrateSchemaApplyArgs, migration: &MigrationContext) -> Result<()> {
let (applied, before_dir) = apply_with_plan_token(args.plan_token.trim(), migration)?;
emit_applied(&applied, args.json, &before_dir)
}
pub(super) fn plan_and_apply(migration: &MigrationContext) -> Result<PathBuf> {
let plan = build_plan(&migration.db_path)?;
let token = plan.plan_token.ok_or_else(|| {
BeadsError::internal(format!(
"schema migration is not eligible for this database: {}",
plan.note
))
})?;
apply_with_plan_token(&token, migration).map(|(_, before_dir)| before_dir)
}
#[allow(clippy::too_many_lines)]
fn apply_with_plan_token(
plan_token: &str,
migration: &MigrationContext,
) -> Result<(AppliedMigrationReceipt, PathBuf)> {
if plan_token.is_empty() {
return Err(BeadsError::internal(
"schema migration apply requires a non-empty --plan-token",
));
}
#[cfg(any(target_os = "linux", target_os = "android", target_vendor = "apple"))]
verify_migration_exchange_support(&migration.db_path, exchange_database_paths)?;
if let Some(resumed) = resume_commit_ready_migration(plan_token, migration)? {
return Ok(resumed);
}
let plan = build_plan(&migration.db_path)?;
let Some(recomputed_token) = plan.plan_token.as_deref() else {
return Err(BeadsError::internal(
"schema migration apply refused because no migration is currently eligible",
));
};
if !constant_time_text_eq(recomputed_token, plan_token) {
return Err(BeadsError::internal(format!(
"schema migration plan token is stale or belongs to a different database state \
(provided {plan_token}, recomputed {recomputed_token}); run \
`br doctor migrate-schema plan` again"
)));
}
let forecast = plan
.forecast
.clone()
.ok_or_else(|| BeadsError::internal("eligible migration plan omitted its forecast"))?;
let run_id = allocate_run_id(&migration.beads_dir)?;
let run_dir = migration_runs_root(&migration.beads_dir).join(&run_id);
let before_dir = run_dir.join("before");
ensure_new_directory(&before_dir)?;
copy_family_to_backup(&migration.db_path, &before_dir, &plan.raw_witness)?;
verify_backup_family(&migration.db_path, &before_dir, &plan.raw_witness)?;
let marked_at = Utc::now().to_rfc3339();
let prepared = PreparedMigrationReceipt {
schema_version: PREPARED_SCHEMA.to_string(),
run_id: run_id.clone(),
database_path: plan.database_path.clone(),
plan_token: recomputed_token.to_string(),
marked_at: marked_at.clone(),
forecast: forecast.clone(),
raw_before: plan.raw_witness.clone(),
logical_before: plan.logical_witness.clone(),
};
write_json_new(&run_dir.join("prepared.json"), &prepared)?;
let prepared_receipt_sha256 = file_sha256(&run_dir.join("prepared.json"))?;
sync_directory(&run_dir)?;
migration.write_authority.verify_database_authority()?;
let mut failed_stage: Option<String> = None;
let migration_result = apply_reviewed_migration(
&migration.db_path,
&forecast,
&marked_at,
&run_dir,
&plan.raw_witness,
&migration.write_authority,
&mut failed_stage,
);
let authority_result = migration.write_authority.verify_database_authority();
let authority_verified_after = authority_result.is_ok();
let migration_result = match (migration_result, authority_result) {
(Ok(effects), Ok(())) => Ok(effects),
(Err(error), Ok(())) => Err(error),
(Ok(_), Err(authority_error)) => {
mark_stage(&mut failed_stage, "post-apply-authority-verification");
Err(authority_error)
}
(Err(error), Err(authority_error)) => Err(BeadsError::WithContext {
context: format!(
"reviewed schema migration failed and its database authority could not be \
re-verified afterward ({authority_error})"
),
source: Box::new(error),
}),
};
let effects = match migration_result {
Ok(effects) => effects,
Err(error) => {
let failed = FailedMigrationReceipt {
schema_version: FAILED_SCHEMA.to_string(),
run_id: run_id.clone(),
database_path: plan.database_path,
plan_token: recomputed_token.to_string(),
marked_at,
error: error.to_string(),
failed_stage,
raw_before: plan.raw_witness,
logical_before: plan.logical_witness,
raw_observed_after_failure: raw_family_witness(&migration.db_path).ok(),
logical_observed_after_failure: authority_verified_after
.then(|| logical_witness(&migration.db_path).ok())
.flatten(),
};
write_json_new(&run_dir.join("failed.json"), &failed)?;
return Err(BeadsError::WithContext {
context: format!(
"reviewed schema migration run {run_id} failed; the verified pre-state \
remains at {}",
before_dir.display()
),
source: Box::new(error),
});
}
};
let mut attestation_errors = Vec::new();
let maintenance_matches_forecast =
effects.post_migration_maintenance_completed == forecast.post_migration_maintenance;
if effects.from_version != forecast.from_version
|| effects.to_version != forecast.to_version
|| effects.content_hash_rows_rebuilt != forecast.content_hash_rows_rebuilt
|| effects.gate_result_history_created != forecast.gate_result_history_created
|| effects.prerequisites_column_added != forecast.prerequisites_column_added
|| effects.dependency_type_key_added != forecast.dependency_type_key_added
|| effects.legacy_v15_tables_converted != forecast.legacy_v15_conversion.is_some()
|| !maintenance_matches_forecast
{
attestation_errors.push(format!(
"committed effects differ from the reviewed forecast \
(forecast={forecast:?}, effects={effects:?})"
));
}
let logical_after = match logical_witness(&migration.db_path) {
Ok(witness) => Some(witness),
Err(error) => {
attestation_errors.push(format!(
"could not capture the committed logical witness: {error}"
));
None
}
};
let raw_after = match raw_family_witness(&migration.db_path) {
Ok(witness) => Some(witness),
Err(error) => {
attestation_errors.push(format!(
"could not capture the committed raw witness: {error}"
));
None
}
};
if let Some(logical_after) = logical_after.as_ref()
&& (logical_after.user_version != forecast.to_version
|| !integrity_check_is_clean(&logical_after.integrity_check))
{
attestation_errors.push(format!(
"committed logical witness did not attest target version {} and integrity=ok",
forecast.to_version
));
}
match current_runtime_shape_is_canonical(&migration.db_path) {
Ok(true) => {}
Ok(false) => {
attestation_errors.push(
"committed database does not have the canonical current runtime shape".to_string(),
);
}
Err(error) => {
attestation_errors.push(format!(
"could not attest the committed canonical runtime shape: {error}"
));
}
}
let attested = attestation_errors.is_empty();
let applied = AppliedMigrationReceipt {
schema_version: APPLIED_SCHEMA.to_string(),
run_id: run_id.clone(),
database_path: plan.database_path,
plan_token: recomputed_token.to_string(),
prepared_receipt_sha256,
marked_at,
forecast,
effects,
raw_before: plan.raw_witness,
logical_before: plan.logical_witness,
raw_after,
logical_after,
attested,
attestation_errors,
undo_command: format!("br doctor migrate-schema undo {run_id}"),
};
let commit_ready: CommitReadyMigrationReceipt = read_json(&run_dir.join("commit-ready.json"))?;
validate_applied_against_commit_ready(&applied, &commit_ready, &run_dir)?;
write_json_new(&run_dir.join("applied.json"), &applied)?;
sync_directory(&run_dir)?;
if !applied.attested {
return Err(BeadsError::internal(format!(
"schema migration run {run_id} committed but failed post-commit attestation: {}; \
an undo-capable applied receipt was persisted at {}; run `{}` before further \
tracker writes",
applied.attestation_errors.join("; "),
run_dir.join("applied.json").display(),
applied.undo_command
)));
}
Ok((applied, before_dir))
}
fn emit_applied(applied: &AppliedMigrationReceipt, json: bool, before_dir: &Path) -> Result<()> {
if json {
println!(
"{}",
serde_json::to_string_pretty(&applied).map_err(BeadsError::Json)?
);
} else {
println!(
"Applied reviewed schema migration {} -> {} (run {})",
applied.forecast.from_version, applied.forecast.to_version, applied.run_id
);
println!("Undo: {}", applied.undo_command);
println!("Recovery bundle: {}", before_dir.display());
}
Ok(())
}
fn persist_commit_ready_marker(
run_dir: &Path,
db_path: &Path,
logical_after: &LogicalDatabaseWitness,
effects: ReviewedSchemaMigrationEffectsReceipt,
) -> Result<()> {
let prepared_path = run_dir.join("prepared.json");
let prepared_receipt_sha256 = file_sha256(&prepared_path)?;
let prepared: PreparedMigrationReceipt = read_json(&prepared_path)?;
let run_id = run_dir
.file_name()
.and_then(|value| value.to_str())
.ok_or_else(|| BeadsError::internal("schema migration run directory has no UTF-8 name"))?;
if prepared.run_id != run_id || prepared.database_path != db_path.display().to_string() {
return Err(BeadsError::internal(
"prepared schema-migration receipt does not name its commit-ready run and database",
));
}
let marker = CommitReadyMigrationReceipt {
schema_version: COMMIT_READY_SCHEMA.to_string(),
run_id: prepared.run_id.clone(),
database_path: prepared.database_path.clone(),
plan_token: prepared.plan_token.clone(),
prepared_receipt_sha256,
marked_at: prepared.marked_at.clone(),
forecast: prepared.forecast.clone(),
effects,
raw_before: prepared.raw_before.clone(),
logical_before: prepared.logical_before.clone(),
logical_after: logical_after.clone(),
};
validate_commit_ready_marker(&marker, run_dir)?;
let marker_path = run_dir.join("commit-ready.json");
write_json_new(&marker_path, &marker)?;
let persisted: CommitReadyMigrationReceipt = read_json(&marker_path)?;
validate_commit_ready_marker(&persisted, run_dir)?;
if persisted != marker {
return Err(BeadsError::internal(
"persisted schema-migration commit-ready marker changed during read-back",
));
}
Ok(())
}
fn validate_commit_ready_marker(
marker: &CommitReadyMigrationReceipt,
run_dir: &Path,
) -> Result<()> {
validate_raw_family_witness(&marker.raw_before)?;
let run_id = run_dir
.file_name()
.and_then(|value| value.to_str())
.ok_or_else(|| BeadsError::internal("schema migration run directory has no UTF-8 name"))?;
if marker.schema_version != COMMIT_READY_SCHEMA || marker.run_id != run_id {
return Err(BeadsError::internal(format!(
"invalid schema-migration commit-ready marker in {}",
run_dir.display()
)));
}
let prepared_path = run_dir.join("prepared.json");
let prepared_sha256 = file_sha256(&prepared_path)?;
let prepared: PreparedMigrationReceipt = read_json(&prepared_path)?;
if prepared.schema_version != PREPARED_SCHEMA
|| !constant_time_text_eq(&prepared_sha256, &marker.prepared_receipt_sha256)
|| prepared.run_id != marker.run_id
|| prepared.database_path != marker.database_path
|| !constant_time_text_eq(&prepared.plan_token, &marker.plan_token)
|| prepared.marked_at != marker.marked_at
|| prepared.forecast != marker.forecast
|| prepared.raw_before != marker.raw_before
|| prepared.logical_before != marker.logical_before
{
return Err(BeadsError::internal(format!(
"schema-migration commit-ready marker in {} is not bound to its prepared receipt",
run_dir.display()
)));
}
let recomputed_token = compute_plan_token(
&marker.database_path,
&marker.logical_before,
&marker.forecast,
)?;
let maintenance_completion_disagrees_with_forecast =
marker.effects.post_migration_maintenance_completed
!= marker.forecast.post_migration_maintenance;
if !constant_time_text_eq(&recomputed_token, &marker.plan_token)
|| marker.effects.from_version != marker.forecast.from_version
|| marker.effects.to_version != marker.forecast.to_version
|| marker.effects.content_hash_rows_rebuilt != marker.forecast.content_hash_rows_rebuilt
|| marker.effects.gate_result_history_created != marker.forecast.gate_result_history_created
|| marker.effects.prerequisites_column_added != marker.forecast.prerequisites_column_added
|| marker.effects.dependency_type_key_added != marker.forecast.dependency_type_key_added
|| marker.effects.legacy_v15_tables_converted
!= marker.forecast.legacy_v15_conversion.is_some()
|| maintenance_completion_disagrees_with_forecast
|| marker.logical_after.user_version != marker.forecast.to_version
|| !integrity_check_is_clean(&marker.logical_after.integrity_check)
{
return Err(BeadsError::internal(format!(
"schema-migration commit-ready marker in {} has inconsistent effects or witnesses",
run_dir.display()
)));
}
Ok(())
}
#[allow(clippy::too_many_lines)]
fn resume_commit_ready_migration(
plan_token: &str,
migration: &MigrationContext,
) -> Result<Option<(AppliedMigrationReceipt, PathBuf)>> {
let root = migration_runs_root(&migration.beads_dir);
let entries = match fs::read_dir(&root) {
Ok(entries) => entries,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(error) => return Err(BeadsError::Io(error)),
};
let database_path = migration.db_path.display().to_string();
let mut matched = Vec::new();
for entry in entries {
let entry = entry.map_err(BeadsError::Io)?;
let file_type = entry.file_type().map_err(BeadsError::Io)?;
if !file_type.is_dir() || file_type.is_symlink() {
continue;
}
let run_dir = entry.path();
let marker_path = run_dir.join("commit-ready.json");
if secure_file_metadata(&marker_path)?.is_none() {
continue;
}
let marker: CommitReadyMigrationReceipt = read_json(&marker_path)?;
validate_commit_ready_marker(&marker, &run_dir)?;
if marker.database_path != database_path
|| !constant_time_text_eq(&marker.plan_token, plan_token)
{
continue;
}
matched.push((run_dir, marker));
}
matched.sort_by(|left, right| left.1.run_id.cmp(&right.1.run_id));
let Some((run_dir, marker)) = matched.pop() else {
return Ok(None);
};
let before_dir = run_dir.join("before");
verify_backup_family(&migration.db_path, &before_dir, &marker.raw_before)?;
let applied_path = run_dir.join("applied.json");
if secure_file_metadata(&applied_path)?.is_some() {
let applied: AppliedMigrationReceipt = read_json(&applied_path)?;
validate_applied_against_commit_ready(&applied, &marker, &run_dir)?;
let logical_live = logical_witness(&migration.db_path).ok();
let projection_unchanged = verify_legacy_v15_projection(
&migration.db_path,
applied.forecast.legacy_v15_conversion.as_ref(),
false,
)
.is_ok();
let raw_live = raw_family_witness(&migration.db_path)?;
if require_unchanged_applied_state(
&applied,
&raw_live,
logical_live.as_ref(),
&applied.run_id,
projection_unchanged,
)
.is_ok()
{
return Err(BeadsError::internal(format!(
"schema migration plan token was already consumed by completed run {}; \
run `br doctor migrate-schema plan` again for the current database state",
applied.run_id
)));
}
return Ok(None);
}
migration.write_authority.verify_database_authority()?;
let raw_live_before_logical_probe = raw_family_witness(&migration.db_path)?;
let original_main = component_for_suffix(&marker.raw_before, "")?;
let live_main = component_for_suffix(&raw_live_before_logical_probe, "")?;
let main_still_matches_original = live_main.present == original_main.present
&& live_main.length == original_main.length
&& live_main.sha256 == original_main.sha256;
if raw_live_before_logical_probe == marker.raw_before || main_still_matches_original {
let raw_live = restore_interrupted_preinstall_family(
&migration.db_path,
&before_dir,
&run_dir.join("maintenance-displaced"),
&marker.raw_before,
&migration.write_authority,
)?;
let logical_live = logical_witness(&migration.db_path)?;
migration.write_authority.verify_database_authority()?;
let raw_live_after_probe = raw_family_witness(&migration.db_path)?;
if logical_live != marker.logical_before
|| !stable_raw_eq(&raw_live_after_probe, &marker.raw_before)
|| raw_live != marker.raw_before
{
return Err(BeadsError::internal(format!(
"commit-ready schema migration {} could not prove the exact original family \
after reconciling an interrupted pre-install state; retained recovery \
artifacts at {}",
marker.run_id,
run_dir.display()
)));
}
let failed = FailedMigrationReceipt {
schema_version: FAILED_SCHEMA.to_string(),
run_id: marker.run_id.clone(),
database_path: marker.database_path.clone(),
plan_token: marker.plan_token.clone(),
marked_at: marker.marked_at.clone(),
error: "commit-ready migration intent was interrupted before the replacement became live; the original logical generation remains authoritative"
.to_string(),
failed_stage: None,
raw_before: marker.raw_before.clone(),
logical_before: marker.logical_before.clone(),
raw_observed_after_failure: Some(raw_live_after_probe),
logical_observed_after_failure: Some(logical_live),
};
let failed_path = run_dir.join("failed.json");
if secure_file_metadata(&failed_path)?.is_some() {
let existing: FailedMigrationReceipt = read_json(&failed_path)?;
validate_failed_against_commit_ready(&existing, &marker, &run_dir)?;
} else {
write_json_new(&failed_path, &failed)?;
}
return Ok(None);
}
let logical_live = logical_witness(&migration.db_path)?;
if logical_live != marker.logical_after {
return Err(BeadsError::internal(format!(
"commit-ready schema migration {} cannot be resumed because the live database is \
an unrecognized generation; retained all recovery artifacts at {}",
marker.run_id,
run_dir.display()
)));
}
verify_legacy_v15_projection(
&migration.db_path,
marker.forecast.legacy_v15_conversion.as_ref(),
false,
)?;
if !current_runtime_shape_is_canonical(&migration.db_path)? {
return Err(BeadsError::internal(format!(
"commit-ready schema migration {} matches logically but not the canonical runtime shape",
marker.run_id
)));
}
let raw_live = raw_family_witness(&migration.db_path)?;
let retained_original_guard = certify_resumed_installed_generation(
&migration.db_path,
&before_dir,
&run_dir,
&marker,
&migration.write_authority,
)?;
migration.write_authority.finalize_database_replacement()?;
let applied = AppliedMigrationReceipt {
schema_version: APPLIED_SCHEMA.to_string(),
run_id: marker.run_id.clone(),
database_path: marker.database_path.clone(),
plan_token: marker.plan_token.clone(),
prepared_receipt_sha256: marker.prepared_receipt_sha256.clone(),
marked_at: marker.marked_at.clone(),
forecast: marker.forecast.clone(),
effects: marker.effects,
raw_before: marker.raw_before.clone(),
logical_before: marker.logical_before.clone(),
raw_after: Some(raw_live),
logical_after: Some(logical_live),
attested: true,
attestation_errors: Vec::new(),
undo_command: format!("br doctor migrate-schema undo {}", marker.run_id),
};
validate_applied_against_commit_ready(&applied, &marker, &run_dir)?;
write_json_new(&applied_path, &applied)?;
sync_directory(&run_dir)?;
drop(retained_original_guard);
Ok(Some((applied, before_dir)))
}
fn certify_resumed_installed_generation(
db_path: &Path,
before_dir: &Path,
run_dir: &Path,
marker: &CommitReadyMigrationReceipt,
write_authority: &Arc<DatabaseFamilyWriteLock>,
) -> Result<File> {
verify_backup_family(db_path, before_dir, &marker.raw_before)?;
write_authority.verify_database_authority()?;
let displaced_dir = run_dir.join("maintenance-displaced");
let displaced_metadata = fs::symlink_metadata(&displaced_dir).map_err(BeadsError::Io)?;
if !displaced_metadata.is_dir() || displaced_metadata.file_type().is_symlink() {
return Err(BeadsError::internal(format!(
"schema-migration displaced path is not a real directory: {}",
displaced_dir.display()
)));
}
let displaced_main = backup_component_path(&displaced_dir, db_path, "")?;
let candidate_path = maintenance_candidate_path(db_path, run_dir)?;
let displaced_exists = secure_file_metadata(&displaced_main)?.is_some();
let candidate_exists = secure_file_metadata(&candidate_path)?.is_some();
if displaced_exists && candidate_exists {
return Err(BeadsError::internal(format!(
"resumed schema migration {} found two possible retained original mains; refusing to choose between {} and {}",
marker.run_id,
displaced_main.display(),
candidate_path.display()
)));
}
let guard = if displaced_exists {
write_authority.lock_database_replacement_candidate(&displaced_main)?
} else if candidate_exists {
let guard = write_authority.lock_database_replacement_candidate(&candidate_path)?;
write_authority.verify_locked_database_replacement_candidate(&candidate_path, &guard)?;
rename_path_no_replace(&candidate_path, &displaced_main)?;
write_authority.verify_locked_database_replacement_candidate(&displaced_main, &guard)?;
guard
} else {
return Err(BeadsError::internal(format!(
"resumed schema migration {} has no retained original main database",
marker.run_id
)));
};
write_authority.verify_locked_database_replacement_candidate(&displaced_main, &guard)?;
let retained_logical = logical_witness(&displaced_main)?;
if retained_logical != marker.logical_before {
return Err(BeadsError::internal(format!(
"resumed schema migration {} retained an unrecognized displaced generation at {}",
marker.run_id,
displaced_main.display()
)));
}
sync_directory(before_dir)?;
sync_directory(&displaced_dir)?;
if let Some(parent) = db_path.parent() {
sync_directory(parent)?;
}
sync_directory(run_dir)?;
write_authority.verify_locked_database_replacement_candidate(&displaced_main, &guard)?;
write_authority.verify_database_authority()?;
Ok(guard)
}
fn restore_interrupted_preinstall_family(
db_path: &Path,
before_dir: &Path,
displaced_dir: &Path,
expected: &RawFamilyWitness,
write_authority: &Arc<DatabaseFamilyWriteLock>,
) -> Result<RawFamilyWitness> {
write_authority.verify_database_authority()?;
let expected_main = component_for_suffix(expected, "")?;
let live_main = family_component_path(db_path, "");
let live_main_metadata = secure_file_metadata(&live_main)?.ok_or_else(|| {
BeadsError::internal("interrupted schema migration no longer has a live main database")
})?;
verify_component_bytes(&live_main, &live_main_metadata, expected_main)?;
let displaced_main = backup_component_path(displaced_dir, db_path, "")?;
if secure_file_metadata(&displaced_main)?.is_some() {
return Err(BeadsError::internal(format!(
"interrupted schema migration already displaced its original main database to {}; \
refusing to classify it as a pre-install state",
displaced_main.display()
)));
}
for component in expected
.components
.iter()
.filter(|component| !component.suffix.is_empty())
{
let live = family_component_path(db_path, &component.suffix);
let displaced = backup_component_path(displaced_dir, db_path, &component.suffix)?;
let live_metadata = secure_file_metadata(&live)?;
let displaced_metadata = secure_file_metadata(&displaced)?;
if !component.present {
if live_metadata.is_some() {
return Err(BeadsError::internal(format!(
"interrupted schema migration found an unexpected live sidecar {}; \
refusing to remove or overwrite it",
live.display()
)));
}
continue;
}
if let Some(metadata) = live_metadata {
verify_component_bytes(&live, &metadata, component)?;
continue;
}
if let Some(metadata) = displaced_metadata
&& verify_component_bytes(&displaced, &metadata, component).is_ok()
{
rename_path_no_replace(&displaced, &live)?;
let restored_metadata = secure_file_metadata(&live)?.ok_or_else(|| {
BeadsError::internal(format!(
"interrupted schema-migration sidecar disappeared after restoration: {}",
live.display()
))
})?;
verify_component_bytes(&live, &restored_metadata, component)?;
set_file_permissions(&live, component.unix_mode)?;
continue;
}
let backup = backup_component_path(before_dir, db_path, &component.suffix)?;
copy_regular_file_new(&backup, &live, component.unix_mode)?;
}
if let Some(parent) = db_path.parent() {
sync_directory(parent)?;
}
match fs::symlink_metadata(displaced_dir) {
Ok(metadata) if metadata.is_dir() && !metadata.file_type().is_symlink() => {
sync_directory(displaced_dir)?;
}
Ok(_) => {
return Err(BeadsError::internal(format!(
"schema-migration displaced path is not a real directory: {}",
displaced_dir.display()
)));
}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(error) => return Err(BeadsError::Io(error)),
}
write_authority.verify_database_authority()?;
let restored = raw_family_witness(db_path)?;
if restored != *expected {
return Err(BeadsError::internal(
"interrupted schema migration did not restore the exact receipt-bound raw family",
));
}
Ok(restored)
}
fn validate_failed_against_commit_ready(
failed: &FailedMigrationReceipt,
marker: &CommitReadyMigrationReceipt,
run_dir: &Path,
) -> Result<()> {
if failed.schema_version != FAILED_SCHEMA
|| failed.run_id != marker.run_id
|| failed.database_path != marker.database_path
|| !constant_time_text_eq(&failed.plan_token, &marker.plan_token)
|| failed.marked_at != marker.marked_at
|| failed.raw_before != marker.raw_before
|| failed.logical_before != marker.logical_before
{
return Err(BeadsError::internal(format!(
"failed schema-migration receipt in {} is inconsistent with its commit-ready marker",
run_dir.display()
)));
}
Ok(())
}
fn validate_applied_against_commit_ready(
applied: &AppliedMigrationReceipt,
marker: &CommitReadyMigrationReceipt,
run_dir: &Path,
) -> Result<()> {
validate_commit_ready_marker(marker, run_dir)?;
let raw_after = applied.raw_after.as_ref().ok_or_else(|| {
BeadsError::internal(format!(
"applied schema-migration receipt in {} omits its raw committed witness",
run_dir.display()
))
})?;
validate_raw_family_witness(raw_after)?;
if applied.schema_version != APPLIED_SCHEMA
|| applied.run_id != marker.run_id
|| applied.database_path != marker.database_path
|| !constant_time_text_eq(&applied.plan_token, &marker.plan_token)
|| !constant_time_text_eq(
&applied.prepared_receipt_sha256,
&marker.prepared_receipt_sha256,
)
|| applied.marked_at != marker.marked_at
|| applied.forecast != marker.forecast
|| applied.effects != marker.effects
|| applied.raw_before != marker.raw_before
|| applied.logical_before != marker.logical_before
|| applied.logical_after.as_ref() != Some(&marker.logical_after)
{
return Err(BeadsError::internal(format!(
"applied schema-migration receipt in {} is inconsistent with its commit-ready marker",
run_dir.display()
)));
}
Ok(())
}
fn apply_reviewed_migration(
db_path: &Path,
forecast: &MigrationForecast,
marked_at: &str,
run_dir: &Path,
raw_before: &RawFamilyWitness,
write_authority: &Arc<DatabaseFamilyWriteLock>,
failed_stage: &mut Option<String>,
) -> Result<ReviewedSchemaMigrationEffectsReceipt> {
run_post_migration_maintenance(
db_path,
forecast,
marked_at,
run_dir,
raw_before,
write_authority,
failed_stage,
)
}
fn mark_stage(failed_stage: &mut Option<String>, stage: &str) {
*failed_stage = Some(stage.to_string());
}
fn canonical_index_name(statement: &str) -> Result<&str> {
let create_line = statement
.lines()
.map(str::trim)
.find(|line| line.starts_with("CREATE INDEX ") || line.starts_with("CREATE UNIQUE INDEX "))
.ok_or_else(|| {
BeadsError::internal(
"canonical index statement did not contain a CREATE INDEX declaration",
)
})?;
let mut tokens = create_line.split_ascii_whitespace();
if tokens.next() != Some("CREATE") {
return Err(BeadsError::internal(
"canonical index statement did not begin with CREATE",
));
}
let mut token = tokens.next();
if token == Some("UNIQUE") {
token = tokens.next();
}
if token != Some("INDEX")
|| tokens.next() != Some("IF")
|| tokens.next() != Some("NOT")
|| tokens.next() != Some("EXISTS")
{
return Err(BeadsError::internal(
"canonical index statement did not use CREATE [UNIQUE] INDEX IF NOT EXISTS",
));
}
let name = tokens
.next()
.ok_or_else(|| BeadsError::internal("canonical index statement omitted its index name"))?;
if !name
.chars()
.all(|ch| ch.is_ascii_alphanumeric() || ch == '_')
{
return Err(BeadsError::internal(format!(
"canonical index statement used an unsupported quoted or non-ASCII name: {name}"
)));
}
Ok(name)
}
fn vacuum_candidate_from_private_source(
db_path: &Path,
candidate_path: &Path,
run_dir: &Path,
raw_before: &RawFamilyWitness,
) -> Result<()> {
let before_dir = run_dir.join("before");
verify_backup_family(db_path, &before_dir, raw_before)?;
let retained_source = backup_component_path(&before_dir, db_path, "")?;
let private_source_dir = run_dir.join("maintenance-vacuum-source");
ensure_new_directory(&private_source_dir)?;
copy_family_to_backup(&retained_source, &private_source_dir, raw_before)?;
verify_backup_family(&retained_source, &private_source_dir, raw_before)?;
sync_directory(&private_source_dir)?;
sync_directory(run_dir)?;
let private_source = backup_component_path(&private_source_dir, &retained_source, "")?;
let source_conn = Connection::open(private_source.to_string_lossy().into_owned())?;
let escaped_path = candidate_path.to_string_lossy().replace('\'', "''");
let candidate_result = source_conn
.execute(&format!("VACUUM INTO '{escaped_path}'"))
.map(|_| ())
.map_err(BeadsError::Database);
let close_result = close_connection(source_conn);
match (candidate_result, close_result) {
(Err(error), _) | (Ok(()), Err(error)) => Err(error),
(Ok(()), Ok(())) => Ok(()),
}
}
#[allow(clippy::too_many_lines)]
fn run_post_migration_maintenance(
db_path: &Path,
forecast: &MigrationForecast,
marked_at: &str,
run_dir: &Path,
raw_before: &RawFamilyWitness,
write_authority: &Arc<DatabaseFamilyWriteLock>,
failed_stage: &mut Option<String>,
) -> Result<ReviewedSchemaMigrationEffectsReceipt> {
let from = forecast.from_version;
let to = forecast.to_version;
mark_stage(failed_stage, "source-witness");
write_authority.verify_database_authority()?;
verify_legacy_v15_projection(db_path, forecast.legacy_v15_conversion.as_ref(), true)?;
let source_logical = logical_witness(db_path)?;
let source_permissions_witness = raw_family_witness(db_path)?;
let source_unix_mode = component_for_suffix(&source_permissions_witness, "")?.unix_mode;
mark_stage(failed_stage, "vacuum-candidate");
let candidate_path = maintenance_candidate_path(db_path, run_dir)?;
require_absent_family(&candidate_path)?;
vacuum_candidate_from_private_source(db_path, &candidate_path, run_dir, raw_before)?;
let effects = if from == to {
ReviewedSchemaMigrationEffectsReceipt {
from_version: from,
to_version: to,
content_hash_rows_rebuilt: 0,
gate_result_history_created: false,
prerequisites_column_added: false,
dependency_type_key_added: false,
legacy_v15_tables_converted: false,
post_migration_maintenance_completed: false,
}
} else {
mark_stage(failed_stage, "candidate-schema-migration");
let conn = Connection::open(candidate_path.to_string_lossy().into_owned())?;
conn.execute("PRAGMA foreign_keys = ON")?;
conn.execute("BEGIN IMMEDIATE")?;
let result = (|| {
if forecast.legacy_v15_conversion.is_some() {
apply_reviewed_legacy_v15_core_conversion_in_transaction(&conn)?;
}
run_reviewed_schema_migration_steps_in_transaction(&conn, from, to, marked_at)
})();
let result = match result {
Ok(effects) => match conn.execute("COMMIT") {
Ok(_) => Ok(effects),
Err(error) => {
let _ = conn.execute("ROLLBACK");
Err(BeadsError::Database(error))
}
},
Err(error) => {
let _ = conn.execute("ROLLBACK");
Err(error)
}
};
close_connection(conn)?;
let mut effects = ReviewedSchemaMigrationEffectsReceipt::from(result?);
effects.legacy_v15_tables_converted = forecast.legacy_v15_conversion.is_some();
effects
};
mark_stage(failed_stage, "candidate-maintenance");
let candidate_conn = Connection::open(candidate_path.to_string_lossy().into_owned())?;
let maintenance_result = (|| {
if !crate::storage::schema::issues_required_checks_canonical(&candidate_conn) {
crate::storage::schema::rebuild_issues_table(&candidate_conn)?;
}
let canonical_index_statements =
crate::storage::schema::canonical_index_creation_statements();
for statement in &canonical_index_statements {
let name = canonical_index_name(statement)?;
candidate_conn.execute(&format!("DROP INDEX IF EXISTS {}", quote_identifier(name)))?;
}
for statement in canonical_index_statements {
candidate_conn.execute(statement)?;
}
candidate_conn.execute("REINDEX")?;
candidate_conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")?;
Ok(())
})();
let close_result = close_connection(candidate_conn);
match (maintenance_result, close_result) {
(Err(error), _) | (Ok(()), Err(error)) => return Err(error),
(Ok(()), Ok(())) => {}
}
mark_stage(failed_stage, "candidate-durability");
set_file_permissions(&candidate_path, source_unix_mode)?;
OpenOptions::new()
.read(true)
.write(true)
.open(&candidate_path)
.and_then(|file| file.sync_all())
.map_err(|error| BeadsError::WithContext {
context: format!(
"could not make the schema-migration candidate durable at {}",
candidate_path.display()
),
source: Box::new(BeadsError::Io(error)),
})?;
if let Some(parent) = candidate_path.parent() {
sync_directory(parent)?;
}
mark_stage(failed_stage, "candidate-attestation");
verify_legacy_v15_projection(
&candidate_path,
forecast.legacy_v15_conversion.as_ref(),
false,
)?;
let candidate_logical = logical_witness(&candidate_path)?;
let candidate_matches_reviewed_operation = if from == to {
logical_witnesses_match_except_integrity(&source_logical, &candidate_logical)
} else {
candidate_logical.user_version == to && current_runtime_shape_is_canonical(&candidate_path)?
};
if !integrity_check_is_clean(&candidate_logical.integrity_check)
|| !candidate_matches_reviewed_operation
{
return Err(BeadsError::internal(format!(
"copy-on-write migration candidate did not attest the reviewed operation \
(from={from}, to={to}, source integrity={:?}, candidate integrity={:?}, source \
contents={}, candidate contents={}); the candidate is retained at {}",
source_logical.integrity_check,
candidate_logical.integrity_check,
source_logical.contents_sha256,
candidate_logical.contents_sha256,
candidate_path.display()
)));
}
let source_logical_after = logical_witness(db_path)?;
verify_legacy_v15_projection(db_path, forecast.legacy_v15_conversion.as_ref(), true)?;
if source_logical_after != source_logical {
return Err(BeadsError::internal(format!(
"live database changed while the copy-on-write migration candidate was prepared; \
refusing installation and retaining the candidate at {}",
candidate_path.display()
)));
}
let source_raw = raw_family_witness(db_path)?;
let candidate_raw = raw_family_witness(&candidate_path)?;
mark_stage(failed_stage, "candidate-sidecar-retention");
let candidate_sidecars_dir = run_dir.join("maintenance-candidate-sidecars");
move_present_sidecars_new(
&candidate_path,
&candidate_sidecars_dir,
&candidate_raw,
"candidate-sidecar-retention",
)?;
mark_stage(failed_stage, "commit-ready-marker");
let mut committed_effects = effects;
committed_effects.post_migration_maintenance_completed = true;
persist_commit_ready_marker(run_dir, db_path, &candidate_logical, committed_effects)?;
mark_stage(failed_stage, "lock-replacement-candidate");
let replacement_lock = write_authority.lock_database_replacement_candidate(&candidate_path)?;
mark_stage(failed_stage, "displace-live-sidecars");
let displaced_dir = run_dir.join("maintenance-displaced");
ensure_new_directory(&displaced_dir)?;
move_present_sidecars_new(
db_path,
&displaced_dir,
&source_raw,
"displace-live-sidecars",
)?;
let displaced_main = backup_component_path(&displaced_dir, db_path, "")?;
mark_stage(failed_stage, "install-compacted-candidate");
if let Err(failure) = install_compacted_candidate(
&candidate_path,
db_path,
&displaced_main,
replacement_lock,
write_authority,
) {
let CompactedInstallFailure {
disposition,
error,
stage,
} = failure;
if let Some(stage) = stage {
mark_stage(failed_stage, stage);
}
if disposition == CompactedInstallFailureDisposition::LiveStateUncertain {
return Err(BeadsError::WithContext {
context: format!(
"compacted database installation stopped with an uncertain live-main \
disposition; retained the original sidecars at {} instead of mixing \
database generations",
displaced_dir.display()
),
source: Box::new(error),
});
}
let rollback = restore_present_sidecars(
db_path,
&displaced_dir,
&source_raw,
"rollback-restore-live-sidecars",
);
return match rollback {
Ok(()) => Err(error),
Err(rollback_error) => Err(BeadsError::internal(format!(
"compacted database installation failed ({error}); restoring retained sidecars \
also failed ({rollback_error})"
))),
};
}
mark_stage(failed_stage, "post-install-verification");
let post_install_result = (|| {
verify_backup_family(db_path, &displaced_dir, &source_raw)?;
let installed_logical = logical_witness(db_path)?;
if installed_logical != candidate_logical {
return Err(BeadsError::internal(format!(
"installed compacted database did not match its attested candidate \
(installed integrity={:?}, candidate integrity={:?})",
installed_logical.integrity_check, candidate_logical.integrity_check
)));
}
sync_directory(run_dir)
})();
if let Err(error) = post_install_result {
return Err(rollback_after_compacted_install_failure(
db_path,
&displaced_main,
&displaced_dir,
&source_raw,
run_dir,
write_authority,
error,
));
}
mark_stage(failed_stage, "finalize-replacement");
if let Err(error) = write_authority.finalize_database_replacement() {
return Err(rollback_after_compacted_install_failure(
db_path,
&displaced_main,
&displaced_dir,
&source_raw,
run_dir,
write_authority,
error,
));
}
Ok(committed_effects)
}
fn logical_witnesses_match_except_integrity(
left: &LogicalDatabaseWitness,
right: &LogicalDatabaseWitness,
) -> bool {
left.user_version == right.user_version
&& left.schema_sha256 == right.schema_sha256
&& left.contents_sha256 == right.contents_sha256
&& left.tables == right.tables
}
fn maintenance_candidate_path(db_path: &Path, run_dir: &Path) -> Result<PathBuf> {
let run_id = run_dir
.file_name()
.and_then(|value| value.to_str())
.ok_or_else(|| BeadsError::internal("schema migration run directory has no UTF-8 name"))?;
validate_run_id(run_id)?;
let database_name = db_path
.file_name()
.and_then(|value| value.to_str())
.ok_or_else(|| BeadsError::internal("database path has no UTF-8 file name"))?;
Ok(db_path.with_file_name(format!(".{database_name}.schema-migration-{run_id}.vacuum")))
}
fn require_absent_family(base_path: &Path) -> Result<()> {
for suffix in FAMILY_SUFFIXES {
let path = family_component_path(base_path, suffix);
if secure_file_metadata(&path)?.is_some() {
return Err(BeadsError::internal(format!(
"refusing to overwrite retained schema-migration candidate {}",
path.display()
)));
}
}
Ok(())
}
#[cfg(any(target_os = "linux", target_os = "android", target_vendor = "apple"))]
fn rename_path_no_replace(source: &Path, destination: &Path) -> Result<()> {
use rustix::fs::{CWD, RenameFlags, renameat_with};
match renameat_with(CWD, source, CWD, destination, RenameFlags::NOREPLACE) {
Ok(()) => Ok(()),
Err(error) if error == rustix::io::Errno::EXIST => Err(BeadsError::SyncConflict {
message: format!(
"refusing to overwrite schema-migration artifact {}",
destination.display()
),
}),
Err(error) => Err(BeadsError::Io(std::io::Error::from(error))),
}
}
#[cfg(windows)]
fn rename_path_no_replace(source: &Path, destination: &Path) -> Result<()> {
match crate::sync::rename_path_no_replace_windows(source, destination) {
Ok(()) => Ok(()),
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {
Err(BeadsError::SyncConflict {
message: format!(
"refusing to overwrite schema-migration artifact {}",
destination.display()
),
})
}
Err(error) => Err(BeadsError::Io(error)),
}
}
#[cfg(not(any(
target_os = "linux",
target_os = "android",
target_vendor = "apple",
windows
)))]
fn rename_path_no_replace(_source: &Path, _destination: &Path) -> Result<()> {
Err(BeadsError::Config(
"this platform does not provide the atomic no-replace rename required by reviewed schema migration"
.to_string(),
))
}
const FILE_CONTENTION_RETRY_ATTEMPTS: u32 = 10;
const FILE_CONTENTION_RETRY_STEP: std::time::Duration = std::time::Duration::from_millis(45);
#[cfg(windows)]
fn is_windows_file_contention(error: &BeadsError) -> bool {
const ERROR_ACCESS_DENIED: i32 = 5;
const ERROR_SHARING_VIOLATION: i32 = 32;
match error {
BeadsError::Io(io_error) => matches!(
io_error.raw_os_error(),
Some(ERROR_ACCESS_DENIED | ERROR_SHARING_VIOLATION)
),
_ => false,
}
}
#[cfg(not(windows))]
const fn is_windows_file_contention(_error: &BeadsError) -> bool {
false
}
fn rename_stage_no_replace(stage: &str, source: &Path, destination: &Path) -> Result<()> {
retry_through_file_contention(
stage,
source,
destination,
std::thread::sleep,
is_windows_file_contention,
|| rename_path_no_replace(source, destination),
)
}
fn retry_through_file_contention<S, C, F>(
stage: &str,
source: &Path,
destination: &Path,
mut sleep: S,
is_transient: C,
mut operation: F,
) -> Result<()>
where
S: FnMut(std::time::Duration),
C: Fn(&BeadsError) -> bool,
F: FnMut() -> Result<()>,
{
let mut attempt: u32 = 1;
loop {
let error = match operation() {
Ok(()) => return Ok(()),
Err(error) => error,
};
if !is_transient(&error) {
return Err(error);
}
if attempt >= FILE_CONTENTION_RETRY_ATTEMPTS {
return Err(BeadsError::WithContext {
context: format!(
"schema-migration stage {stage} could not rename {} to {} after {attempt} \
attempts; another process still holds the file open without delete sharing",
source.display(),
destination.display()
),
source: Box::new(error),
});
}
sleep(FILE_CONTENTION_RETRY_STEP.saturating_mul(attempt));
attempt += 1;
}
}
fn move_present_sidecars_new(
source_base: &Path,
destination_dir: &Path,
expected: &RawFamilyWitness,
stage: &str,
) -> Result<()> {
let present_sidecars = expected
.components
.iter()
.filter(|component| component.present && !component.suffix.is_empty())
.collect::<Vec<_>>();
if present_sidecars.is_empty() {
return Ok(());
}
ensure_directory(destination_dir)?;
let rollback_stage = format!("{stage}-rollback");
let mut moved = Vec::with_capacity(present_sidecars.len());
for component in present_sidecars {
let source = family_component_path(source_base, &component.suffix);
let prepared_move = (|| {
let destination =
backup_component_path(destination_dir, source_base, &component.suffix)?;
let metadata = secure_file_metadata(&source)?.ok_or_else(|| {
BeadsError::internal(format!(
"schema-migration sidecar disappeared before retention: {}",
source.display()
))
})?;
verify_component_bytes(&source, &metadata, component)?;
Ok(destination)
})();
let destination = match prepared_move {
Ok(destination) => destination,
Err(error) => {
let rollback =
restore_moved_components(source_base, destination_dir, &moved, &rollback_stage);
return match rollback {
Ok(()) => Err(error),
Err(rollback_error) => Err(BeadsError::internal(format!(
"could not prepare schema-migration sidecar move for {} ({error}); \
rollback of prior moves also failed ({rollback_error})",
source.display()
))),
};
}
};
if let Err(error) = rename_stage_no_replace(stage, &source, &destination) {
let rollback =
restore_moved_components(source_base, destination_dir, &moved, &rollback_stage);
return match rollback {
Ok(()) => Err(error),
Err(rollback_error) => Err(BeadsError::internal(format!(
"could not retain schema-migration sidecar {} ({error}); rollback of prior \
sidecar moves also failed: {rollback_error}",
source.display()
))),
};
}
moved.push((*component).clone());
let moved_verification = (|| {
let moved_metadata = secure_file_metadata(&destination)?.ok_or_else(|| {
BeadsError::internal(format!(
"schema-migration sidecar disappeared after retention: {}",
destination.display()
))
})?;
verify_component_bytes(&destination, &moved_metadata, component)
})();
if let Err(error) = moved_verification {
let rollback =
restore_moved_components(source_base, destination_dir, &moved, &rollback_stage);
return match rollback {
Ok(()) => Err(error),
Err(rollback_error) => Err(BeadsError::internal(format!(
"retained schema-migration sidecar {} failed its post-move witness \
({error}); rollback also failed ({rollback_error})",
destination.display()
))),
};
}
}
let sync_result = if let Some(parent) = source_base.parent() {
sync_directory(parent).and_then(|()| sync_directory(destination_dir))
} else {
sync_directory(destination_dir)
};
if let Err(error) = sync_result {
let rollback =
restore_moved_components(source_base, destination_dir, &moved, &rollback_stage);
return match rollback {
Ok(()) => Err(error),
Err(rollback_error) => Err(BeadsError::internal(format!(
"retained schema-migration sidecars could not be made durable ({error}); \
rollback also failed ({rollback_error})"
))),
};
}
Ok(())
}
fn restore_moved_components(
destination_base: &Path,
source_dir: &Path,
components: &[RawComponentWitness],
stage: &str,
) -> Result<()> {
for component in components.iter().rev() {
let source = backup_component_path(source_dir, destination_base, &component.suffix)?;
let destination = family_component_path(destination_base, &component.suffix);
let source_metadata = secure_file_metadata(&source)?.ok_or_else(|| {
BeadsError::internal(format!(
"retained schema-migration component disappeared before restoration: {}",
source.display()
))
})?;
verify_component_bytes(&source, &source_metadata, component)?;
rename_stage_no_replace(stage, &source, &destination)?;
let restored_metadata = secure_file_metadata(&destination)?.ok_or_else(|| {
BeadsError::internal(format!(
"schema-migration component disappeared after restoration: {}",
destination.display()
))
})?;
verify_component_bytes(&destination, &restored_metadata, component)?;
}
if let Some(parent) = destination_base.parent() {
sync_directory(parent)?;
}
sync_directory(source_dir)
}
fn restore_present_sidecars(
destination_base: &Path,
source_dir: &Path,
expected: &RawFamilyWitness,
stage: &str,
) -> Result<()> {
let components = expected
.components
.iter()
.filter(|component| component.present && !component.suffix.is_empty())
.cloned()
.collect::<Vec<_>>();
restore_moved_components(destination_base, source_dir, &components, stage)
}
#[cfg(any(target_os = "linux", target_os = "android", target_vendor = "apple"))]
fn exchange_database_paths(left: &Path, right: &Path) -> Result<()> {
use rustix::fs::{CWD, RenameFlags, renameat_with};
renameat_with(CWD, left, CWD, right, RenameFlags::EXCHANGE)
.map_err(|error| BeadsError::Io(std::io::Error::from(error)))
}
#[cfg(any(target_os = "linux", target_os = "android", target_vendor = "apple"))]
fn verify_migration_exchange_support<F>(db_path: &Path, exchange: F) -> Result<()>
where
F: FnOnce(&Path, &Path) -> Result<()>,
{
use rustix::fs::{FlockOperation, flock};
use std::os::unix::fs::MetadataExt;
let parent = db_path
.parent()
.ok_or_else(|| BeadsError::Config("migration database has no parent directory".into()))?;
let probe = tempfile::Builder::new()
.prefix(".br-schema-exchange-")
.tempdir_in(parent)?;
let outcome = (|| -> Result<()> {
let first_path = probe.path().join("first");
let second_path = probe.path().join("second");
let mut first = open_private_file_new(&first_path)?;
let mut second = open_private_file_new(&second_path)?;
first.write_all(b"first exchange probe")?;
second.write_all(b"second exchange probe")?;
for file in [&first, &second] {
flock(file, FlockOperation::NonBlockingLockExclusive).map_err(std::io::Error::from)?;
}
let identity = |metadata: &fs::Metadata| (metadata.dev(), metadata.ino());
let first_before = identity(&first.metadata()?);
let second_before = identity(&second.metadata()?);
let database_device = fs::symlink_metadata(db_path)?.dev();
if first_before == second_before
|| first_before.0 != database_device
|| second_before.0 != database_device
{
return Err(BeadsError::Config(
"exchange probe cannot establish distinct files on the database filesystem".into(),
));
}
exchange(&first_path, &second_path)?;
let first_after = fs::symlink_metadata(&first_path)?;
let second_after = fs::symlink_metadata(&second_path)?;
if !first_after.is_file()
|| !second_after.is_file()
|| identity(&first_after) != second_before
|| identity(&second_after) != first_before
|| identity(&first.metadata()?) != first_before
|| identity(&second.metadata()?) != second_before
{
return Err(BeadsError::Config(
"atomic exchange did not preserve both locked file identities".into(),
));
}
Ok(())
})();
if let Err(error) = outcome {
let retained = probe.keep();
return Err(BeadsError::Config(format!(
"schema migration filesystem exchange preflight failed ({error}); refusing \
database replacement. Use the native host binary or a native Linux volume \
instead of an incompatible shared mount. Probe retained at {}",
retained.display()
)));
}
Ok(())
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum CompactedInstallFailureDisposition {
OriginalRestored,
LiveStateUncertain,
}
#[derive(Debug)]
struct CompactedInstallFailure {
disposition: CompactedInstallFailureDisposition,
error: BeadsError,
stage: Option<&'static str>,
}
impl CompactedInstallFailure {
fn restored(error: BeadsError) -> Self {
Self {
disposition: CompactedInstallFailureDisposition::OriginalRestored,
error,
stage: None,
}
}
fn uncertain(error: BeadsError) -> Self {
Self {
disposition: CompactedInstallFailureDisposition::LiveStateUncertain,
error,
stage: None,
}
}
fn at_stage(mut self, stage: &'static str) -> Self {
self.stage = Some(stage);
self
}
fn from_original_state(
error: BeadsError,
write_authority: &Arc<DatabaseFamilyWriteLock>,
) -> Self {
match write_authority.verify_database_authority() {
Ok(()) => Self::restored(error),
Err(authority_error) => Self::uncertain(BeadsError::WithContext {
context: format!(
"schema-migration installation failed before replacement, but the original \
database authority could not be re-verified ({authority_error})"
),
source: Box::new(error),
}),
}
}
fn after_rollback(error: BeadsError, rollback: Result<()>) -> Self {
match rollback {
Ok(()) => Self::restored(error),
Err(rollback_error) => Self::uncertain(BeadsError::WithContext {
context: format!(
"schema-migration installation failed and rollback could not prove the \
original database generation durable and authoritative ({rollback_error})"
),
source: Box::new(error),
}),
}
}
}
impl std::fmt::Display for CompactedInstallFailure {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
std::fmt::Display::fmt(&self.error, formatter)
}
}
fn rollback_installed_main_to_original<F>(
_candidate_path: &Path,
db_path: &Path,
displaced_main: &Path,
write_authority: &Arc<DatabaseFamilyWriteLock>,
sync: &mut F,
) -> Result<()>
where
F: FnMut(&Path) -> Result<()>,
{
if write_authority.database_target_authority_state()? != DatabaseTargetAuthorityState::Held {
return Err(BeadsError::SyncConflict {
message: "refusing schema-migration rollback because the live database is no longer the authority-held replacement"
.to_string(),
});
}
#[cfg(any(target_os = "linux", target_os = "android", target_vendor = "apple"))]
{
exchange_database_paths(db_path, displaced_main)?;
write_authority.restore_retained_database_inode_after_authorized_replace()?;
}
#[cfg(not(any(target_os = "linux", target_os = "android", target_vendor = "apple")))]
{
rename_stage_no_replace(
"install-rollback-rename-live-to-candidate",
db_path,
_candidate_path,
)?;
rename_stage_no_replace(
"install-rollback-rename-displaced-to-live",
displaced_main,
db_path,
)?;
write_authority.restore_retained_database_inode_after_authorized_replace()?;
}
write_authority.verify_database_authority()?;
if let Some(parent) = db_path.parent() {
sync(parent)?;
}
if let Some(parent) = displaced_main.parent() {
sync(parent)?;
}
Ok(())
}
fn install_compacted_candidate(
candidate_path: &Path,
db_path: &Path,
displaced_main: &Path,
replacement_lock: File,
write_authority: &Arc<DatabaseFamilyWriteLock>,
) -> std::result::Result<(), CompactedInstallFailure> {
install_compacted_candidate_with_sync(
candidate_path,
db_path,
displaced_main,
replacement_lock,
write_authority,
sync_directory,
)
}
#[allow(clippy::too_many_lines)]
fn install_compacted_candidate_with_sync<F>(
candidate_path: &Path,
db_path: &Path,
displaced_main: &Path,
replacement_lock: File,
write_authority: &Arc<DatabaseFamilyWriteLock>,
mut sync: F,
) -> std::result::Result<(), CompactedInstallFailure>
where
F: FnMut(&Path) -> Result<()>,
{
let displaced_metadata = match secure_file_metadata(displaced_main) {
Ok(metadata) => metadata,
Err(error) => {
return Err(
CompactedInstallFailure::from_original_state(error, write_authority)
.at_stage("install-preflight"),
);
}
};
if displaced_metadata.is_some() {
return Err(CompactedInstallFailure::from_original_state(
BeadsError::internal(format!(
"refusing to overwrite retained pre-compaction database {}",
displaced_main.display()
)),
write_authority,
)
.at_stage("install-preflight"));
}
#[cfg(any(target_os = "linux", target_os = "android", target_vendor = "apple"))]
{
if let Err(error) = exchange_database_paths(candidate_path, db_path) {
return Err(
CompactedInstallFailure::from_original_state(error, write_authority)
.at_stage("install-exchange-candidate-with-live"),
);
}
if let Err(error) =
write_authority.verify_staged_database_recovery_authority(candidate_path)
{
let rollback = exchange_database_paths(candidate_path, db_path)
.and_then(|()| write_authority.verify_database_authority())
.and_then(|()| db_path.parent().map_or(Ok(()), &mut sync));
return Err(CompactedInstallFailure::after_rollback(error, rollback)
.at_stage("install-verify-staged-authority"));
}
if let Err(error) = write_authority.adopt_locked_database_replacement(replacement_lock) {
let rollback = exchange_database_paths(candidate_path, db_path)
.and_then(|()| {
write_authority.restore_retained_database_inode_after_authorized_replace()
})
.and_then(|()| write_authority.verify_database_authority())
.and_then(|()| db_path.parent().map_or(Ok(()), &mut sync));
return Err(CompactedInstallFailure::after_rollback(error, rollback)
.at_stage("install-adopt-replacement"));
}
if let Err(error) = rename_stage_no_replace(
"install-rename-candidate-to-displaced",
candidate_path,
displaced_main,
) {
let rollback = exchange_database_paths(db_path, candidate_path)
.and_then(|()| {
write_authority.restore_retained_database_inode_after_authorized_replace()
})
.and_then(|()| write_authority.verify_database_authority())
.and_then(|()| db_path.parent().map_or(Ok(()), &mut sync));
return Err(CompactedInstallFailure::after_rollback(error, rollback)
.at_stage("install-rename-candidate-to-displaced"));
}
}
#[cfg(not(any(target_os = "linux", target_os = "android", target_vendor = "apple")))]
{
if let Err(error) =
rename_stage_no_replace("install-rename-live-to-displaced", db_path, displaced_main)
{
return Err(
CompactedInstallFailure::from_original_state(error, write_authority)
.at_stage("install-rename-live-to-displaced"),
);
}
if let Err(error) =
write_authority.verify_staged_database_recovery_authority(displaced_main)
{
let rollback = rename_stage_no_replace(
"install-rollback-rename-displaced-to-live",
displaced_main,
db_path,
)
.and_then(|()| write_authority.verify_database_authority())
.and_then(|()| db_path.parent().map_or(Ok(()), &mut sync));
return Err(CompactedInstallFailure::after_rollback(error, rollback)
.at_stage("install-verify-staged-authority"));
}
if let Err(error) =
rename_stage_no_replace("install-rename-candidate-to-live", candidate_path, db_path)
{
let rollback = rename_stage_no_replace(
"install-rollback-rename-displaced-to-live",
displaced_main,
db_path,
)
.and_then(|()| write_authority.verify_database_authority())
.and_then(|()| db_path.parent().map_or(Ok(()), &mut sync));
return Err(CompactedInstallFailure::after_rollback(error, rollback)
.at_stage("install-rename-candidate-to-live"));
}
if let Err(error) = write_authority.adopt_locked_database_replacement(replacement_lock) {
let rollback = rename_stage_no_replace(
"install-rollback-rename-live-to-candidate",
db_path,
candidate_path,
)
.and_then(|()| {
rename_stage_no_replace(
"install-rollback-rename-displaced-to-live",
displaced_main,
db_path,
)
})
.and_then(|()| {
write_authority.restore_retained_database_inode_after_authorized_replace()
})
.and_then(|()| write_authority.verify_database_authority())
.and_then(|()| db_path.parent().map_or(Ok(()), &mut sync));
return Err(CompactedInstallFailure::after_rollback(error, rollback)
.at_stage("install-adopt-replacement"));
}
}
let durability_result = db_path
.parent()
.map_or(Ok(()), &mut sync)
.and_then(|()| displaced_main.parent().map_or(Ok(()), &mut sync));
if let Err(error) = durability_result {
let rollback = rollback_installed_main_to_original(
candidate_path,
db_path,
displaced_main,
write_authority,
&mut sync,
);
return Err(
CompactedInstallFailure::after_rollback(error, rollback).at_stage("install-durability")
);
}
Ok(())
}
fn rollback_after_compacted_install_failure(
db_path: &Path,
displaced_main: &Path,
displaced_dir: &Path,
source_raw: &RawFamilyWitness,
run_dir: &Path,
write_authority: &Arc<DatabaseFamilyWriteLock>,
operation_error: BeadsError,
) -> BeadsError {
let target_state = match write_authority.database_target_authority_state() {
Ok(state) => state,
Err(authority_error) => {
return BeadsError::WithContext {
context: format!(
"schema-migration post-installation validation failed, and the live database \
authority could not be classified ({authority_error}); retained the original \
family at {} and did not mix generations",
displaced_dir.display()
),
source: Box::new(operation_error),
};
}
};
if target_state != DatabaseTargetAuthorityState::Held {
return BeadsError::WithContext {
context: format!(
"schema-migration post-installation validation failed while the live database \
had authority state {target_state:?}; retained the original family at {} and \
left the live generation untouched",
displaced_dir.display()
),
source: Box::new(operation_error),
};
}
let failed_dir = run_dir.join("maintenance-failed-new-family");
let rollback_result = (|| {
ensure_new_directory(&failed_dir)?;
let installed_raw = raw_family_witness(db_path)?;
move_present_sidecars_new(
db_path,
&failed_dir,
&installed_raw,
"rollback-quarantine-installed-sidecars",
)?;
rollback_compacted_install(db_path, displaced_main, &failed_dir, write_authority)?;
restore_present_sidecars(
db_path,
displaced_dir,
source_raw,
"rollback-restore-live-sidecars",
)?;
write_authority.verify_database_authority()?;
let restored_raw = raw_family_witness(db_path)?;
if restored_raw != *source_raw {
return Err(BeadsError::internal(
"schema-migration rollback did not reproduce the attested original database family",
));
}
Ok(())
})();
match rollback_result {
Ok(()) => BeadsError::WithContext {
context: format!(
"schema-migration post-installation validation failed; restored the original \
database family and retained the rejected replacement at {}",
failed_dir.display()
),
source: Box::new(operation_error),
},
Err(rollback_error) => BeadsError::WithContext {
context: format!(
"schema-migration post-installation validation failed and rollback could not \
prove the original family restored ({rollback_error}); recovery artifacts remain \
under {}",
run_dir.display()
),
source: Box::new(operation_error),
},
}
}
fn rollback_compacted_install(
db_path: &Path,
displaced_main: &Path,
failed_dir: &Path,
write_authority: &Arc<DatabaseFamilyWriteLock>,
) -> Result<()> {
if write_authority.database_target_authority_state()? != DatabaseTargetAuthorityState::Held {
return Err(BeadsError::SyncConflict {
message: "refusing to roll back a compacted database that is no longer the authority-held live generation"
.to_string(),
});
}
ensure_directory(failed_dir)?;
let failed_main = backup_component_path(failed_dir, db_path, "")?;
if secure_file_metadata(&failed_main)?.is_some() {
return Err(BeadsError::internal(format!(
"refusing to overwrite retained failed compacted database {}",
failed_main.display()
)));
}
#[cfg(any(target_os = "linux", target_os = "android", target_vendor = "apple"))]
{
exchange_database_paths(db_path, displaced_main)?;
write_authority.restore_retained_database_inode_after_authorized_replace()?;
rename_path_no_replace(displaced_main, &failed_main)?;
}
#[cfg(not(any(target_os = "linux", target_os = "android", target_vendor = "apple")))]
{
rename_stage_no_replace("rollback-quarantine-installed-main", db_path, &failed_main)?;
rename_stage_no_replace("rollback-restore-displaced-main", displaced_main, db_path)?;
write_authority.restore_retained_database_inode_after_authorized_replace()?;
}
if let Some(parent) = db_path.parent() {
sync_directory(parent)?;
}
sync_directory(failed_dir)
}
#[allow(
clippy::too_many_lines,
reason = "undo is a linear crash-recovery protocol whose operation ordering is safety-critical"
)]
fn execute_undo(args: &DoctorMigrateSchemaUndoArgs, migration: &MigrationContext) -> Result<()> {
validate_run_id(&args.run_id)?;
let run_dir = migration_runs_root(&migration.beads_dir).join(&args.run_id);
if run_dir.join("undone.json").exists() {
let receipt: UndoReceipt = read_json(&run_dir.join("undone.json"))?;
validate_completed_undo_receipt(&receipt, args, migration)?;
return emit_undo(&receipt, args.json);
}
let applied_path = run_dir.join("applied.json");
let applied_receipt_sha256 = file_sha256(&applied_path)?;
let applied: AppliedMigrationReceipt = read_json(&applied_path)?;
validate_applied_receipt(&applied, args, migration, &run_dir)?;
let before_dir = run_dir.join("before");
verify_backup_family(&migration.db_path, &before_dir, &applied.raw_before)?;
#[cfg(any(target_os = "linux", target_os = "android", target_vendor = "apple"))]
if !args.dry_run {
verify_migration_exchange_support(&migration.db_path, exchange_database_paths)?;
}
let undo_prepared_path = run_dir.join("undo-prepared.json");
let mut receipt = if undo_prepared_path.exists() {
if args.dry_run {
return Err(BeadsError::internal(format!(
"schema migration undo for run {} is already prepared; rerun without \
--dry-run to resume it",
args.run_id
)));
}
let receipt: UndoReceipt = read_json(&undo_prepared_path)?;
validate_prepared_undo_receipt(
&receipt,
args,
migration,
&run_dir,
&applied_receipt_sha256,
&applied,
)?;
receipt
} else {
let logical_live = logical_witness(&migration.db_path).ok();
let projection_unchanged = verify_legacy_v15_projection(
&migration.db_path,
applied.forecast.legacy_v15_conversion.as_ref(),
false,
)
.is_ok();
let raw_live = raw_family_witness(&migration.db_path)?;
require_unchanged_applied_state(
&applied,
&raw_live,
logical_live.as_ref(),
&args.run_id,
projection_unchanged,
)?;
let quarantine_id = format!(
"undo-{}-{}",
Utc::now().format("%Y%m%dT%H%M%S%.6fZ"),
std::process::id()
);
let quarantine_dir = run_dir.join("undo-quarantine").join(quarantine_id);
let receipt = UndoReceipt {
schema_version: UNDO_SCHEMA.to_string(),
run_id: args.run_id.clone(),
dry_run: args.dry_run,
database_path: migration.db_path.display().to_string(),
quarantine_path: quarantine_dir.display().to_string(),
applied_receipt_sha256: applied_receipt_sha256.clone(),
raw_expected_before: applied.raw_before.clone(),
logical_expected_before: applied.logical_before.clone(),
raw_live_before_undo: raw_live,
logical_live_before_undo: logical_live,
raw_restored: None,
logical_restored: None,
};
if args.dry_run {
emit_undo(&receipt, args.json)?;
return Ok(());
}
let quarantine_parent = quarantine_dir.parent().ok_or_else(|| {
BeadsError::internal("schema migration undo quarantine path has no parent")
})?;
ensure_directory(quarantine_parent)?;
set_private_directory_permissions(quarantine_parent)?;
write_json_new(&undo_prepared_path, &receipt)?;
receipt
};
let quarantine_dir = validate_quarantine_path(&receipt, &run_dir)?;
if quarantine_dir.exists() {
ensure_directory(&quarantine_dir)?;
} else {
ensure_new_directory(&quarantine_dir)?;
}
require_atomic_undo_exchange()?;
let restored_candidate_dir = run_dir.join("undo-restored-candidate");
ensure_directory(&restored_candidate_dir)?;
set_private_directory_permissions(&restored_candidate_dir)?;
let restored_candidate =
backup_component_path(&restored_candidate_dir, &migration.db_path, "")?;
let quarantined_main_guard = install_undo_restored_main_resuming(
&migration.db_path,
&before_dir,
&restored_candidate,
&quarantine_dir,
&receipt.raw_live_before_undo,
&applied.raw_before,
&migration.write_authority,
)?;
quarantine_live_family_resuming(
&migration.db_path,
&quarantine_dir,
&receipt.raw_live_before_undo,
&applied.raw_before,
)?;
restore_backup_family_resuming(&migration.db_path, &before_dir, &applied.raw_before)?;
let raw_restored_exact = raw_family_witness(&migration.db_path)?;
if raw_restored_exact != applied.raw_before {
return Err(BeadsError::internal(format!(
"schema migration undo for run {} did not reproduce the exact receipt-bound raw \
pre-state; the displaced applied state remains quarantined at {}",
args.run_id,
quarantine_dir.display()
)));
}
let logical_restored = logical_witness(&migration.db_path)?;
let raw_restored_after_probe = raw_family_witness(&migration.db_path)?;
if logical_restored != applied.logical_before
|| !stable_raw_eq(&raw_restored_after_probe, &raw_restored_exact)
{
return Err(BeadsError::internal(format!(
"schema migration undo for run {} did not reproduce the verified pre-state; \
the displaced applied state remains quarantined at {}",
args.run_id,
quarantine_dir.display()
)));
}
receipt.raw_restored = Some(raw_restored_exact);
receipt.logical_restored = Some(logical_restored);
receipt.dry_run = false;
migration.write_authority.verify_database_authority()?;
if let Some(parent) = migration.db_path.parent() {
sync_directory(parent)?;
}
sync_directory(&quarantine_dir)?;
sync_directory(&restored_candidate_dir)?;
sync_directory(&run_dir)?;
migration.write_authority.verify_database_authority()?;
migration.write_authority.finalize_database_replacement()?;
drop(quarantined_main_guard);
write_json_new(&run_dir.join("undone.json"), &receipt)?;
sync_directory(&run_dir)?;
emit_undo(&receipt, args.json)
}
#[cfg(any(target_os = "linux", target_os = "android", target_vendor = "apple"))]
#[allow(clippy::unnecessary_wraps)]
fn require_atomic_undo_exchange() -> Result<()> {
Ok(())
}
#[cfg(not(any(target_os = "linux", target_os = "android", target_vendor = "apple")))]
fn require_atomic_undo_exchange() -> Result<()> {
Err(BeadsError::Config(
"schema-migration undo requires an atomic path-exchange primitive on this platform"
.to_string(),
))
}
#[cfg(any(target_os = "linux", target_os = "android", target_vendor = "apple"))]
#[allow(clippy::too_many_arguments, clippy::too_many_lines)]
fn install_undo_restored_main_resuming(
db_path: &Path,
before_dir: &Path,
restored_candidate: &Path,
quarantine_dir: &Path,
applied: &RawFamilyWitness,
restored: &RawFamilyWitness,
write_authority: &Arc<DatabaseFamilyWriteLock>,
) -> Result<Option<File>> {
let applied_main = component_for_suffix(applied, "")?;
let restored_main = component_for_suffix(restored, "")?;
if !applied_main.present || !restored_main.present {
return Err(BeadsError::internal(
"schema-migration undo receipts must witness both main database files",
));
}
let quarantined_main = backup_component_path(quarantine_dir, db_path, "")?;
let live_metadata = secure_file_metadata(db_path)?
.ok_or_else(|| BeadsError::internal("schema-migration undo found no live main database"))?;
let live_matches_applied = component_bytes_match(db_path, &live_metadata, applied_main)?;
let live_matches_restored = component_bytes_match(db_path, &live_metadata, restored_main)?;
if let Some(quarantined_metadata) = secure_file_metadata(&quarantined_main)? {
verify_component_bytes(&quarantined_main, &quarantined_metadata, applied_main)?;
if !live_matches_restored {
return Err(BeadsError::internal(
"schema-migration undo has a quarantined applied main but the live main is not the receipt-bound original",
));
}
let guard = write_authority.lock_database_replacement_candidate(&quarantined_main)?;
write_authority.verify_locked_database_replacement_candidate(&quarantined_main, &guard)?;
write_authority.verify_database_authority()?;
return Ok(Some(guard));
}
if let Some(candidate_metadata) = secure_file_metadata(restored_candidate)? {
if component_bytes_match(restored_candidate, &candidate_metadata, applied_main)?
&& live_matches_restored
{
let guard = write_authority.lock_database_replacement_candidate(restored_candidate)?;
write_authority
.verify_locked_database_replacement_candidate(restored_candidate, &guard)?;
write_authority.verify_database_authority()?;
rename_path_no_replace(restored_candidate, &quarantined_main)?;
write_authority
.verify_locked_database_replacement_candidate(&quarantined_main, &guard)?;
sync_directory(quarantine_dir)?;
if let Some(parent) = db_path.parent() {
sync_directory(parent)?;
}
return Ok(Some(guard));
}
if !component_bytes_match(restored_candidate, &candidate_metadata, restored_main)? {
return Err(BeadsError::internal(format!(
"schema-migration undo staging path contains an unrecognized generation: {}",
restored_candidate.display()
)));
}
} else {
if !live_matches_applied {
return Err(BeadsError::internal(
"schema-migration undo cannot classify the live main database before staging",
));
}
let backup_main = backup_component_path(before_dir, db_path, "")?;
copy_regular_file_new(&backup_main, restored_candidate, restored_main.unix_mode)?;
if let Some(parent) = restored_candidate.parent() {
sync_directory(parent)?;
}
}
if !live_matches_applied {
return Err(BeadsError::internal(
"schema-migration undo staging is prepared, but the live main is no longer the applied generation",
));
}
write_authority.verify_database_authority()?;
let replacement_lock =
write_authority.lock_database_replacement_candidate(restored_candidate)?;
write_authority
.verify_locked_database_replacement_candidate(restored_candidate, &replacement_lock)?;
exchange_database_paths(restored_candidate, db_path)?;
if let Err(error) =
write_authority.verify_staged_database_recovery_authority(restored_candidate)
{
let rollback = exchange_database_paths(restored_candidate, db_path)
.and_then(|()| write_authority.verify_database_authority());
return match rollback {
Ok(()) => Err(error),
Err(rollback_error) => Err(BeadsError::WithContext {
context: format!(
"schema-migration undo rejected a foreign displaced main and rollback could not re-prove the applied authority ({rollback_error})"
),
source: Box::new(error),
}),
};
}
if let Err(error) = write_authority.adopt_locked_database_replacement(replacement_lock) {
let rollback = exchange_database_paths(restored_candidate, db_path)
.and_then(|()| {
write_authority.restore_retained_database_inode_after_authorized_replace()
})
.and_then(|()| write_authority.verify_database_authority());
return match rollback {
Ok(()) => Err(error),
Err(rollback_error) => Err(BeadsError::WithContext {
context: format!(
"schema-migration undo could not adopt its prelocked original and rollback could not re-prove the applied authority ({rollback_error})"
),
source: Box::new(error),
}),
};
}
rename_path_no_replace(restored_candidate, &quarantined_main)?;
if let Some(parent) = db_path.parent() {
sync_directory(parent)?;
}
sync_directory(quarantine_dir)?;
write_authority.verify_database_authority()?;
Ok(None)
}
#[cfg(not(any(target_os = "linux", target_os = "android", target_vendor = "apple")))]
#[allow(clippy::too_many_arguments)]
fn install_undo_restored_main_resuming(
_db_path: &Path,
_before_dir: &Path,
_restored_candidate: &Path,
_quarantine_dir: &Path,
_applied: &RawFamilyWitness,
_restored: &RawFamilyWitness,
_write_authority: &Arc<DatabaseFamilyWriteLock>,
) -> Result<Option<File>> {
require_atomic_undo_exchange()?;
unreachable!("unsupported schema-migration undo exchange returned success")
}
fn component_bytes_match(
path: &Path,
metadata: &fs::Metadata,
expected: &RawComponentWitness,
) -> Result<bool> {
let (length, sha256) = hash_regular_file(path, metadata)?;
Ok(expected.present
&& expected.length == Some(length)
&& expected.sha256.as_deref() == Some(&sha256))
}
fn validate_applied_receipt(
applied: &AppliedMigrationReceipt,
args: &DoctorMigrateSchemaUndoArgs,
migration: &MigrationContext,
run_dir: &Path,
) -> Result<()> {
validate_raw_family_witness(&applied.raw_before)?;
if let Some(raw_after) = applied.raw_after.as_ref() {
validate_raw_family_witness(raw_after)?;
}
if applied.schema_version != APPLIED_SCHEMA {
return Err(BeadsError::internal(format!(
"unsupported applied schema-migration receipt contract {:?}",
applied.schema_version
)));
}
if applied.run_id != args.run_id {
return Err(BeadsError::internal(format!(
"applied receipt run-id mismatch (path={}, receipt={})",
args.run_id, applied.run_id
)));
}
if applied.database_path != migration.db_path.display().to_string() {
return Err(BeadsError::internal(format!(
"schema migration run {} belongs to {}, not {}",
args.run_id,
applied.database_path,
migration.db_path.display()
)));
}
if applied.attested != applied.attestation_errors.is_empty() {
return Err(BeadsError::internal(format!(
"schema migration run {} has an internally inconsistent attestation receipt",
args.run_id
)));
}
if applied.raw_after.is_none() && applied.logical_after.is_none() {
return Err(BeadsError::internal(format!(
"schema migration run {} has no committed-state witness and cannot be safely undone",
args.run_id
)));
}
let recomputed_token = compute_plan_token(
&applied.database_path,
&applied.logical_before,
&applied.forecast,
)?;
if !constant_time_text_eq(&recomputed_token, &applied.plan_token) {
return Err(BeadsError::internal(format!(
"schema migration run {} has a plan token that does not bind its recorded pre-state",
args.run_id
)));
}
let prepared_path = run_dir.join("prepared.json");
let prepared_sha256 = file_sha256(&prepared_path)?;
if !constant_time_text_eq(&prepared_sha256, &applied.prepared_receipt_sha256) {
return Err(BeadsError::internal(format!(
"schema migration run {} failed its prepared-to-applied receipt hash chain",
args.run_id
)));
}
let prepared: PreparedMigrationReceipt = read_json(&prepared_path)?;
if prepared.schema_version != PREPARED_SCHEMA
|| prepared.run_id != applied.run_id
|| prepared.database_path != applied.database_path
|| !constant_time_text_eq(&prepared.plan_token, &applied.plan_token)
|| prepared.marked_at != applied.marked_at
|| prepared.forecast != applied.forecast
|| prepared.raw_before != applied.raw_before
|| prepared.logical_before != applied.logical_before
{
return Err(BeadsError::internal(format!(
"schema migration run {} has inconsistent prepared and applied receipts",
args.run_id
)));
}
let marker: CommitReadyMigrationReceipt = read_json(&run_dir.join("commit-ready.json"))?;
validate_applied_against_commit_ready(applied, &marker, run_dir)?;
Ok(())
}
fn validate_completed_undo_receipt(
receipt: &UndoReceipt,
args: &DoctorMigrateSchemaUndoArgs,
migration: &MigrationContext,
) -> Result<()> {
validate_raw_family_witness(&receipt.raw_expected_before)?;
validate_raw_family_witness(&receipt.raw_live_before_undo)?;
let run_dir = migration_runs_root(&migration.beads_dir).join(&args.run_id);
if receipt.schema_version != UNDO_SCHEMA
|| receipt.run_id != args.run_id
|| receipt.database_path != migration.db_path.display().to_string()
|| receipt.dry_run
{
return Err(BeadsError::internal(format!(
"schema migration run {} has an invalid completed undo receipt",
args.run_id
)));
}
let applied_path = run_dir.join("applied.json");
let applied_sha256 = file_sha256(&applied_path)?;
if !constant_time_text_eq(&applied_sha256, &receipt.applied_receipt_sha256) {
return Err(BeadsError::internal(format!(
"schema migration run {} failed its applied-to-undo receipt hash chain",
args.run_id
)));
}
let applied: AppliedMigrationReceipt = read_json(&applied_path)?;
validate_applied_receipt(&applied, args, migration, &run_dir)?;
let quarantine_dir = validate_quarantine_path(receipt, &run_dir)?;
let logical_restored = receipt.logical_restored.as_ref().ok_or_else(|| {
BeadsError::internal(format!(
"schema migration run {} completed undo receipt omits its logical witness",
args.run_id
))
})?;
let raw_restored = receipt.raw_restored.as_ref().ok_or_else(|| {
BeadsError::internal(format!(
"schema migration run {} completed undo receipt omits its raw witness",
args.run_id
))
})?;
validate_raw_family_witness(raw_restored)?;
if receipt.raw_expected_before != applied.raw_before
|| receipt.logical_expected_before != applied.logical_before
|| receipt.logical_live_before_undo.as_ref() != applied.logical_after.as_ref()
|| raw_restored != &applied.raw_before
|| logical_restored != &applied.logical_before
{
return Err(BeadsError::internal(format!(
"schema migration run {} has an internally inconsistent completed undo receipt",
args.run_id
)));
}
verify_backup_family(
&migration.db_path,
&quarantine_dir,
&receipt.raw_live_before_undo,
)?;
if logical_witness(&migration.db_path)? != *logical_restored
|| !stable_raw_eq(&raw_family_witness(&migration.db_path)?, raw_restored)
{
return Err(BeadsError::internal(format!(
"schema migration run {} was previously undone, but the live database family has since changed",
args.run_id
)));
}
Ok(())
}
fn validate_prepared_undo_receipt(
receipt: &UndoReceipt,
args: &DoctorMigrateSchemaUndoArgs,
migration: &MigrationContext,
run_dir: &Path,
applied_receipt_sha256: &str,
applied: &AppliedMigrationReceipt,
) -> Result<()> {
validate_raw_family_witness(&receipt.raw_expected_before)?;
validate_raw_family_witness(&receipt.raw_live_before_undo)?;
if receipt.schema_version != UNDO_SCHEMA
|| receipt.run_id != args.run_id
|| receipt.dry_run
|| receipt.database_path != migration.db_path.display().to_string()
|| !constant_time_text_eq(&receipt.applied_receipt_sha256, applied_receipt_sha256)
|| receipt.raw_expected_before != applied.raw_before
|| receipt.logical_expected_before != applied.logical_before
|| receipt.raw_restored.is_some()
|| receipt.logical_restored.is_some()
{
return Err(BeadsError::internal(format!(
"schema migration run {} has an invalid or inconsistent prepared undo receipt",
args.run_id
)));
}
if let Some(expected) = applied.logical_after.as_ref() {
if receipt.logical_live_before_undo.as_ref() != Some(expected) {
return Err(BeadsError::internal(format!(
"schema migration run {} prepared undo does not bind the applied logical state",
args.run_id
)));
}
} else if let Some(expected) = applied.raw_after.as_ref()
&& !stable_raw_eq(&receipt.raw_live_before_undo, expected)
{
return Err(BeadsError::internal(format!(
"schema migration run {} prepared undo does not bind the fallback applied raw state",
args.run_id
)));
}
let _ = validate_quarantine_path(receipt, run_dir)?;
Ok(())
}
fn require_unchanged_applied_state(
applied: &AppliedMigrationReceipt,
raw_live: &RawFamilyWitness,
logical_live: Option<&LogicalDatabaseWitness>,
run_id: &str,
projection_unchanged: bool,
) -> Result<()> {
let unchanged = if let Some(expected) = applied.logical_after.as_ref() {
logical_live == Some(expected)
} else {
applied
.raw_after
.as_ref()
.is_some_and(|expected| stable_raw_eq(raw_live, expected))
};
if !unchanged || !projection_unchanged {
return Err(BeadsError::internal(format!(
"schema migration undo refused because the live database has changed since run \
{run_id}; preserving both states is safer than overwriting newer tracker work"
)));
}
Ok(())
}
fn validate_quarantine_path(receipt: &UndoReceipt, run_dir: &Path) -> Result<PathBuf> {
let quarantine_dir = PathBuf::from(&receipt.quarantine_path);
let expected_parent = run_dir.join("undo-quarantine");
if quarantine_dir.parent() != Some(expected_parent.as_path()) {
return Err(BeadsError::internal(format!(
"schema migration undo quarantine path escapes its run directory: {}",
quarantine_dir.display()
)));
}
let quarantine_id = quarantine_dir
.file_name()
.and_then(|value| value.to_str())
.ok_or_else(|| BeadsError::internal("schema migration undo quarantine id is not UTF-8"))?;
validate_run_id(quarantine_id)?;
Ok(quarantine_dir)
}
fn emit_undo(receipt: &UndoReceipt, json: bool) -> Result<()> {
if json {
println!(
"{}",
serde_json::to_string_pretty(receipt).map_err(BeadsError::Json)?
);
} else if receipt.dry_run {
println!(
"Undo preconditions verified for schema migration run {}",
receipt.run_id
);
println!(
"Current state would be quarantined at {}",
receipt.quarantine_path
);
} else {
println!("Undid schema migration run {}", receipt.run_id);
println!(
"Displaced applied state retained at {}",
receipt.quarantine_path
);
}
Ok(())
}
fn current_schema_version() -> Result<u32> {
u32::try_from(CURRENT_SCHEMA_VERSION).map_err(|_| {
BeadsError::internal(format!(
"current schema version {CURRENT_SCHEMA_VERSION} cannot be represented as u32"
))
})
}
fn open_read_only(path: &Path) -> Result<Connection> {
open_with_flags(
path.to_string_lossy().as_ref(),
OpenFlags::SQLITE_OPEN_READ_ONLY,
)
.map_err(BeadsError::Database)
}
fn close_connection(conn: Connection) -> Result<()> {
conn.close().map_err(BeadsError::Database)
}
fn query_user_version(conn: &Connection) -> Result<u32> {
let row = conn.query_row("PRAGMA user_version")?;
row.get(0)
.and_then(SqliteValue::as_integer)
.and_then(|value| u32::try_from(value).ok())
.ok_or_else(|| BeadsError::internal("PRAGMA user_version was not a nonnegative u32"))
}
fn named_table_exists(conn: &Connection, table: &str) -> Result<bool> {
let escaped = table.replace('\'', "''");
let rows = conn.query(&format!(
"SELECT 1 FROM sqlite_master WHERE type='table' AND name='{escaped}' LIMIT 1"
))?;
Ok(!rows.is_empty())
}
fn require_source_tables(conn: &Connection, from: u32) -> Result<()> {
for table in ["issues", "dirty_issues", "export_hashes"] {
if !named_table_exists(conn, table)? {
return Err(BeadsError::internal(format!(
"reviewed {from}->{} migration requires table {table}, but it is absent",
CURRENT_SCHEMA_VERSION
)));
}
}
Ok(())
}
fn query_count(conn: &Connection, sql: &str) -> Result<u64> {
let row = conn.query_row(sql)?;
row.get(0)
.and_then(SqliteValue::as_integer)
.and_then(|value| u64::try_from(value).ok())
.ok_or_else(|| {
BeadsError::internal(format!("count query returned an invalid value: {sql}"))
})
}
fn current_runtime_shape_is_canonical(db_path: &Path) -> Result<bool> {
let conn = open_read_only(db_path)?;
let canonical =
query_user_version(&conn)? == current_schema_version()? && runtime_schema_compatible(&conn);
close_connection(conn)?;
Ok(canonical)
}
const LEGACY_V15_PROFILE: &str = "legacy-v15-typed-dependencies-null-dirty-hashes-v1";
const LEGACY_V15_NEW_TABLES: &[&str] = &[
"capacity_exemption_history",
"capacity_exemptions",
"capacity_occupancy",
];
fn legacy_projection_rowid(conn: &Connection, table: &str) -> Result<Option<&'static str>> {
if table == "sqlite_sequence" {
return Ok(None);
}
let escaped = table.replace('\'', "''");
let rows = conn.query(&format!("PRAGMA main.table_list('{escaped}')"))?;
let row = rows
.iter()
.find(|row| row.get(1).and_then(SqliteValue::as_text) == Some(table))
.ok_or_else(|| BeadsError::internal(format!("table_list omitted {table}")))?;
if row.get(2).and_then(SqliteValue::as_text) != Some("table") {
return Err(BeadsError::Config(format!(
"legacy v15 conversion refused: {table} is not an ordinary table"
)));
}
match row.get(4).and_then(SqliteValue::as_integer) {
Some(1) => return Ok(None),
Some(0) => {}
_ => return Err(BeadsError::internal("table_list omitted its rowid mode")),
}
let columns = conn.query(&format!("PRAGMA main.table_xinfo('{escaped}')"))?;
if columns.is_empty() {
return Err(BeadsError::internal(format!("table_xinfo omitted {table}")));
}
["rowid", "_rowid_", "oid"]
.into_iter()
.find(|alias| {
!columns.iter().any(|column| {
column
.get(1)
.and_then(SqliteValue::as_text)
.is_some_and(|name| name.eq_ignore_ascii_case(alias))
})
})
.map(Some)
.ok_or_else(|| {
BeadsError::Config(format!(
"legacy v15 conversion refused: {table} shadows every hidden rowid alias"
))
})
}
fn legacy_table_projection(
conn: &Connection,
table: &str,
project_source: bool,
) -> Result<LegacyTableProjection> {
let rowid = legacy_projection_rowid(conn, table)?;
let prefix = rowid.map_or_else(String::new, |alias| format!("{alias}, "));
let columns = match (project_source, table) {
(_, "sqlite_sequence") => "name, seq",
(true, "issues") => "*, ''",
(true, "dirty_issues") => "issue_id, marked_at",
(true, "child_counters") => "parent_id, last_child",
_ => "*",
};
let mut rows: Vec<Vec<SqliteValue>> = conn
.query(&format!(
"SELECT {prefix}{columns} FROM main.{}",
quote_identifier(table)
))?
.into_iter()
.map(|row| row.values().to_vec())
.collect();
if project_source && table == "sqlite_sequence" {
for name in ["comments", "events"] {
let maximum = conn
.query_row(&format!("SELECT MAX(id) FROM main.{name}"))?
.get(0)
.and_then(SqliteValue::as_integer)
.unwrap_or(0)
.max(0);
rows.push(vec![name.into(), maximum.into()]);
}
}
legacy_rows_projection(table, rowid.is_some(), &rows)
}
fn legacy_rows_projection(
table: &str,
rowid_included: bool,
rows: &[Vec<SqliteValue>],
) -> Result<LegacyTableProjection> {
let mut encoded_rows: Vec<Vec<u8>> = rows
.iter()
.map(|row| {
let mut encoded = Vec::new();
for value in row {
encode_sqlite_value(&mut encoded, value);
}
encoded
})
.collect();
encoded_rows.sort();
let mut hasher = Sha256::new();
hash_len_prefixed(&mut hasher, table.as_bytes());
for row in encoded_rows {
hash_len_prefixed(&mut hasher, &row);
}
Ok(LegacyTableProjection {
table: LogicalTableWitness {
name: table.to_owned(),
row_count: u64::try_from(rows.len())
.map_err(|_| BeadsError::internal("projection row count does not fit u64"))?,
rows_sha256: hex_digest(hasher.finalize().as_slice()),
},
rowid_included,
})
}
fn legacy_preserved_schema_witness(conn: &Connection) -> Result<String> {
let mut hasher = Sha256::new();
for row in conn.query(
"SELECT type, name, tbl_name, COALESCE(sql, '') FROM main.sqlite_schema \
ORDER BY type, name, tbl_name, COALESCE(sql, '')",
)? {
let object_type = row.get(0).and_then(SqliteValue::as_text).unwrap_or("");
let name = row.get(1).and_then(SqliteValue::as_text).unwrap_or("");
let table = row.get(2).and_then(SqliteValue::as_text).unwrap_or("");
if matches!(
table,
"issues"
| "dependencies"
| "comments"
| "events"
| "dirty_issues"
| "export_hashes"
| "child_counters"
) || LEGACY_V15_NEW_TABLES.contains(&table)
|| REVIEWED_LEGACY_V15_RETIRED_INDEXES.contains(&name)
|| (object_type == "index" && is_reviewed_migration_managed_index(table, name))
{
continue;
}
for value in row.values().iter().take(3) {
hash_sqlite_value(&mut hasher, value);
}
let sql = row.get(3).and_then(SqliteValue::as_text).unwrap_or("");
hash_len_prefixed(&mut hasher, ddl_token_fingerprint(sql, true).as_bytes());
}
Ok(hex_digest(hasher.finalize().as_slice()))
}
fn legacy_v15_projection(conn: &Connection, project_source: bool) -> Result<LegacyV15Projection> {
if !project_source {
for row in conn.query("SELECT name FROM main.sqlite_schema WHERE type='index'")? {
if let Some(name) = row.get(0).and_then(SqliteValue::as_text)
&& REVIEWED_LEGACY_V15_RETIRED_INDEXES.contains(&name)
{
return Err(BeadsError::Config(format!(
"legacy v15 conversion left retired index {name} in the destination"
)));
}
}
}
let mut tables = Vec::new();
for row in conn.query("SELECT name FROM main.sqlite_schema WHERE type='table' ORDER BY name")? {
let name = row
.get(0)
.and_then(SqliteValue::as_text)
.ok_or_else(|| BeadsError::internal("projection table name is not text"))?;
tables.push(legacy_table_projection(conn, name, project_source)?);
}
if project_source {
for name in LEGACY_V15_NEW_TABLES {
tables.push(legacy_rows_projection(name, true, &[])?);
}
}
tables.sort_by(|left, right| left.table.name.cmp(&right.table.name));
Ok(LegacyV15Projection {
profile: LEGACY_V15_PROFILE.to_owned(),
tables,
preserved_schema_sha256: legacy_preserved_schema_witness(conn)?,
retired_indexes: REVIEWED_LEGACY_V15_RETIRED_INDEXES
.iter()
.map(|name| (*name).to_owned())
.collect(),
})
}
fn verify_legacy_v15_projection(
db_path: &Path,
expected: Option<&LegacyV15Projection>,
project_source: bool,
) -> Result<()> {
let Some(expected) = expected else {
return Ok(());
};
let conn = open_read_only(db_path)?;
let observed = legacy_v15_projection(&conn, project_source);
let closed = close_connection(conn);
let observed = observed?;
closed?;
if &observed != expected {
return Err(BeadsError::internal(format!(
"legacy v15 conversion projection mismatch at {}; refusing to install or \
overwrite this database (expected={expected:?}, observed={observed:?})",
db_path.display()
)));
}
Ok(())
}
fn logical_witness(db_path: &Path) -> Result<LogicalDatabaseWitness> {
let conn = open_read_only(db_path)?;
let integrity_before = match integrity_check_messages(&conn) {
Ok(messages) => messages.join("\n"),
Err(error) => {
let _ = close_connection(conn);
return Err(error);
}
};
if let Err(error) = conn.execute("BEGIN DEFERRED TRANSACTION") {
let _ = close_connection(conn);
return Err(BeadsError::Database(error));
}
let result = logical_witness_from_connection(&conn, integrity_before.clone());
let transaction_result = conn
.execute("ROLLBACK")
.map(|_| ())
.map_err(BeadsError::Database);
let integrity_after = if transaction_result.is_ok() {
integrity_check_messages(&conn).map(|messages| messages.join("\n"))
} else {
Ok(String::new())
};
let close_result = close_connection(conn);
let mut witness = result?;
transaction_result?;
let integrity_after = integrity_after?;
close_result?;
if integrity_before != integrity_after {
return Err(BeadsError::internal(format!(
"database integrity changed while capturing the migration witness \
(before={integrity_before:?}, after={integrity_after:?})"
)));
}
witness.integrity_check = integrity_after;
Ok(witness)
}
fn logical_witness_from_connection(
conn: &Connection,
integrity_check: String,
) -> Result<LogicalDatabaseWitness> {
let user_version = query_user_version(conn)?;
let schema_rows = conn.query(
"SELECT type, name, tbl_name, COALESCE(sql, '') \
FROM sqlite_master \
ORDER BY type, name, tbl_name, COALESCE(sql, '')",
)?;
let mut schema_hasher = Sha256::new();
for row in &schema_rows {
for index in 0..3 {
hash_sqlite_value(
&mut schema_hasher,
row.get(index).unwrap_or(&SqliteValue::Null),
);
}
let ddl = row.get(3).and_then(SqliteValue::as_text).unwrap_or("");
hash_len_prefixed(
&mut schema_hasher,
ddl_token_fingerprint(ddl, false).as_bytes(),
);
}
let schema_sha256 = hex_digest(schema_hasher.finalize().as_slice());
let table_rows = conn.query(
"SELECT name FROM sqlite_master \
WHERE type='table' \
ORDER BY name",
)?;
let mut tables = Vec::with_capacity(table_rows.len());
let mut contents_hasher = Sha256::new();
for table_row in table_rows {
let name = table_row
.get(0)
.and_then(SqliteValue::as_text)
.ok_or_else(|| BeadsError::internal("sqlite_master table name was not text"))?
.to_string();
let quoted = quote_identifier(&name);
let columns = conn.query(&format!("PRAGMA table_info({quoted})"))?.len();
let rows = conn.query(&format!("SELECT * FROM {quoted}"))?;
let mut encoded_rows = Vec::with_capacity(rows.len());
for row in &rows {
let mut encoded = Vec::new();
for index in 0..columns {
encode_sqlite_value(&mut encoded, row.get(index).unwrap_or(&SqliteValue::Null));
}
encoded_rows.push(encoded);
}
encoded_rows.sort();
let mut table_hasher = Sha256::new();
hash_len_prefixed(&mut table_hasher, name.as_bytes());
for encoded in &encoded_rows {
hash_len_prefixed(&mut table_hasher, encoded);
hash_len_prefixed(&mut contents_hasher, name.as_bytes());
hash_len_prefixed(&mut contents_hasher, encoded);
}
tables.push(LogicalTableWitness {
name,
row_count: u64::try_from(encoded_rows.len()).map_err(|_| {
BeadsError::internal("logical witness row count does not fit in u64")
})?,
rows_sha256: hex_digest(table_hasher.finalize().as_slice()),
});
}
Ok(LogicalDatabaseWitness {
user_version,
integrity_check,
schema_sha256,
contents_sha256: hex_digest(contents_hasher.finalize().as_slice()),
tables,
})
}
fn integrity_check_messages(conn: &Connection) -> Result<Vec<String>> {
let rows = conn.query("PRAGMA integrity_check")?;
let mut messages = Vec::new();
for row in rows {
for value in row.values() {
if let Some(text) = value.as_text() {
let trimmed = text.trim();
if !trimmed.is_empty() {
messages.push(trimmed.to_string());
}
}
}
}
if messages.is_empty() {
messages.push("integrity_check returned no diagnostic rows".to_string());
}
Ok(messages)
}
fn integrity_check_is_clean(integrity_check: &str) -> bool {
integrity_check.trim().eq_ignore_ascii_case("ok")
}
fn admit_preexisting_index_damage(
probe_db: &Path,
rehearsal_dir: &Path,
expected: &LogicalDatabaseWitness,
policy: PreexistingIndexCorruption,
) -> Result<IndexCorruptionFinding> {
let summary = integrity_summary(&expected.integrity_check);
if !index_rebuild_restores_integrity(probe_db, rehearsal_dir, expected)? {
return Err(BeadsError::internal(format!(
"private engine recovery did not produce clean integrity ({summary}), and \
rebuilding every index on a private copy did not repair it, so the damage is not \
confined to indexes. The live database was not touched; run `br doctor` for \
diagnosis"
)));
}
let finding = IndexCorruptionFinding {
integrity_check: expected.integrity_check.clone(),
remediation: IndexCorruptionFinding::REMEDIATION,
index_entries_without_table_rows: integrity_reports_index_entries_without_table_rows(
&expected.integrity_check,
),
};
match policy {
PreexistingIndexCorruption::Refuse => {
let warning = if finding.index_entries_without_table_rows {
format!(" {}", IndexCorruptionFinding::MISSING_ROWS_WARNING)
} else {
String::new()
};
Err(BeadsError::internal(format!(
"private engine recovery did not produce clean integrity because the database \
already has corrupt indexes ({summary}). A private rehearsal showed that \
rebuilding them repairs it without changing any row. The live database was not \
touched. Run `br doctor migrate-schema recover` to restore the WAL index, then \
`{}` to rebuild the indexes.{warning}",
IndexCorruptionFinding::REMEDIATION
)))
}
PreexistingIndexCorruption::Admit => Ok(finding),
}
}
fn integrity_summary(integrity_check: &str) -> String {
const SHOWN: usize = 3;
let messages = integrity_check
.lines()
.map(str::trim)
.filter(|line| !line.is_empty() && !line.starts_with("*** in database"))
.collect::<Vec<_>>();
let mut summary = messages
.iter()
.take(SHOWN)
.copied()
.collect::<Vec<_>>()
.join("; ");
if messages.len() > SHOWN {
summary = format!("{summary}; and {} more", messages.len() - SHOWN);
}
summary
}
fn index_rebuild_restores_integrity(
db_path: &Path,
rehearsal_dir: &Path,
before: &LogicalDatabaseWitness,
) -> Result<bool> {
ensure_new_directory(rehearsal_dir)?;
let family = recovery_family_witness(db_path)?;
copy_family_to_backup(db_path, rehearsal_dir, &family)?;
verify_backup_family(db_path, rehearsal_dir, &family)?;
let copy = backup_component_path(rehearsal_dir, db_path, "")?;
let conn = Connection::open(copy.to_string_lossy().into_owned())?;
let rebuilt = (|| -> Result<()> {
let names = conn
.query(
"SELECT name FROM sqlite_master WHERE type = 'index' \
AND name NOT LIKE 'sqlite_autoindex_%' AND sql IS NOT NULL ORDER BY name",
)?
.iter()
.filter_map(|row| row.get(0).and_then(SqliteValue::as_text).map(String::from))
.collect::<Vec<_>>();
conn.execute("BEGIN IMMEDIATE")?;
for name in &names {
if let Err(error) = conn.execute(&format!("REINDEX {}", quote_identifier(name))) {
let _ = conn.execute("ROLLBACK");
return Err(error.into());
}
}
conn.execute("COMMIT")?;
Ok(())
})();
let closed = close_connection(conn);
let after = rebuilt.and(closed).and_then(|()| logical_witness(©));
match after {
Ok(after) => Ok(integrity_check_is_clean(&after.integrity_check)
&& logical_witnesses_match_except_integrity(before, &after)),
Err(error) => {
tracing::warn!(
%error,
copy = %copy.display(),
"rebuilding indexes on a private recovery copy failed"
);
Ok(false)
}
}
}
fn integrity_check_is_repairable(integrity_check: &str) -> bool {
let mut saw_repairable = false;
for message in integrity_check
.lines()
.map(str::trim)
.filter(|line| !line.is_empty())
{
let lower = message.to_ascii_lowercase();
if lower.contains("never used")
|| lower.contains("missing from index")
|| lower.contains("out of order")
{
saw_repairable = true;
continue;
}
if lower.contains("*** in database") {
continue;
}
return false;
}
saw_repairable
}
fn quote_identifier(identifier: &str) -> String {
format!("\"{}\"", identifier.replace('"', "\"\""))
}
fn ddl_token_fingerprint(sql: &str, preserve_grouping: bool) -> String {
fn flush(word: &mut String, tokens: &mut Vec<String>) {
if !word.is_empty() {
tokens.push(word.to_ascii_lowercase());
word.clear();
}
}
let mut tokens: Vec<String> = Vec::new();
let mut word = String::new();
let mut chars = sql.chars().peekable();
while let Some(c) = chars.next() {
match c {
'-' if chars.peek() == Some(&'-') => {
flush(&mut word, &mut tokens);
for next in chars.by_ref() {
if next == '\n' {
break;
}
}
}
'/' if chars.peek() == Some(&'*') => {
flush(&mut word, &mut tokens);
chars.next();
let mut previous = '\0';
for next in chars.by_ref() {
if previous == '*' && next == '/' {
break;
}
previous = next;
}
}
'\'' => {
flush(&mut word, &mut tokens);
let mut literal = String::from("'");
loop {
match chars.next() {
Some('\'') if chars.peek() == Some(&'\'') => {
literal.push_str("''");
chars.next();
}
Some('\'') | None => {
literal.push('\'');
break;
}
Some(inner) => literal.push(inner),
}
}
tokens.push(literal);
}
'"' | '`' | '[' if preserve_grouping => {
flush(&mut word, &mut tokens);
let closing = if c == '[' { ']' } else { c };
let mut identifier = String::new();
while let Some(inner) = chars.next() {
if inner == closing {
if chars.peek() == Some(&closing) {
identifier.push(inner);
chars.next();
} else {
break;
}
} else {
identifier.push(inner);
}
}
if fsqlite_ast::quote_ident_if_needed(&identifier) == identifier {
tokens.push(identifier);
} else {
tokens.push(format!("quoted:{}", hex_digest(identifier.as_bytes())));
}
}
'(' | ')' if preserve_grouping => {
flush(&mut word, &mut tokens);
tokens.push(c.to_string());
}
c if c.is_whitespace() || matches!(c, '"' | '`' | '[' | ']' | '(' | ')' | ';') => {
flush(&mut word, &mut tokens);
}
c if c.is_alphanumeric() || matches!(c, '_' | '$' | '.') => word.push(c),
other => {
flush(&mut word, &mut tokens);
tokens.push(other.to_string());
}
}
}
flush(&mut word, &mut tokens);
tokens.join(" ")
}
fn hash_sqlite_value(hasher: &mut Sha256, value: &SqliteValue) {
let mut encoded = Vec::new();
encode_sqlite_value(&mut encoded, value);
hash_len_prefixed(hasher, &encoded);
}
fn encode_sqlite_value(output: &mut Vec<u8>, value: &SqliteValue) {
match value {
SqliteValue::Null => output.push(0),
SqliteValue::Integer(value) => {
output.push(1);
output.extend_from_slice(&value.to_le_bytes());
}
SqliteValue::Float(value) => {
output.push(2);
output.extend_from_slice(&value.to_bits().to_le_bytes());
}
SqliteValue::Text(value) => {
output.push(3);
append_len_prefixed(output, value.as_bytes());
}
SqliteValue::Blob(value) => {
output.push(4);
append_len_prefixed(output, value.as_ref());
}
}
}
fn append_len_prefixed(output: &mut Vec<u8>, bytes: &[u8]) {
output.extend_from_slice(&u64::try_from(bytes.len()).unwrap_or(u64::MAX).to_le_bytes());
output.extend_from_slice(bytes);
}
fn hash_len_prefixed(hasher: &mut Sha256, bytes: &[u8]) {
hasher.update(u64::try_from(bytes.len()).unwrap_or(u64::MAX).to_le_bytes());
hasher.update(bytes);
}
fn raw_family_witness(db_path: &Path) -> Result<RawFamilyWitness> {
raw_family_witness_for_suffixes(db_path, FAMILY_SUFFIXES.iter().copied())
}
fn raw_family_witness_for_suffixes<'a>(
db_path: &Path,
suffixes: impl Iterator<Item = &'a str>,
) -> Result<RawFamilyWitness> {
let mut components = Vec::with_capacity(FAMILY_SUFFIXES.len());
for suffix in suffixes {
let path = family_component_path(db_path, suffix);
match secure_file_metadata(&path)? {
Some(metadata) => {
let (length, sha256) = hash_regular_file(&path, &metadata)?;
components.push(RawComponentWitness {
suffix: suffix.to_string(),
present: true,
length: Some(length),
sha256: Some(sha256),
unix_mode: unix_file_mode(&metadata),
});
}
None => components.push(RawComponentWitness {
suffix: suffix.to_string(),
present: false,
length: None,
sha256: None,
unix_mode: None,
}),
}
}
Ok(RawFamilyWitness { components })
}
fn family_component_path(db_path: &Path, suffix: &str) -> PathBuf {
let mut value = OsString::from(db_path.as_os_str());
value.push(suffix);
PathBuf::from(value)
}
fn backup_component_path(before_dir: &Path, db_path: &Path, suffix: &str) -> Result<PathBuf> {
let file_name = db_path.file_name().ok_or_else(|| {
BeadsError::internal(format!(
"database path has no file name: {}",
db_path.display()
))
})?;
let mut backup_name = OsString::from(file_name);
backup_name.push(suffix);
Ok(before_dir.join(backup_name))
}
fn secure_file_metadata(path: &Path) -> Result<Option<fs::Metadata>> {
match fs::symlink_metadata(path) {
Ok(metadata) => {
if metadata.file_type().is_symlink() {
return Err(BeadsError::internal(format!(
"refusing schema migration file-family symlink {}",
path.display()
)));
}
if !metadata.is_file() {
return Err(BeadsError::internal(format!(
"schema migration family member is not a regular file: {}",
path.display()
)));
}
Ok(Some(metadata))
}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
Err(error) => Err(BeadsError::Io(error)),
}
}
fn refuse_non_regular_component(db_path: &Path) -> Result<()> {
if secure_file_metadata(db_path)?.is_none() {
return Err(BeadsError::DatabaseNotFound {
path: db_path.to_path_buf(),
});
}
for suffix in &FAMILY_SUFFIXES[1..] {
let _ = secure_file_metadata(&family_component_path(db_path, suffix))?;
}
Ok(())
}
fn hash_regular_file(path: &Path, expected: &fs::Metadata) -> Result<(u64, String)> {
let mut file = File::open(path).map_err(BeadsError::Io)?;
let opened = file.metadata().map_err(BeadsError::Io)?;
if !same_file_identity(expected, &opened) {
return Err(BeadsError::internal(format!(
"schema migration file changed identity while opening {}",
path.display()
)));
}
let mut hasher = Sha256::new();
let mut length = 0_u64;
let mut buffer = vec![0_u8; 1024 * 1024];
loop {
let read = file.read(&mut buffer).map_err(BeadsError::Io)?;
if read == 0 {
break;
}
length = length
.checked_add(u64::try_from(read).unwrap_or(u64::MAX))
.ok_or_else(|| BeadsError::internal("schema migration file length overflow"))?;
hasher.update(&buffer[..read]);
}
Ok((length, hex_digest(hasher.finalize().as_slice())))
}
#[cfg(unix)]
fn same_file_identity(expected: &fs::Metadata, opened: &fs::Metadata) -> bool {
use std::os::unix::fs::MetadataExt;
expected.dev() == opened.dev()
&& expected.ino() == opened.ino()
&& expected.len() == opened.len()
}
#[cfg(not(unix))]
fn same_file_identity(expected: &fs::Metadata, opened: &fs::Metadata) -> bool {
expected.len() == opened.len()
}
#[cfg(unix)]
#[allow(clippy::unnecessary_wraps)]
fn unix_file_mode(metadata: &fs::Metadata) -> Option<u32> {
use std::os::unix::fs::PermissionsExt;
Some(metadata.permissions().mode())
}
#[cfg(not(unix))]
fn unix_file_mode(_metadata: &fs::Metadata) -> Option<u32> {
None
}
fn hex_digest(bytes: &[u8]) -> String {
const HEX: &[u8; 16] = b"0123456789abcdef";
let mut output = String::with_capacity(bytes.len() * 2);
for byte in bytes {
output.push(char::from(HEX[usize::from(byte >> 4)]));
output.push(char::from(HEX[usize::from(byte & 0x0f)]));
}
output
}
fn compute_plan_token(
database_path: &str,
logical: &LogicalDatabaseWitness,
forecast: &MigrationForecast,
) -> Result<String> {
let material = PlanTokenMaterial {
contract: PLAN_SCHEMA,
database_path,
from_version: forecast.from_version,
to_version: forecast.to_version,
logical_witness: logical,
forecast,
};
let encoded = serde_json::to_vec(&material).map_err(BeadsError::Json)?;
Ok(hex_digest(Sha256::digest(encoded).as_slice()))
}
fn constant_time_text_eq(left: &str, right: &str) -> bool {
let left = left.as_bytes();
let right = right.as_bytes();
let mut difference = left.len() ^ right.len();
let length = left.len().max(right.len());
for index in 0..length {
let left_byte = left.get(index).copied().unwrap_or(0);
let right_byte = right.get(index).copied().unwrap_or(0);
difference |= usize::from(left_byte ^ right_byte);
}
difference == 0
}
fn migration_runs_root(beads_dir: &Path) -> PathBuf {
beads_dir.join(".br_recovery").join("schema-migrations")
}
fn allocate_run_id(beads_dir: &Path) -> Result<String> {
let recovery_root = beads_dir.join(".br_recovery");
ensure_directory(&recovery_root)?;
set_private_directory_permissions(&recovery_root)?;
let root = migration_runs_root(beads_dir);
ensure_directory(&root)?;
set_private_directory_permissions(&root)?;
for counter in 0_u32..1000 {
let run_id = format!(
"{}-{}-{counter}",
Utc::now().format("%Y%m%dT%H%M%S%.6fZ"),
std::process::id()
);
let candidate = root.join(&run_id);
match fs::create_dir(&candidate) {
Ok(()) => {
set_private_directory_permissions(&candidate)?;
sync_directory(&root)?;
return Ok(run_id);
}
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {}
Err(error) => return Err(BeadsError::Io(error)),
}
}
Err(BeadsError::internal(
"could not allocate a unique schema-migration run id",
))
}
fn ensure_directory(path: &Path) -> Result<()> {
if path.exists() {
let metadata = fs::symlink_metadata(path).map_err(BeadsError::Io)?;
if metadata.file_type().is_symlink() || !metadata.is_dir() {
return Err(BeadsError::internal(format!(
"schema migration artifact path is not a real directory: {}",
path.display()
)));
}
return Ok(());
}
let parent = path.parent().ok_or_else(|| {
BeadsError::internal(format!(
"schema migration artifact directory has no parent: {}",
path.display()
))
})?;
ensure_directory(parent)?;
match fs::create_dir(path) {
Ok(()) => {
set_private_directory_permissions(path)?;
sync_directory(parent)?;
Ok(())
}
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {
let metadata = fs::symlink_metadata(path).map_err(BeadsError::Io)?;
if metadata.file_type().is_symlink() || !metadata.is_dir() {
Err(BeadsError::internal(format!(
"schema migration artifact path raced to a non-directory: {}",
path.display()
)))
} else {
Ok(())
}
}
Err(error) => Err(BeadsError::Io(error)),
}
}
fn ensure_new_directory(path: &Path) -> Result<()> {
let parent = path.parent().ok_or_else(|| {
BeadsError::internal(format!(
"schema migration directory has no parent: {}",
path.display()
))
})?;
ensure_directory(parent)?;
fs::create_dir(path).map_err(BeadsError::Io)?;
set_private_directory_permissions(path)?;
sync_directory(parent)
}
fn copy_family_to_backup(
db_path: &Path,
before_dir: &Path,
expected: &RawFamilyWitness,
) -> Result<()> {
for component in &expected.components {
if !component.present {
continue;
}
let source = family_component_path(db_path, &component.suffix);
let destination = backup_component_path(before_dir, db_path, &component.suffix)?;
copy_regular_file_new(&source, &destination, None)?;
}
sync_directory(before_dir)
}
fn copy_regular_file_new(
source: &Path,
destination: &Path,
restored_unix_mode: Option<u32>,
) -> Result<()> {
let source_metadata = secure_file_metadata(source)?.ok_or_else(|| {
BeadsError::internal(format!(
"schema migration source disappeared before copy: {}",
source.display()
))
})?;
let mut source_file = File::open(source).map_err(BeadsError::Io)?;
let opened_metadata = source_file.metadata().map_err(BeadsError::Io)?;
if !same_file_identity(&source_metadata, &opened_metadata) {
return Err(BeadsError::internal(format!(
"schema migration source changed identity before copy: {}",
source.display()
)));
}
let mut destination_file = open_private_file_new(destination)?;
std::io::copy(&mut source_file, &mut destination_file).map_err(BeadsError::Io)?;
destination_file.sync_all().map_err(BeadsError::Io)?;
set_file_permissions(destination, restored_unix_mode)?;
Ok(())
}
fn verify_backup_family(
db_path: &Path,
before_dir: &Path,
expected: &RawFamilyWitness,
) -> Result<()> {
for component in &expected.components {
let backup = backup_component_path(before_dir, db_path, &component.suffix)?;
let metadata = secure_file_metadata(&backup)?;
if !component.present {
if metadata.is_some() {
return Err(BeadsError::internal(format!(
"unexpected backup exists for absent family member {}",
backup.display()
)));
}
continue;
}
let metadata = metadata.ok_or_else(|| {
BeadsError::internal(format!(
"required schema migration backup is missing: {}",
backup.display()
))
})?;
let (length, sha256) = hash_regular_file(&backup, &metadata)?;
if component.length != Some(length) || component.sha256.as_deref() != Some(&sha256) {
return Err(BeadsError::internal(format!(
"schema migration backup hash mismatch for {}",
backup.display()
)));
}
}
Ok(())
}
fn write_json_new<T: Serialize>(path: &Path, value: &T) -> Result<()> {
let encoded = serde_json::to_vec_pretty(value).map_err(BeadsError::Json)?;
let parent = path.parent().ok_or_else(|| {
BeadsError::internal(format!(
"schema migration receipt path has no parent: {}",
path.display()
))
})?;
let file_name = path.file_name().ok_or_else(|| {
BeadsError::internal(format!(
"schema migration receipt path has no file name: {}",
path.display()
))
})?;
let nonce = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map_err(|error| {
BeadsError::internal(format!("system clock precedes Unix epoch: {error}"))
})?
.as_nanos();
let mut staged = None;
for attempt in 0_u16..=u16::MAX {
let mut staged_name = OsString::from(".");
staged_name.push(file_name);
staged_name.push(format!(
".write-{}-{nonce}-{attempt}.tmp",
std::process::id()
));
let staged_path = parent.join(staged_name);
match open_private_file_new(&staged_path) {
Ok(file) => {
staged = Some((staged_path, file));
break;
}
Err(BeadsError::Io(error)) if error.kind() == std::io::ErrorKind::AlreadyExists => {}
Err(error) => return Err(error),
}
}
let (staged_path, mut file) = staged.ok_or_else(|| {
BeadsError::internal(format!(
"could not allocate a unique staged receipt beside {}",
path.display()
))
})?;
file.write_all(&encoded).map_err(BeadsError::Io)?;
file.write_all(b"\n").map_err(BeadsError::Io)?;
file.sync_all().map_err(BeadsError::Io)?;
drop(file);
rename_path_no_replace(&staged_path, path)?;
sync_directory(parent)?;
Ok(())
}
fn file_sha256(path: &Path) -> Result<String> {
let metadata = secure_file_metadata(path)?.ok_or_else(|| {
BeadsError::internal(format!(
"schema migration artifact does not exist: {}",
path.display()
))
})?;
let (_, sha256) = hash_regular_file(path, &metadata)?;
Ok(sha256)
}
fn read_json<T: for<'de> Deserialize<'de>>(path: &Path) -> Result<T> {
let metadata = secure_file_metadata(path)?.ok_or_else(|| {
BeadsError::internal(format!(
"schema migration receipt does not exist: {}",
path.display()
))
})?;
let (length, _) = hash_regular_file(path, &metadata)?;
if length > 16 * 1024 * 1024 {
return Err(BeadsError::internal(format!(
"schema migration receipt exceeds 16 MiB: {}",
path.display()
)));
}
let bytes = fs::read(path).map_err(BeadsError::Io)?;
serde_json::from_slice(&bytes).map_err(BeadsError::Json)
}
fn stable_raw_eq(left: &RawFamilyWitness, right: &RawFamilyWitness) -> bool {
left.components
.iter()
.filter(|component| component.suffix != "-shm")
.eq(right
.components
.iter()
.filter(|component| component.suffix != "-shm"))
}
fn quarantine_live_family_resuming(
db_path: &Path,
quarantine_dir: &Path,
applied: &RawFamilyWitness,
restored: &RawFamilyWitness,
) -> Result<()> {
for component in &applied.components {
let source = family_component_path(db_path, &component.suffix);
let destination = backup_component_path(quarantine_dir, db_path, &component.suffix)?;
let source_metadata = secure_file_metadata(&source)?;
let destination_metadata = secure_file_metadata(&destination)?;
if component.present {
if let Some(metadata) = destination_metadata {
verify_component_bytes(&destination, &metadata, component)?;
if let Some(source_metadata) = source_metadata {
let restored_component = component_for_suffix(restored, &component.suffix)?;
if !restored_component.present {
return Err(BeadsError::internal(format!(
"schema migration undo found both quarantined and unexpected live \
copies for {}",
source.display()
)));
}
verify_component_bytes(&source, &source_metadata, restored_component)?;
}
continue;
}
let source_metadata = source_metadata.ok_or_else(|| {
BeadsError::internal(format!(
"schema migration undo cannot resume because both live and quarantined \
applied components are missing for {}",
source.display()
))
})?;
verify_component_bytes(&source, &source_metadata, component)?;
rename_path_no_replace(&source, &destination)?;
let quarantined_metadata = secure_file_metadata(&destination)?.ok_or_else(|| {
BeadsError::internal(format!(
"schema migration undo component disappeared after quarantine: {}",
destination.display()
))
})?;
verify_component_bytes(&destination, &quarantined_metadata, component)?;
set_file_permissions(&destination, None)?;
continue;
}
if destination_metadata.is_some() {
return Err(BeadsError::internal(format!(
"schema migration undo quarantine contains an unexpected component {}",
destination.display()
)));
}
if let Some(source_metadata) = source_metadata {
let restored_component = component_for_suffix(restored, &component.suffix)?;
if !restored_component.present {
return Err(BeadsError::internal(format!(
"schema migration undo found an unexpected live component {}",
source.display()
)));
}
verify_component_bytes(&source, &source_metadata, restored_component)?;
}
}
if let Some(parent) = db_path.parent() {
sync_directory(parent)?;
}
sync_directory(quarantine_dir)
}
fn restore_backup_family_resuming(
db_path: &Path,
before_dir: &Path,
expected: &RawFamilyWitness,
) -> Result<()> {
for component in &expected.components {
let destination = family_component_path(db_path, &component.suffix);
let destination_metadata = secure_file_metadata(&destination)?;
if !component.present {
if destination_metadata.is_some() {
return Err(BeadsError::internal(format!(
"schema migration restore found a live component that should be absent: {}",
destination.display()
)));
}
continue;
}
let source = backup_component_path(before_dir, db_path, &component.suffix)?;
if let Some(metadata) = destination_metadata {
verify_component_bytes(&destination, &metadata, component)?;
set_file_permissions(&destination, component.unix_mode)?;
} else {
copy_regular_file_new(&source, &destination, component.unix_mode)?;
}
}
if let Some(parent) = db_path.parent() {
sync_directory(parent)?;
}
Ok(())
}
fn component_for_suffix<'a>(
family: &'a RawFamilyWitness,
suffix: &str,
) -> Result<&'a RawComponentWitness> {
family
.components
.iter()
.find(|component| component.suffix == suffix)
.ok_or_else(|| {
BeadsError::internal(format!(
"schema migration witness omits required family suffix {suffix:?}"
))
})
}
fn validate_raw_family_witness(family: &RawFamilyWitness) -> Result<()> {
if family.components.len() != FAMILY_SUFFIXES.len() {
return Err(BeadsError::internal(format!(
"schema migration raw witness has {} components, expected {}",
family.components.len(),
FAMILY_SUFFIXES.len()
)));
}
for (component, expected_suffix) in family.components.iter().zip(FAMILY_SUFFIXES) {
let has_valid_hash = component.sha256.as_deref().is_some_and(|hash| {
hash.len() == 64 && hash.bytes().all(|byte| byte.is_ascii_hexdigit())
});
if component.suffix != *expected_suffix
|| (component.present && (component.length.is_none() || !has_valid_hash))
|| (!component.present
&& (component.length.is_some()
|| component.sha256.is_some()
|| component.unix_mode.is_some()))
{
return Err(BeadsError::internal(format!(
"schema migration raw witness has an invalid component for suffix \
{expected_suffix:?}"
)));
}
}
Ok(())
}
fn verify_component_bytes(
path: &Path,
metadata: &fs::Metadata,
expected: &RawComponentWitness,
) -> Result<()> {
let (length, sha256) = hash_regular_file(path, metadata)?;
if !expected.present
|| expected.length != Some(length)
|| expected.sha256.as_deref() != Some(&sha256)
{
return Err(BeadsError::internal(format!(
"schema migration component does not match its receipt: {}",
path.display()
)));
}
Ok(())
}
fn validate_run_id(run_id: &str) -> Result<()> {
if run_id.is_empty()
|| run_id == "."
|| run_id == ".."
|| run_id.contains('/')
|| run_id.contains('\\')
{
return Err(BeadsError::internal(format!(
"invalid schema migration run id {run_id:?}"
)));
}
Ok(())
}
#[cfg(unix)]
fn open_private_file_new(path: &Path) -> Result<File> {
use std::os::unix::fs::OpenOptionsExt;
OpenOptions::new()
.create_new(true)
.write(true)
.mode(0o600)
.open(path)
.map_err(BeadsError::Io)
}
#[cfg(not(unix))]
fn open_private_file_new(path: &Path) -> Result<File> {
OpenOptions::new()
.create_new(true)
.write(true)
.open(path)
.map_err(BeadsError::Io)
}
#[cfg(unix)]
fn set_private_directory_permissions(path: &Path) -> Result<()> {
use std::os::unix::fs::PermissionsExt;
fs::set_permissions(path, fs::Permissions::from_mode(0o700)).map_err(BeadsError::Io)
}
#[cfg(not(unix))]
fn set_private_directory_permissions(_path: &Path) -> Result<()> {
Ok(())
}
#[cfg(unix)]
fn set_file_permissions(path: &Path, restored_unix_mode: Option<u32>) -> Result<()> {
use std::os::unix::fs::PermissionsExt;
let mode = restored_unix_mode.unwrap_or(0o600) & 0o7777;
fs::set_permissions(path, fs::Permissions::from_mode(mode)).map_err(BeadsError::Io)
}
#[cfg(not(unix))]
fn set_file_permissions(_path: &Path, _restored_unix_mode: Option<u32>) -> Result<()> {
Ok(())
}
#[cfg(unix)]
fn sync_directory(path: &Path) -> Result<()> {
File::open(path)
.and_then(|file| file.sync_all())
.map_err(BeadsError::Io)
}
#[cfg(not(unix))]
fn sync_directory(_path: &Path) -> Result<()> {
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use tempfile::TempDir;
#[test]
fn recovery_556_evidence_follows_nondefault_database_parent() {
let root = tempfile::tempdir().unwrap();
let ambient = root.path().join("ambient/.beads");
let selected = root.path().join("external");
fs::create_dir_all(&ambient).unwrap();
fs::create_dir_all(&selected).unwrap();
let directory = recovery_artifacts_dir(&selected.join("nondefault.db")).unwrap();
assert_eq!(directory, fs::canonicalize(&selected).unwrap());
assert_ne!(directory, fs::canonicalize(&ambient).unwrap());
assert!(!ambient.join(".br_recovery").exists());
}
#[test]
fn recovery_556_bare_relative_filename_uses_current_directory_without_chdir() {
let directory = recovery_artifacts_dir(Path::new("nondefault.db")).unwrap();
assert_eq!(directory, fs::canonicalize(".").unwrap());
}
#[test]
fn recovery_556_evidence_resolution_does_not_create_a_missing_database() {
let root = tempfile::tempdir().unwrap();
let selected = root.path().join("missing.db");
assert_eq!(
recovery_artifacts_dir(&selected).unwrap(),
fs::canonicalize(root.path()).unwrap()
);
assert!(!selected.exists());
}
#[test]
fn recovery_556_frame_refusal_is_a_conflict_not_corruption() {
let root = tempfile::tempdir().unwrap();
let db = root.path().join("framed.db");
fs::write(family_component_path(&db, "-wal"), [0_u8; 33]).unwrap();
let error = require_frame_free_engine_recovery(&db).unwrap_err();
assert!(matches!(error, BeadsError::SyncConflict { .. }));
assert!(!db.exists());
assert!(!family_component_path(&db, "-shm").exists());
}
#[test]
fn private_index_engine_never_needs_wal_index_recovery_gh520() {
let temp = TempDir::new().unwrap();
let db = temp.path().join("beads.db");
crate::franken_sync::wal_index::tests::write_stock_header_only_family(&db);
assert!(!wal_index_needs_recovery_for_engine(&db, true).unwrap());
assert!(!wal_index_needs_recovery_for_engine(&db, false).unwrap());
assert!(!wal_index_needs_recovery(&db).unwrap());
let shm = family_component_path(&db, "-shm");
fs::rename(&shm, temp.path().join("retained-shm")).unwrap();
assert!(missing_wal_index_for_engine(&db, true).unwrap());
assert!(!missing_wal_index_for_engine(&db, false).unwrap());
assert!(wal_index_needs_recovery_for_engine(&db, true).unwrap());
assert!(!wal_index_needs_recovery_for_engine(&db, false).unwrap());
assert_eq!(
missing_wal_index(&db).unwrap(),
crate::franken_sync::wal_index::STARTUP_WAL_INDEX_RECOVERY
);
}
fn make_run(root: &Path, name: &str, database_path: &str, receipts: &[&str]) -> PathBuf {
make_run_with_scope(
root,
name,
database_path,
receipts,
receipts.contains(&RECOVERY_SUPERSEDED_RECEIPT),
)
}
fn make_run_with_scope(
root: &Path,
name: &str,
database_path: &str,
receipts: &[&str],
index_only: bool,
) -> PathBuf {
let run = root.join(name);
let before = run.join("recovery-before");
fs::create_dir_all(&before).unwrap();
let database = Path::new(database_path);
let mut raw = RawFamilyWitness {
components: Vec::new(),
};
for suffix in std::iter::once("").chain(config::db_sidecar_suffixes().copied()) {
let bytes: Option<&[u8]> = match suffix {
"" => Some(b"retained"),
"-wal" => Some(&[7; 32]),
"-shm" => Some(&[8; 96]),
_ => None,
};
raw.components.push(RawComponentWitness {
suffix: suffix.to_string(),
present: bytes.is_some(),
length: bytes.map(|bytes| u64::try_from(bytes.len()).unwrap()),
sha256: bytes.map(|bytes| hex_digest(Sha256::digest(bytes).as_slice())),
unix_mode: None,
});
if let Some(bytes) = bytes {
if !index_only || suffix == "-shm" {
fs::write(
backup_component_path(&before, database, suffix).unwrap(),
bytes,
)
.unwrap();
}
if !index_only {
let probe = run.join("recovery-probe");
fs::create_dir_all(&probe).unwrap();
fs::write(
backup_component_path(&probe, database, suffix).unwrap(),
bytes,
)
.unwrap();
}
}
}
let logical = serde_json::json!({
"user_version": 17,
"integrity_check": "ok",
"schema_sha256": "a".repeat(64),
"contents_sha256": "b".repeat(64),
"tables": [],
});
for name in receipts {
let prepared = *name == "recovery-prepared.json";
let stage = match *name {
"recovery-prepared.json" if index_only => "live-preflight",
"recovery-prepared.json" => "private-recovery",
"recovery-complete.json" => "complete",
RECOVERY_SUPERSEDED_RECEIPT => "superseded",
_ => "failed",
};
fs::write(
run.join(name),
serde_json::json!({
"schema_version": "br.doctor.schema_migration.recovery.v1",
"database_path": database_path,
"backup_path": before,
"backup_scope": if index_only { RECOVERY_BACKUP_WAL_INDEX_ONLY } else { RECOVERY_BACKUP_COMPLETE_FAMILY },
"stage": stage,
"raw_before": raw,
"raw_after": if prepared { serde_json::Value::Null } else { serde_json::to_value(&raw).unwrap() },
"logical_after": if prepared { serde_json::Value::Null } else { logical.clone() },
"error": null,
}).to_string(),
)
.unwrap();
}
run
}
#[test]
fn recovery_retention_admits_the_probe_quarantine_of_a_finished_run_gh545() {
use crate::franken_sync::wal_index::tests::write_quarantine_layout;
let temp = TempDir::new().unwrap();
let beads_dir = temp.path().join(".beads");
let db_path = beads_dir.join("beads.db");
let database = db_path.display().to_string();
let root = migration_runs_root(&beads_dir);
fs::create_dir_all(&root).unwrap();
let now = Utc::now();
let receipts = ["recovery-prepared.json", "recovery-complete.json"];
let status = |run: &Path| recovery_run_status(run, run, &db_path).unwrap();
let run = |counter| make_run(&root, &run_name(now, 30, counter), &database, &receipts);
let probe = |run: &Path| run.join("recovery-probe");
let plain = run(0);
assert_eq!(status(&plain), Some(RecoveryRunStatus::Finished));
let quarantined = make_run(&root, &run_name(now, 40, 1), &database, &receipts);
write_quarantine_layout(&probe(&quarantined), "probe", "beads.db", true);
assert_eq!(status(&quarantined), Some(RecoveryRunStatus::Finished));
let unfinished = run(2);
write_quarantine_layout(&probe(&unfinished), "probe", "beads.db", false);
assert_eq!(status(&unfinished), Some(RecoveryRunStatus::Retained));
let two = run(3);
write_quarantine_layout(&probe(&two), "a", "beads.db", true);
write_quarantine_layout(&probe(&two), "b", "beads.db", true);
assert_eq!(status(&two), Some(RecoveryRunStatus::Retained));
let foreign = run(4);
write_quarantine_layout(&probe(&foreign), "probe", "other.db", true);
assert_eq!(status(&foreign), Some(RecoveryRunStatus::Retained));
let stray = run(5);
fs::create_dir(probe(&stray).join("stray")).unwrap();
assert_eq!(status(&stray), Some(RecoveryRunStatus::Retained));
let pruned = prune_recovery_runs(&beads_dir, &db_path, &plain, now).unwrap();
assert!(
pruned
.iter()
.any(|path| path.ends_with(quarantined.file_name().unwrap())),
"{pruned:?}"
);
assert!(!quarantined.exists());
for kept in [&unfinished, &two, &foreign, &stray] {
assert!(kept.exists(), "{}", kept.display());
}
}
fn run_name(now: chrono::DateTime<Utc>, days_ago: i64, counter: u32) -> String {
format!(
"{}-4242-{counter}",
(now - chrono::Duration::days(days_ago)).format("%Y%m%dT%H%M%S%.6fZ")
)
}
#[test]
fn recovery_run_names_carry_their_creation_time() {
let run_id = allocate_run_id(TempDir::new().unwrap().path()).unwrap();
let created = recovery_run_timestamp(&run_id).expect("allocate_run_id format parses");
assert!((Utc::now() - created).num_seconds().abs() < 60, "{run_id}");
assert!(recovery_run_timestamp("not-a-run").is_none());
assert!(recovery_run_timestamp("20269999T000000.000000Z-1-0").is_none());
}
#[test]
#[allow(clippy::too_many_lines)]
fn recovery_retention_prunes_only_old_finished_runs_of_this_database() {
let temp = TempDir::new().unwrap();
let beads_dir = temp.path().join(".beads");
let db_path = beads_dir.join("beads.db");
let database = db_path.display().to_string();
let root = migration_runs_root(&beads_dir);
fs::create_dir_all(&root).unwrap();
let now = Utc::now();
let prepared = "recovery-prepared.json";
let complete = "recovery-complete.json";
let current = make_run(
&root,
&run_name(now, 0, 0),
&database,
&[prepared, complete],
);
let recent: Vec<PathBuf> = (1..=3)
.map(|day| {
make_run(
&root,
&run_name(now, day, 0),
&database,
&[prepared, complete],
)
})
.collect();
let superseded_recent = make_run(
&root,
&run_name(now, 4, 0),
&database,
&[prepared, RECOVERY_SUPERSEDED_RECEIPT],
);
let young = make_run(
&root,
&run_name(now, 5, 0),
&database,
&[prepared, complete],
);
let old_finished: Vec<PathBuf> = (10..=12)
.map(|day| {
make_run(
&root,
&run_name(now, day, 0),
&database,
&[prepared, complete],
)
})
.collect();
let old_superseded = make_run(
&root,
&run_name(now, 13, 0),
&database,
&[prepared, RECOVERY_SUPERSEDED_RECEIPT],
);
let old_failed = make_run(
&root,
&run_name(now, 20, 0),
&database,
&[prepared, "recovery-failed.json"],
);
let old_in_progress = make_run(&root, &run_name(now, 21, 0), &database, &[prepared]);
let old_other_database = make_run(
&root,
&run_name(now, 22, 0),
"/elsewhere/.beads/beads.db",
&[prepared, complete],
);
let old_migration = make_run(
&root,
&run_name(now, 23, 0),
&database,
&["prepared.json", "applied.json"],
);
let unreadable_receipt = root.join(run_name(now, 24, 0));
fs::create_dir_all(&unreadable_receipt).unwrap();
fs::write(unreadable_receipt.join(prepared), b"{ not json").unwrap();
fs::write(unreadable_receipt.join(complete), b"{}").unwrap();
let unnamed = make_run(&root, "operator-kept-run", &database, &[prepared, complete]);
let original = make_run(
&root,
&run_name(now, 30, 0),
&database,
&[prepared, complete],
);
let interrupted = root.join(format!("{PRUNING_PREFIX}{}", run_name(now, 30, 0)));
fs::rename(&original, &interrupted).unwrap();
let mut pruned = prune_recovery_runs(&beads_dir, &db_path, ¤t, now).unwrap();
pruned.sort();
let mut expected: Vec<PathBuf> = old_finished.clone();
expected.sort();
assert_eq!(pruned, expected);
for gone in old_finished.iter().chain([&interrupted]) {
assert!(!gone.exists(), "{} must be removed", gone.display());
}
for kept in recent.iter().chain([
¤t,
&superseded_recent,
&young,
&old_superseded,
&old_failed,
&old_in_progress,
&old_other_database,
&old_migration,
&unreadable_receipt,
&unnamed,
]) {
assert!(kept.is_dir(), "{} must be kept", kept.display());
}
let leftovers: Vec<String> = fs::read_dir(&root)
.unwrap()
.map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned())
.filter(|name| name.starts_with(PRUNING_PREFIX))
.collect();
assert!(leftovers.is_empty(), "{leftovers:?}");
assert!(
prune_recovery_runs(&beads_dir, &db_path, ¤t, now)
.unwrap()
.is_empty()
);
}
#[test]
fn recovery_retention_never_prunes_the_current_run_or_within_the_newest_five() {
let temp = TempDir::new().unwrap();
let beads_dir = temp.path().join(".beads");
let db_path = beads_dir.join("beads.db");
let database = db_path.display().to_string();
let root = migration_runs_root(&beads_dir);
fs::create_dir_all(&root).unwrap();
let now = Utc::now();
let runs: Vec<PathBuf> = (0..7)
.map(|index| {
make_run(
&root,
&run_name(now, 30 + index, 0),
&database,
&["recovery-prepared.json", "recovery-complete.json"],
)
})
.collect();
let current = runs.last().unwrap().clone();
let pruned = prune_recovery_runs(&beads_dir, &db_path, ¤t, now).unwrap();
assert_eq!(pruned, vec![runs[5].clone()]);
for (index, run) in runs.iter().enumerate() {
assert_eq!(run.is_dir(), index != 5, "run {index}");
}
let empty = TempDir::new().unwrap();
assert!(
prune_recovery_runs(empty.path(), &db_path, ¤t, now)
.unwrap()
.is_empty()
);
}
#[test]
fn recovery_retention_keeps_staged_runs_within_the_retained_window() {
let now = Utc::now();
for (count, staged_index, first_age, keep) in [
(7, 0, 30, true),
(7, 4, 30, true),
(7, 6, 30, true),
(7, 5, 0, true),
(5, 3, 30, true),
(7, 5, 30, false),
] {
let temp = TempDir::new().unwrap();
let beads_dir = temp.path().join(".beads");
let db_path = beads_dir.join("beads.db");
let database = db_path.display().to_string();
let root = migration_runs_root(&beads_dir);
let runs: Vec<PathBuf> = (0..count)
.map(|index| {
make_run(
&root,
&run_name(now, first_age + index, 0),
&database,
&["recovery-prepared.json", "recovery-complete.json"],
)
})
.collect();
let current = runs.last().unwrap();
let original = &runs[staged_index];
let staged = root.join(format!(
"{PRUNING_PREFIX}{}",
original.file_name().unwrap().to_string_lossy()
));
fs::rename(original, &staged).unwrap();
prune_recovery_runs(&beads_dir, &db_path, current, now).unwrap();
assert_eq!(
staged.is_dir(),
keep,
"count={count}, staged_index={staged_index}, first_age={first_age}"
);
for (index, run) in runs.iter().take(RECOVERY_RUNS_KEPT).enumerate() {
if index != staged_index {
assert!(run.is_dir(), "newest run {index} must be retained");
}
}
assert!(if original == current {
staged.is_dir()
} else {
current.is_dir()
});
}
}
#[test]
fn recovery_retention_keeps_unproven_verdicts_and_foreign_staged_evidence() {
let temp = TempDir::new().unwrap();
let beads_dir = temp.path().join(".beads");
let db_path = beads_dir.join("beads.db");
let database = db_path.display().to_string();
let root = migration_runs_root(&beads_dir);
let now = Utc::now();
let receipts = ["recovery-prepared.json", "recovery-complete.json"];
let current = make_run(&root, &run_name(now, 0, 0), &database, &receipts);
for day in 1..5 {
make_run(&root, &run_name(now, day, 0), &database, &receipts);
}
let eligible = make_run(&root, &run_name(now, 10, 0), &database, &receipts);
let mut kept = Vec::new();
for (counter, marker) in [(0, ""), (1, "{\"stage\":"), (2, "{}")] {
let run = make_run(&root, &run_name(now, 20, counter), &database, &receipts);
fs::write(run.join("recovery-complete.json"), marker).unwrap();
kept.push(run);
}
for (counter, owner, failed) in [
(0, "/another/tracker/beads.db", false),
(1, database.as_str(), true),
] {
let original_name = run_name(now, 30, counter);
let run = make_run(&root, &original_name, owner, &receipts);
if failed {
fs::write(
run.join("recovery-failed.json"),
b"retained failure evidence",
)
.unwrap();
}
let staged = root.join(format!("{PRUNING_PREFIX}{original_name}"));
fs::rename(run, &staged).unwrap();
kept.push(staged);
}
let empty_staged = root.join(format!("{PRUNING_PREFIX}{}", run_name(now, 40, 0)));
fs::create_dir_all(&empty_staged).unwrap();
kept.push(empty_staged);
let retained: Vec<(PathBuf, Vec<u8>)> = kept
.iter()
.flat_map(|run| {
[
"recovery-prepared.json",
"recovery-complete.json",
"recovery-failed.json",
"recovery-before/beads.db",
]
.into_iter()
.filter_map(|name| {
let path = run.join(name);
fs::read(&path).ok().map(|bytes| (path, bytes))
})
})
.collect();
let pruned = prune_recovery_runs(&beads_dir, &db_path, ¤t, now).unwrap();
assert!(pruned.contains(&eligible));
assert!(
!eligible.exists(),
"an old verified complete run must still be pruned"
);
let lost: Vec<_> = kept.iter().filter(|run| !run.is_dir()).collect();
assert!(
lost.is_empty(),
"unproven or foreign recovery evidence was deleted: {lost:?}"
);
assert_eq!(pruned, vec![eligible]);
for (path, bytes) in retained {
assert_eq!(
fs::read(&path).unwrap(),
bytes,
"{} must stay byte-identical",
path.display()
);
}
}
#[test]
#[allow(clippy::too_many_lines)]
fn recovery_retention_requires_agreeing_receipts_and_intact_owned_payloads() {
let temp = TempDir::new().unwrap();
let beads_dir = temp.path().join(".beads");
let db_path = beads_dir.join("beads.db");
let database = db_path.display().to_string();
let root = migration_runs_root(&beads_dir);
let now = Utc::now();
let receipts = ["recovery-prepared.json", "recovery-complete.json"];
let current = make_run(&root, &run_name(now, 0, 0), &database, &receipts);
for day in 1..5 {
make_run(&root, &run_name(now, day, 0), &database, &receipts);
}
let mut eligible = Vec::new();
for (counter, legacy) in [(0, false), (1, true)] {
let run = make_run_with_scope(
&root,
&run_name(now, 10, counter),
&database,
&receipts,
true,
);
if legacy {
let path = run.join("recovery-prepared.json");
let mut receipt: serde_json::Value =
serde_json::from_slice(&fs::read(&path).unwrap()).unwrap();
receipt["stage"] = "live-recovery".into();
fs::write(path, receipt.to_string()).unwrap();
}
eligible.push(run);
}
let mut kept = Vec::new();
for (counter, (field, value)) in [
(
"schema_version",
serde_json::json!("br.doctor.schema_migration.recovery.v2"),
),
("stage", serde_json::json!("live-verification")),
("database_path", serde_json::json!("/another/beads.db")),
("backup_path", serde_json::json!("/another/recovery-before")),
("backup_scope", serde_json::json!("wal-index-only")),
("raw_before", serde_json::json!({"components": []})),
("raw_after", serde_json::Value::Null),
("logical_after", serde_json::Value::Null),
("error", serde_json::json!("verification failed")),
("failure_kind", serde_json::Value::Null),
(
"index_corruption",
serde_json::json!({"integrity_check": "damaged"}),
),
("unreviewed", serde_json::json!(true)),
]
.into_iter()
.enumerate()
{
let run = make_run(
&root,
&run_name(now, 20, u32::try_from(counter).unwrap()),
&database,
&receipts,
);
let path = run.join("recovery-complete.json");
let mut receipt: serde_json::Value =
serde_json::from_slice(&fs::read(&path).unwrap()).unwrap();
receipt[field] = value;
fs::write(path, receipt.to_string()).unwrap();
kept.push(run);
}
for (counter, relative, bytes) in [
(0, "recovery-before/beads.db", "changed original backup"),
(1, "recovery-probe/beads.db-wal", "changed private WAL"),
(
2,
"recovery-before/operator-evidence",
"unowned backup evidence",
),
(
3,
"recovery-probe/operator-evidence",
"unowned private evidence",
),
(
4,
".recovery-complete.json.write-interrupted.tmp",
"unfinished publication",
),
(5, "recovery-superseded.json", "contradictory verdict"),
] {
let run = make_run(&root, &run_name(now, 30, counter), &database, &receipts);
fs::write(run.join(relative), bytes).unwrap();
kept.push(run);
}
let duplicate = make_run(&root, &run_name(now, 40, 0), &database, &receipts);
let path = duplicate.join("recovery-complete.json");
let marker = fs::read_to_string(&path).unwrap();
fs::write(path, marker.replacen('{', "{\"stage\":\"failed\",", 1)).unwrap();
kept.push(duplicate);
let missing = make_run(&root, &run_name(now, 40, 1), &database, &receipts);
let path = missing.join("recovery-prepared.json");
let mut marker: serde_json::Value =
serde_json::from_slice(&fs::read(&path).unwrap()).unwrap();
marker.as_object_mut().unwrap().remove("raw_after");
fs::write(path, marker.to_string()).unwrap();
kept.push(missing);
let damaged = make_run(&root, &run_name(now, 40, 2), &database, &receipts);
let path = damaged.join("recovery-complete.json");
let mut marker: serde_json::Value =
serde_json::from_slice(&fs::read(&path).unwrap()).unwrap();
marker["logical_after"]["integrity_check"] = "damaged rows".into();
fs::write(path, marker.to_string()).unwrap();
kept.push(damaged);
let mut pruned = prune_recovery_runs(&beads_dir, &db_path, ¤t, now).unwrap();
pruned.sort();
eligible.sort();
assert_eq!(
pruned, eligible,
"both current and exact legacy complete index-only runs still qualify"
);
for run in kept {
assert!(
run.is_dir(),
"unproven evidence must remain: {}",
run.display()
);
}
}
#[test]
fn recovery_retention_preserves_staging_collisions_and_replaced_directories() {
let temp = TempDir::new().unwrap();
let beads_dir = temp.path().join(".beads");
let db_path = beads_dir.join("beads.db");
let database = db_path.display().to_string();
let root = migration_runs_root(&beads_dir);
let now = Utc::now();
let receipts = ["recovery-prepared.json", "recovery-complete.json"];
let current = make_run(&root, &run_name(now, 0, 0), &database, &receipts);
for day in 1..5 {
make_run(&root, &run_name(now, day, 0), &database, &receipts);
}
let name = run_name(now, 20, 0);
let original = make_run(&root, &name, &database, &receipts);
let collision = root.join(format!("{PRUNING_PREFIX}{name}"));
fs::create_dir(&collision).unwrap();
let collision_directory = open_recovery_pruning_entry(&collision, true)
.unwrap()
.unwrap();
assert!(
prune_recovery_runs(&beads_dir, &db_path, ¤t, now)
.unwrap()
.is_empty()
);
assert!(original.is_dir());
assert!(
recovery_pruning_entry_unchanged(&collision, &collision_directory, true).unwrap(),
"an empty staging destination must not be overwritten"
);
let name = run_name(now, 30, 0);
let replaced = make_run(&root, &name, &database, &receipts);
let observed = open_recovery_pruning_entry(&replaced, true)
.unwrap()
.unwrap();
let retained = temp.path().join("retained-previous-directory");
fs::rename(&replaced, &retained).unwrap();
make_run(&root, &name, &database, &receipts);
assert!(!remove_verified_recovery_run(&replaced, &replaced, &db_path, &observed).unwrap());
assert!(retained.is_dir());
assert!(
replaced.is_dir(),
"a different directory generation must not inherit deletion authority"
);
let observed = open_recovery_pruning_entry(&replaced, true)
.unwrap()
.unwrap();
fs::write(replaced.join("recovery-complete.json"), b"{}").unwrap();
assert!(!remove_verified_recovery_run(&replaced, &replaced, &db_path, &observed).unwrap());
assert!(
replaced.is_dir(),
"completion must be rechecked after discovery"
);
}
#[test]
fn recovery_retention_rejects_nested_unknown_duplicate_and_missing_proof() {
let temp = TempDir::new().unwrap();
let beads_dir = temp.path().join(".beads");
let db_path = beads_dir.join("beads.db");
let database = db_path.display().to_string();
let root = migration_runs_root(&beads_dir);
let now = Utc::now();
let receipts = ["recovery-prepared.json", "recovery-complete.json"];
let current = make_run(&root, &run_name(now, 0, 0), &database, &receipts);
for day in 1..5 {
make_run(&root, &run_name(now, day, 0), &database, &receipts);
}
let mut kept = Vec::new();
for counter in 0..6 {
let run = make_run(&root, &run_name(now, 20, counter), &database, &receipts);
let path = run.join("recovery-complete.json");
let mut marker: serde_json::Value =
serde_json::from_slice(&fs::read(&path).unwrap()).unwrap();
match counter {
0 => marker["raw_before"]["unreviewed"] = true.into(),
1 => marker["raw_after"]["components"][0]["unreviewed"] = true.into(),
2 => {
marker["raw_after"]["components"][0]
.as_object_mut()
.unwrap()
.remove("unix_mode");
}
3 => marker["logical_after"]["unreviewed"] = true.into(),
4 => {
marker["logical_after"]["tables"] = serde_json::json!([{
"name": "issues", "row_count": 1, "rows_sha256": "c".repeat(64), "unreviewed": true,
}]);
}
_ => {}
}
let mut bytes = marker.to_string();
if counter == 5 {
bytes = bytes.replacen(
"\"integrity_check\":\"ok\"",
"\"integrity_check\":\"damaged\",\"integrity_check\":\"ok\"",
1,
);
}
fs::write(path, bytes).unwrap();
kept.push(run);
}
assert!(
prune_recovery_runs(&beads_dir, &db_path, ¤t, now)
.unwrap()
.is_empty()
);
for run in kept {
assert!(
run.is_dir(),
"nested unproven evidence must remain: {}",
run.display()
);
}
}
#[cfg(unix)]
#[test]
fn recovery_retention_refuses_aliased_receipts_and_payloads() {
let temp = TempDir::new().unwrap();
let beads_dir = temp.path().join(".beads");
let db_path = beads_dir.join("beads.db");
let database = db_path.display().to_string();
let root = migration_runs_root(&beads_dir);
let now = Utc::now();
let receipts = ["recovery-prepared.json", "recovery-complete.json"];
let current = make_run(&root, &run_name(now, 0, 0), &database, &receipts);
for day in 1..5 {
make_run(&root, &run_name(now, day, 0), &database, &receipts);
}
let mut retained = Vec::new();
for (counter, relative, hard_link) in [
(0, "recovery-complete.json", false),
(1, "recovery-before/beads.db", false),
(2, "recovery-complete.json", true),
(3, "recovery-before/beads.db", true),
] {
let run = make_run(&root, &run_name(now, 20, counter), &database, &receipts);
let path = run.join(relative);
let outside = temp.path().join(format!("retained-alias-target-{counter}"));
fs::rename(&path, &outside).unwrap();
if hard_link {
fs::hard_link(&outside, &path).unwrap();
} else {
std::os::unix::fs::symlink(&outside, &path).unwrap();
}
let bytes = fs::read(&outside).unwrap();
retained.push((run, path, outside, bytes));
}
assert!(
prune_recovery_runs(&beads_dir, &db_path, ¤t, now)
.unwrap()
.is_empty()
);
for (run, path, outside, bytes) in retained {
assert!(
run.is_dir(),
"aliased evidence must remain: {}",
run.display()
);
assert_eq!(fs::read(&path).unwrap(), bytes);
assert_eq!(fs::read(&outside).unwrap(), bytes);
}
}
#[cfg(unix)]
#[test]
fn recovery_retention_ignores_symlinked_runs() {
let temp = TempDir::new().unwrap();
let beads_dir = temp.path().join(".beads");
let db_path = beads_dir.join("beads.db");
let database = db_path.display().to_string();
let root = migration_runs_root(&beads_dir);
fs::create_dir_all(&root).unwrap();
let now = Utc::now();
let outside = make_run(
temp.path(),
"outside-target",
&database,
&["recovery-prepared.json", "recovery-complete.json"],
);
for index in 0..6 {
make_run(
&root,
&run_name(now, index, 0),
&database,
&["recovery-prepared.json", "recovery-complete.json"],
);
}
let link = root.join(run_name(now, 40, 0));
std::os::unix::fs::symlink(&outside, &link).unwrap();
let current = root.join(run_name(now, 0, 0));
prune_recovery_runs(&beads_dir, &db_path, ¤t, now).unwrap();
assert!(
fs::symlink_metadata(&link)
.unwrap()
.file_type()
.is_symlink()
);
assert!(outside.join("recovery-before").join("beads.db").is_file());
}
#[cfg(any(target_os = "linux", target_os = "android", target_vendor = "apple"))]
#[test]
fn exchange_preflight_accepts_real_exchange_without_touching_database() {
let temp = TempDir::new().unwrap();
let database = temp.path().join("beads.db");
fs::write(&database, b"untouched database").unwrap();
verify_migration_exchange_support(&database, exchange_database_paths).unwrap();
assert_eq!(fs::read(&database).unwrap(), b"untouched database");
assert_eq!(fs::read_dir(temp.path()).unwrap().count(), 1);
}
#[cfg(any(target_os = "linux", target_os = "android", target_vendor = "apple"))]
#[test]
fn exchange_preflight_refuses_successful_ordinary_rename_and_retains_probe() {
let temp = TempDir::new().unwrap();
let database = temp.path().join("beads.db");
fs::write(&database, b"untouched database").unwrap();
let error = verify_migration_exchange_support(&database, |left, right| {
fs::rename(left, right).map_err(BeadsError::Io)
})
.unwrap_err();
assert!(
error
.to_string()
.contains("filesystem exchange preflight failed")
);
assert!(error.to_string().contains("Probe retained"));
assert_eq!(fs::read(&database).unwrap(), b"untouched database");
assert_eq!(fs::read_dir(temp.path()).unwrap().count(), 2);
}
#[cfg(any(target_os = "linux", target_os = "android", target_vendor = "apple"))]
#[test]
fn exchange_preflight_refuses_noop_success() {
let temp = TempDir::new().unwrap();
let database = temp.path().join("beads.db");
fs::write(&database, b"untouched database").unwrap();
let error = verify_migration_exchange_support(&database, |_, _| Ok(())).unwrap_err();
assert!(error.to_string().contains("both locked file identities"));
assert_eq!(fs::read(&database).unwrap(), b"untouched database");
}
#[test]
fn ddl_token_fingerprint_ignores_comments_whitespace_parens_quotes_and_case() {
let canonical = "CREATE TABLE \"labels\" (\n -- who carries the label\n issue_id TEXT \
NOT NULL,\n label TEXT NOT NULL CHECK (label <> ''),\n PRIMARY KEY \
(issue_id, label)\n)";
let reserialized = "create table labels(issue_id text not null, label text not null check \
label <> '', primary key issue_id, label)";
assert_eq!(
ddl_token_fingerprint(canonical, false),
ddl_token_fingerprint(reserialized, false)
);
assert_ne!(
ddl_token_fingerprint(canonical, false),
ddl_token_fingerprint(
"CREATE TABLE labels (issue_id TEXT NOT NULL, label TEXT)",
false
),
"a real schema difference must still change the fingerprint"
);
assert_ne!(
ddl_token_fingerprint("CHECK (status IN ('Open'))", false),
ddl_token_fingerprint("CHECK (status IN ('open'))", false),
"string literals keep their case"
);
assert_eq!(
ddl_token_fingerprint("DEFAULT 'it''s' /* block\ncomment */ NOT NULL", false),
"default 'it''s' not null"
);
}
#[test]
fn legacy_schema_fingerprint_keeps_operator_grouping_and_compound_identifiers() {
for (before, changed) in [
(
"CREATE TABLE operator(a, b, c, CHECK(a * (b + c) > 0))",
"CREATE TABLE operator(a, b, c, CHECK(a * b + c > 0))",
),
(
"CREATE TABLE operator(\"a b\" TEXT)",
"CREATE TABLE operator(a b TEXT)",
),
(
"CREATE TABLE operator(\"null\", CHECK(\"null\" IS NULL))",
"CREATE TABLE operator(\"null\", CHECK(null IS NULL))",
),
(
"CREATE TABLE operator(\"1\", CHECK(\"1\" > 0))",
"CREATE TABLE operator(\"1\", CHECK(1 > 0))",
),
(
"CREATE TABLE operator(\"current_timestamp\", CHECK(\"current_timestamp\" IS NULL))",
"CREATE TABLE operator(\"current_timestamp\", CHECK(current_timestamp IS NULL))",
),
(
"CREATE TABLE operator(a, CHECK(\"ALLOWED\" = 'ALLOWED'))",
"CREATE TABLE operator(a, CHECK(\"allowed\" = 'ALLOWED'))",
),
] {
assert_eq!(
ddl_token_fingerprint(before, false),
ddl_token_fingerprint(changed, false),
"control: the inherited layout fingerprint misses this distinction"
);
assert_ne!(
ddl_token_fingerprint(before, true),
ddl_token_fingerprint(changed, true),
"the new conversion must bind the actual operator schema"
);
}
assert_eq!(
ddl_token_fingerprint("CREATE TABLE \"operator\" (\"a b\" TEXT)", true),
ddl_token_fingerprint("create table operator ([a b] text)", true),
"equivalent quoting and keyword case can still be reserialized"
);
}
#[test]
fn recovery_532_classifies_only_typed_live_opener_contention() {
use fsqlite_error::FrankenError;
let busy_errors = [
BeadsError::Database(FrankenError::Busy),
BeadsError::Database(FrankenError::BusyRecovery),
BeadsError::Database(FrankenError::DatabaseLocked {
path: PathBuf::from("beads.db"),
}),
BeadsError::DatabaseLocked {
path: PathBuf::from("beads.db"),
},
BeadsError::WithContext {
context: "opening the live family".to_string(),
source: Box::new(BeadsError::Database(FrankenError::BusyRecovery)),
},
BeadsError::WithContext {
context: "opening the live family".to_string(),
source: Box::new(FrankenError::Busy),
},
];
for error in &busy_errors {
assert_eq!(
RecoveryFailureKind::for_error("live-recovery", error),
RecoveryFailureKind::TransientContention,
);
for stage in [
"private-recovery",
"live-preflight",
"live-verification",
"reinstate-index",
] {
assert_eq!(
RecoveryFailureKind::for_error(stage, error),
RecoveryFailureKind::FailClosed,
"{stage}: {error}",
);
}
}
for error in [
BeadsError::Database(FrankenError::DatabaseCorrupt {
detail: "database is busy (recovery in progress)".to_string(),
}),
BeadsError::Database(FrankenError::BusySnapshot {
conflicting_pages: "1".to_string(),
}),
BeadsError::Io(std::io::Error::from(std::io::ErrorKind::WouldBlock)),
BeadsError::internal("database is busy (recovery in progress)"),
BeadsError::SyncConflict {
message: "busy".to_string(),
},
BeadsError::CommittedStateUnwitnessed {
operation: "recovery".to_string(),
source: Box::new(FrankenError::BusyRecovery),
},
BeadsError::WithContext {
context: "recovery".to_string(),
source: Box::new(BeadsError::CommittedArtifactFailure {
operation: "recovery".to_string(),
primary_path: PathBuf::from("beads.db"),
artifact_path: PathBuf::from("receipt.json"),
source: Box::new(FrankenError::Busy),
}),
},
] {
assert_eq!(
RecoveryFailureKind::for_error("live-recovery", &error),
RecoveryFailureKind::FailClosed,
);
}
}
#[test]
fn recovery_532_legacy_and_unknown_receipts_are_narrowly_classified() {
let mut receipt = serde_json::json!({
"schema_version": "br.doctor.schema_migration.recovery.v1",
"backup_scope": "wal-index-only",
"backup_path": "retained",
"stage": "live-recovery",
"raw_before": {"components": []},
"raw_after": null,
"logical_after": null,
"error": "Database error: database is busy (recovery in progress)",
});
let transient = |value: &serde_json::Value| {
serde_json::from_value::<PriorRecoveryFailure>(value.clone())
.expect("parse receipt")
.is_transient_contention()
};
assert!(
transient(&receipt),
"the exact 0.7.4 receipt must not latch"
);
receipt["error"] = "Database error: database is busy".into();
assert!(transient(&receipt));
for error in [
"Database error: database is busy (recovery in progress); corrupt WAL",
"Database error: database disk image is malformed: busy",
"database is busy (recovery in progress)",
] {
receipt["error"] = error.into();
assert!(!transient(&receipt));
}
receipt["error"] = "Database error: database is busy (recovery in progress)".into();
for kind in [
serde_json::json!("fail-closed"),
serde_json::json!("future-kind"),
serde_json::Value::Null,
serde_json::json!(true),
serde_json::json!({"retryable": true}),
] {
receipt["failure_kind"] = kind;
assert!(
!transient(&receipt),
"unknown classification must keep the latch"
);
}
receipt["failure_kind"] = "transient-contention".into();
assert!(transient(&receipt));
for stage in [
"private-recovery",
"live-preflight",
"live-verification",
"reinstate-index",
] {
receipt["stage"] = stage.into();
assert!(!transient(&receipt), "{stage} is not a live open");
receipt.as_object_mut().unwrap().remove("failure_kind");
assert!(
!transient(&receipt),
"legacy {stage} must also remain blocked"
);
receipt["failure_kind"] = "transient-contention".into();
}
}
#[test]
fn recovery_532_legacy_post_open_and_incomplete_receipts_stay_blocked() {
let receipt = serde_json::json!({
"schema_version": "br.doctor.schema_migration.recovery.v1",
"backup_scope": "wal-index-only",
"backup_path": "retained",
"stage": "live-recovery",
"raw_before": {"components": []},
"raw_after": null,
"logical_after": null,
"error": "Database error: database is busy (recovery in progress)",
});
let transient = |value: &serde_json::Value| {
serde_json::from_value::<PriorRecoveryFailure>(value.clone())
.expect("parse receipt")
.is_transient_contention()
};
for scope in ["wal-index-only", "complete-family"] {
let mut valid = receipt.clone();
valid["backup_scope"] = scope.into();
for classified in [false, true] {
if classified {
valid["failure_kind"] = "transient-contention".into();
}
assert!(transient(&valid), "the emitted pre-open shape is retryable");
for field in [
"schema_version",
"backup_scope",
"raw_after",
"logical_after",
] {
let mut incomplete = valid.clone();
incomplete.as_object_mut().unwrap().remove(field);
assert!(
!transient(&incomplete),
"missing {field} must remain blocked"
);
}
for (field, value) in [
("schema_version", serde_json::json!("future-recovery.v2")),
("backup_scope", serde_json::json!("unknown-scope")),
("raw_after", serde_json::json!({"components": []})),
(
"logical_after",
serde_json::json!({"integrity_check": "busy"}),
),
(
"index_corruption",
serde_json::json!({"integrity_check": "damaged"}),
),
] {
let mut after_open = valid.clone();
after_open[field] = value;
assert!(!transient(&after_open), "{field} must retain the latch");
}
}
}
}
#[test]
fn recovery_532_scanner_retains_evidence_and_finds_real_failure_after_busy() {
let temp = TempDir::new().expect("tempdir");
let db = temp.path().join("beads.db");
fs::write(&db, b"witnessed original bytes").unwrap();
let raw = recovery_family_witness(&db).unwrap();
let root = migration_runs_root(temp.path());
let busy_run = root.join("busy");
fs::create_dir_all(busy_run.join("recovery-before")).unwrap();
let backup = busy_run.join("recovery-before/beads.db-shm");
fs::write(&backup, b"retained index evidence").unwrap();
let mut receipt = serde_json::json!({
"schema_version": "br.doctor.schema_migration.recovery.v1",
"backup_scope": "wal-index-only",
"backup_path": busy_run.join("recovery-before"),
"stage": "live-recovery",
"raw_before": raw,
"raw_after": null,
"logical_after": null,
"error": "Database error: database is busy (recovery in progress)",
"failure_kind": "transient-contention",
});
let busy_path = busy_run.join("recovery-failed.json");
write_json_new(&busy_path, &receipt).unwrap();
let retained = fs::read(&busy_path).unwrap();
for _ in 0..3 {
assert!(prior_failed_recovery(temp.path(), &raw).unwrap().is_none());
assert_eq!(fs::read(&busy_path).unwrap(), retained);
assert_eq!(fs::read(&backup).unwrap(), b"retained index evidence");
}
let failed_run = root.join("genuine-failure");
fs::create_dir_all(&failed_run).unwrap();
receipt["failure_kind"] = "fail-closed".into();
receipt["error"] = "protected database component changed".into();
write_json_new(&failed_run.join("recovery-failed.json"), &receipt).unwrap();
let prior = prior_failed_recovery(temp.path(), &raw).unwrap().unwrap();
assert_eq!(
prior.error.as_deref(),
Some("protected database component changed")
);
assert_eq!(fs::read(&busy_path).unwrap(), retained);
fs::write(&db, b"different original bytes").unwrap();
assert!(
prior_failed_recovery(temp.path(), &recovery_family_witness(&db).unwrap())
.unwrap()
.is_none()
);
}
#[test]
fn recovery_532_private_close_contention_is_fail_closed_before_live_install() {
let (_temp, migration) = reviewed_source_migration_context("beads.db", 17);
truncate_fixture_wal(&migration.db_path);
let before = recovery_family_witness(&migration.db_path).unwrap();
let mut stage = "live-preflight".to_string();
crate::franken_sync::tests::FAIL_NEXT_CLOSE.with(|fault| fault.set(true));
let result = recover_existing_family(
&migration.db_path,
Some(&migration.write_authority),
Some(&mut stage),
);
let unconsumed =
crate::franken_sync::tests::FAIL_NEXT_CLOSE.with(|fault| fault.replace(false));
assert!(
!unconsumed,
"the injected error must occur during close, after open"
);
let error = result.unwrap_err();
let BeadsError::WithContext { context, source } = &error else {
panic!("private close failure must retain its rehearsal evidence: {error}");
};
assert!(context.contains("private index rehearsal retained at"));
assert!(matches!(
source.downcast_ref::<BeadsError>(),
Some(BeadsError::Database(
fsqlite_error::FrankenError::BusyRecovery
))
));
assert_eq!(stage, "private-recovery");
assert_eq!(
RecoveryFailureKind::for_error(&stage, &error),
RecoveryFailureKind::FailClosed
);
assert_eq!(recovery_family_witness(&migration.db_path).unwrap(), before);
require_recovery_payload_unchanged(
&before,
&recovery_family_witness(&migration.db_path).unwrap(),
)
.unwrap();
}
#[test]
fn schema_witness_survives_vacuum_into_reserialization() {
let temp = TempDir::new().expect("tempdir");
let db_path = temp.path().join("beads.db");
drop(crate::storage::SqliteStorage::open(&db_path).expect("create the full br schema"));
let source = logical_witness(&db_path).expect("source witness");
let candidate_path = temp.path().join("candidate.db");
let source_conn =
Connection::open(db_path.to_string_lossy().into_owned()).expect("open source");
let escaped = candidate_path.to_string_lossy().replace('\'', "''");
source_conn
.execute(&format!("VACUUM INTO '{escaped}'"))
.expect("vacuum into candidate");
close_connection(source_conn).expect("close source");
let candidate = logical_witness(&candidate_path).expect("candidate witness");
assert_eq!(candidate.tables, source.tables);
assert_eq!(candidate.contents_sha256, source.contents_sha256);
assert_eq!(
candidate.schema_sha256,
source.schema_sha256,
"VACUUM INTO re-serialized DDL must hash identically (fsqlite {})",
option_env!("BR_FSQLITE_VERSION").unwrap_or("unknown")
);
assert!(logical_witnesses_match_except_integrity(
&source, &candidate
));
}
#[test]
fn plan_token_binds_logical_state_not_sqlite_file_layout() {
let mut logical = LogicalDatabaseWitness {
user_version: 14,
integrity_check: "ok".to_string(),
schema_sha256: "schema".to_string(),
contents_sha256: "contents".to_string(),
tables: Vec::new(),
};
let forecast = MigrationForecast {
from_version: 14,
to_version: 15,
content_hash_rows_rebuilt: 0,
gate_result_history_created: true,
prerequisites_column_added: false,
dependency_type_key_added: false,
legacy_v15_conversion: None,
post_migration_maintenance: true,
};
let first = compute_plan_token("db", &logical, &forecast).unwrap();
assert_eq!(
first,
compute_plan_token("db", &logical, &forecast).unwrap(),
"a repeated logical observation must be stable even when SQLite checkpoints or \
retires sidecars between CLI processes"
);
logical.contents_sha256 = "changed".to_string();
assert_ne!(
first,
compute_plan_token("db", &logical, &forecast).unwrap(),
"any logical content change must stale the token"
);
logical.contents_sha256 = "contents".to_string();
assert_ne!(
first,
compute_plan_token("different-db", &logical, &forecast).unwrap(),
"the absolute database route remains token-bound"
);
}
#[test]
fn constant_time_text_comparison_covers_length_and_content() {
assert!(constant_time_text_eq("abc", "abc"));
assert!(!constant_time_text_eq("abc", "abd"));
assert!(!constant_time_text_eq("abc", "ab"));
assert!(!constant_time_text_eq("", "x"));
}
#[test]
fn stable_raw_comparison_explicitly_classifies_present_shm_as_volatile() {
let component = |suffix: &str, sha256: &str| RawComponentWitness {
suffix: suffix.to_string(),
present: true,
length: Some(4096),
sha256: Some(sha256.repeat(64)),
unix_mode: None,
};
let absent = |suffix: &str| RawComponentWitness {
suffix: suffix.to_string(),
present: false,
length: None,
sha256: None,
unix_mode: None,
};
let before_probe = RawFamilyWitness {
components: vec![
component("", "a"),
absent("-wal"),
component("-shm", "b"),
absent("-journal"),
],
};
let after_probe = RawFamilyWitness {
components: vec![
component("", "a"),
absent("-wal"),
component("-shm", "c"),
absent("-journal"),
],
};
assert_ne!(
before_probe, after_probe,
"the exact pre-open witness must still record the shm bytes that were restored"
);
assert!(
stable_raw_eq(&before_probe, &after_probe),
"post-open validation must explicitly exclude only the volatile shm component"
);
}
#[test]
fn live_index_install_ignores_only_namespace_coordination_sidecars() {
assert_eq!(
NAMESPACE_COORDINATION_SUFFIXES.as_slice(),
config::FSQLITE_NAMESPACE_SIDECAR_SUFFIXES
);
let component = |suffix: &str, sha256: &str| RawComponentWitness {
suffix: suffix.to_string(),
present: true,
length: Some(4096),
sha256: Some(sha256.repeat(64)),
unix_mode: Some(0o644),
};
let absent = |suffix: &str| RawComponentWitness {
suffix: suffix.to_string(),
present: false,
length: None,
sha256: None,
unix_mode: None,
};
let family = |main: &str, wal: &str, namespace_present: bool| RawFamilyWitness {
components: vec![
component("", main),
component("-wal", wal),
component("-shm", "c"),
absent("-journal"),
if namespace_present {
component("-fsqlite-ns-gate", "d")
} else {
absent("-fsqlite-ns-gate")
},
if namespace_present {
component("-fsqlite-ns-use", "e")
} else {
absent("-fsqlite-ns-use")
},
],
};
let expected = family("a", "b", false);
assert!(
family_matches_outside_namespace_coordination(&expected, &family("a", "b", true)),
"sidecars created by taking the engine exclusion are not a family change"
);
assert!(
!family_matches_outside_namespace_coordination(&expected, &family("f", "b", true)),
"a main-file change must still be detected while the sidecars appear"
);
assert!(
!family_matches_outside_namespace_coordination(&expected, &family("a", "f", true)),
"a WAL change must still be detected while the sidecars appear"
);
let mut journal_appeared = family("a", "b", true);
journal_appeared.components[3] = component("-journal", "f");
assert!(
!family_matches_outside_namespace_coordination(&expected, &journal_appeared),
"every other sidecar is still compared"
);
}
#[test]
fn index_entries_without_table_rows_are_recognized() {
for lost in [
"database disk image is malformed: index `idx_comments_issue` contains stale rowid \
5 with no matching table row",
"database disk image is malformed: index `idx_events_issue` contains 7 entries but \
table `events` requires exactly 6",
"*** in database main ***\nindex `i` contains 12 entries but table `t` requires \
exactly 3",
] {
assert!(
integrity_reports_index_entries_without_table_rows(lost),
"{lost}"
);
}
for index_only in [
"database disk image is malformed: table `issues` rowid 30 is missing from index \
`idx_issues_updated_at`",
"database disk image is malformed: index `i` contains 5 entries but table `t` \
requires exactly 6",
"database disk image is malformed: index idx_issues_updated_at root: page 210 \
freeblocks invalid",
"wrong # of entries in index idx_issues_updated_at",
"ok",
] {
assert!(
!integrity_reports_index_entries_without_table_rows(index_only),
"{index_only}"
);
}
}
#[test]
fn integrity_classification_allows_only_known_page_layout_artifacts() {
assert!(integrity_check_is_clean("ok"));
assert!(!integrity_check_is_repairable("ok"));
assert!(integrity_check_is_repairable(
"database disk image is malformed: page 54 is never used"
));
assert!(integrity_check_is_repairable(
"*** in database main ***\nrow 3 missing from index idx_issues_ready"
));
assert!(!integrity_check_is_repairable(
"database disk image is malformed: btree page 12 cell 4"
));
assert!(!integrity_check_is_repairable(
"page 54 is never used\nforeign key constraint failed"
));
}
#[test]
fn run_id_validation_refuses_path_traversal() {
for invalid in ["", ".", "..", "../run", "a/b", "a\\b"] {
assert!(validate_run_id(invalid).is_err(), "{invalid:?}");
}
assert!(validate_run_id("20260727T120000.000000Z-1-0").is_ok());
}
fn reviewed_v14_migration_context_with_database_name(
database_name: &str,
) -> (TempDir, MigrationContext) {
reviewed_source_migration_context(database_name, 14)
}
fn reviewed_source_migration_context(
database_name: &str,
version: u32,
) -> (TempDir, MigrationContext) {
assert!(matches!(version, 14 | 17 | 18));
let temp = TempDir::new().expect("tempdir");
let beads_dir = temp.path().join(".beads");
fs::create_dir(&beads_dir).expect("create beads dir");
let db_path = beads_dir.join(database_name);
let conn = Connection::open(db_path.to_string_lossy().into_owned()).expect("open db");
let mut source_schema = crate::storage::schema::SCHEMA_SQL.replace(
"PRIMARY KEY (issue_id, depends_on_id, type)",
"PRIMARY KEY (issue_id, depends_on_id)",
);
if version < 18 {
source_schema =
source_schema.replace(" prerequisites TEXT NOT NULL DEFAULT '',\n", "");
}
crate::storage::schema::execute_batch(&conn, &source_schema)
.expect("create pre-v18 schema");
conn.execute(
"INSERT INTO issues (
id, title, status, priority, issue_type, created_at, updated_at
) VALUES (
'bd-schema-rehearsal', 'Schema rehearsal', 'open', 2, 'task',
'2026-07-27T12:00:00Z', '2026-07-27T12:00:00Z'
)",
)
.expect("seed issue");
if version == 14 {
conn.execute("DROP TABLE gate_result_history")
.expect("restore v14 shape");
}
conn.execute(&format!("PRAGMA user_version = {version}"))
.expect("stamp matching source version");
close_connection(conn).expect("close fixture");
let authority = Arc::new(
crate::sync::blocking_database_family_write_lock_with_timeout(
&beads_dir,
&db_path,
Some(1000),
)
.expect("acquire test authority"),
);
(
temp,
MigrationContext {
beads_dir,
db_path,
write_authority: authority,
},
)
}
fn reviewed_v14_migration_context() -> (TempDir, MigrationContext) {
reviewed_v14_migration_context_with_database_name("beads.db")
}
fn truncate_fixture_wal(db_path: &Path) {
let conn = Connection::open(db_path.to_string_lossy().into_owned()).unwrap();
conn.execute("PRAGMA wal_checkpoint(TRUNCATE)").unwrap();
close_connection(conn).unwrap();
let wal_size = fs::metadata(family_component_path(db_path, "-wal"))
.unwrap()
.len();
assert!(wal_size <= 32, "fixture WAL must hold no frames");
}
#[test]
fn vacuum_candidate_never_opens_or_rewrites_live_wal_family() {
let (_temp, migration) = reviewed_source_migration_context("beads.db", 17);
let mut writer =
Connection::open(migration.db_path.to_string_lossy().into_owned()).unwrap();
writer.execute("PRAGMA journal_mode = WAL").unwrap();
writer.execute("PRAGMA wal_autocheckpoint = 0").unwrap();
writer
.execute("UPDATE issues SET title = 'private vacuum source sentinel'")
.unwrap();
writer.close_without_checkpoint_in_place().unwrap();
drop(writer);
let logical_before = logical_witness(&migration.db_path).unwrap();
let raw_before = raw_family_witness(&migration.db_path).unwrap();
let wal_path = family_component_path(&migration.db_path, "-wal");
assert!(
fs::metadata(&wal_path).unwrap().len() > 32,
"fixture must retain committed WAL frames"
);
let run_dir = migration.beads_dir.join("private-vacuum-source-test");
let before_dir = run_dir.join("before");
ensure_new_directory(&run_dir).unwrap();
ensure_new_directory(&before_dir).unwrap();
copy_family_to_backup(&migration.db_path, &before_dir, &raw_before).unwrap();
verify_backup_family(&migration.db_path, &before_dir, &raw_before).unwrap();
let candidate = maintenance_candidate_path(&migration.db_path, &run_dir).unwrap();
require_absent_family(&candidate).unwrap();
vacuum_candidate_from_private_source(&migration.db_path, &candidate, &run_dir, &raw_before)
.unwrap();
assert_eq!(
raw_family_witness(&migration.db_path).unwrap(),
raw_before,
"VACUUM candidate construction must not touch any live family byte"
);
assert_eq!(logical_witness(&migration.db_path).unwrap(), logical_before);
let candidate_logical = logical_witness(&candidate).unwrap();
assert_eq!(candidate_logical.user_version, logical_before.user_version);
assert_eq!(
candidate_logical.contents_sha256,
logical_before.contents_sha256
);
assert_eq!(candidate_logical.tables, logical_before.tables);
assert!(
run_dir
.join("maintenance-vacuum-source")
.join("beads.db")
.is_file(),
"the disposable source is retained for interrupted-migration diagnosis"
);
}
#[test]
fn engine_recovery_refuses_framed_wal_and_preserves_family() {
let (_temp, migration, _expected) = engine_recovery_fixture(false);
let before = recovery_family_witness(&migration.db_path).unwrap();
let error = execute_recover(&DoctorMigrateSchemaRecoverArgs { json: true }, &migration)
.unwrap_err();
assert!(
error.to_string().contains(
"engine recovery refused before opening: WAL contains frames that may hold \
committed, uncheckpointed changes"
),
"unexpected refusal: {error}"
);
assert_eq!(
recovery_family_witness(&migration.db_path).unwrap(),
before,
"the refusal must leave every family byte and mode unchanged"
);
assert!(!migration_runs_root(&migration.beads_dir).exists());
}
#[test]
fn engine_recovery_admits_legacy_empty_wal_before_migration() {
let (_temp, migration, expected) = engine_recovery_fixture(true);
let before = recovery_family_witness(&migration.db_path).unwrap();
execute_recover(&DoctorMigrateSchemaRecoverArgs { json: true }, &migration).unwrap();
let after = logical_witness(&migration.db_path).unwrap();
assert_eq!(after, expected);
require_recovery_payload_unchanged(
&before,
&recovery_family_witness(&migration.db_path).unwrap(),
)
.unwrap();
let run = fs::read_dir(migration_runs_root(&migration.beads_dir))
.unwrap()
.next()
.unwrap()
.unwrap()
.path();
verify_backup_family(&migration.db_path, &run.join("recovery-before"), &before).unwrap();
assert_eq!(
before.components.len(),
1 + config::db_sidecar_suffixes().count()
);
assert!(run.join("recovery-complete.json").is_file());
let plan = build_plan(&migration.db_path).unwrap();
assert!(plan.eligible);
execute_apply(
&DoctorMigrateSchemaApplyArgs {
plan_token: plan.plan_token.unwrap(),
json: true,
},
&migration,
)
.unwrap();
let migrated = logical_witness(&migration.db_path).unwrap();
assert_eq!(
migrated.user_version,
u32::try_from(CURRENT_SCHEMA_VERSION).unwrap()
);
assert_eq!(migrated.integrity_check, "ok");
}
fn engine_recovery_fixture(
checkpoint: bool,
) -> (TempDir, MigrationContext, LogicalDatabaseWitness) {
let (temp, migration) = reviewed_source_migration_context("beads.db", 17);
let conn = Connection::open(migration.db_path.to_string_lossy().into_owned()).unwrap();
conn.execute("PRAGMA journal_mode=WAL").unwrap();
conn.execute("UPDATE issues SET title='committed WAL recovery row'")
.unwrap();
if checkpoint {
conn.execute("PRAGMA wal_checkpoint(TRUNCATE)").unwrap();
}
close_connection(conn).unwrap();
let wal_size = fs::metadata(family_component_path(&migration.db_path, "-wal"))
.unwrap()
.len();
if checkpoint {
assert!(wal_size <= 32, "checkpointed fixture has no WAL frames");
} else {
assert!(wal_size > 32, "fixture must retain committed WAL frames");
}
let expected = logical_witness(&migration.db_path).unwrap();
let shm = family_component_path(&migration.db_path, "-shm");
fs::write(&shm, vec![0_u8; 32_768]).unwrap();
assert!(open_read_only(&migration.db_path).is_err());
(temp, migration, expected)
}
#[test]
fn engine_recovery_refuses_peer_before_creating_backup() {
let (_temp, migration) = reviewed_v14_migration_context();
truncate_fixture_wal(&migration.db_path);
let peer = crate::sync::DatabaseOpenerLease::register(&migration.db_path).unwrap();
assert!(peer.is_registered());
let before = recovery_family_witness(&migration.db_path).unwrap();
let error =
execute_recover(&DoctorMigrateSchemaRecoverArgs::default(), &migration).unwrap_err();
assert!(error.to_string().contains("sole opener"));
assert_eq!(before, recovery_family_witness(&migration.db_path).unwrap());
assert!(!migration_runs_root(&migration.beads_dir).exists());
}
#[cfg(unix)]
#[test]
fn engine_recovery_refuses_engine_sidecar_symlink() {
let (_temp, migration) = reviewed_v14_migration_context();
truncate_fixture_wal(&migration.db_path);
let outside = migration.beads_dir.join("outside-marker");
fs::write(&outside, b"untouched").unwrap();
let marker = family_component_path(&migration.db_path, ".fsqlite-migration-state");
if marker.exists() {
fs::rename(&marker, migration.beads_dir.join("retained-marker")).unwrap();
}
std::os::unix::fs::symlink(&outside, &marker).unwrap();
let error =
execute_recover(&DoctorMigrateSchemaRecoverArgs::default(), &migration).unwrap_err();
assert!(error.to_string().contains("symlink"));
assert_eq!(fs::read(&outside).unwrap(), b"untouched");
assert!(!migration_runs_root(&migration.beads_dir).exists());
}
#[test]
fn engine_recovery_preserves_prestate_when_private_open_fails() {
let (_temp, migration) = reviewed_v14_migration_context();
truncate_fixture_wal(&migration.db_path);
fs::write(&migration.db_path, b"not a database").unwrap();
let before = recovery_family_witness(&migration.db_path).unwrap();
let error =
execute_recover(&DoctorMigrateSchemaRecoverArgs::default(), &migration).unwrap_err();
assert!(error.to_string().contains("private-recovery"));
assert_eq!(before, recovery_family_witness(&migration.db_path).unwrap());
let run = fs::read_dir(migration_runs_root(&migration.beads_dir))
.unwrap()
.next()
.unwrap()
.unwrap()
.path();
verify_backup_family(&migration.db_path, &run.join("recovery-before"), &before).unwrap();
assert!(run.join("recovery-failed.json").is_file());
}
fn legacy_v15_migration_context() -> (TempDir, MigrationContext) {
let temp = TempDir::new().unwrap();
let beads_dir = temp.path().join(".beads");
fs::create_dir(&beads_dir).unwrap();
let db_path = beads_dir.join("beads.db");
let conn = Connection::open(db_path.to_string_lossy().into_owned()).unwrap();
crate::storage::schema::execute_batch(
&conn,
&crate::storage::schema::SCHEMA_SQL
.replace(" prerequisites TEXT NOT NULL DEFAULT '',\n", ""),
)
.unwrap();
crate::storage::schema::execute_batch(
&conn,
crate::storage::schema::LEGACY_V15_TEST_CORE_SQL,
)
.unwrap();
crate::storage::schema::execute_batch(
&conn,
r#"
CREATE TABLE "operator'opaque" (key TEXT PRIMARY KEY, payload BLOB);
INSERT INTO "operator'opaque" (rowid, key, payload) VALUES (991, 'retain', X'00FF');
CREATE TABLE operator_without_rowid (key TEXT PRIMARY KEY, payload TEXT) WITHOUT ROWID;
INSERT INTO operator_without_rowid VALUES ('retain', 'full operator value');
DROP INDEX idx_labels_issue;
CREATE INDEX idx_labels_issue ON labels(issue_id);
DROP INDEX idx_blocked_cache_blocked_at;
CREATE INDEX idx_blocked_cache_blocked_at ON blocked_issues_cache(blocked_at);
DROP INDEX idx_config_key;
CREATE INDEX IF NOT EXISTS idx_config_key ON config("key");
"#,
)
.unwrap();
close_connection(conn).unwrap();
let write_authority = Arc::new(
crate::sync::blocking_database_family_write_lock_with_timeout(
&beads_dir,
&db_path,
Some(1000),
)
.unwrap(),
);
(
temp,
MigrationContext {
beads_dir,
db_path,
write_authority,
},
)
}
fn apply_legacy_v15_upgrade(migration: &MigrationContext) -> AppliedMigrationReceipt {
let plan = build_plan(&migration.db_path).expect("plan the reviewed legacy profile");
assert!(plan.eligible);
let forecast = plan.forecast.as_ref().unwrap();
assert_eq!((forecast.from_version, forecast.to_version), (15, 19));
assert!(!forecast.dependency_type_key_added);
assert!(forecast.prerequisites_column_added);
assert_eq!(forecast.content_hash_rows_rebuilt, 0);
let projection = forecast.legacy_v15_conversion.as_ref().unwrap();
assert_eq!(projection.profile, LEGACY_V15_PROFILE);
assert_eq!(projection.retired_indexes.len(), 7);
assert!(projection.tables.iter().any(|table| {
table.table.name == "operator_without_rowid" && !table.rowid_included
}));
execute_apply(
&DoctorMigrateSchemaApplyArgs {
plan_token: plan.plan_token.unwrap(),
json: false,
},
migration,
)
.expect("install the fully projected legacy conversion");
let runs: Vec<_> = fs::read_dir(migration_runs_root(&migration.beads_dir))
.unwrap()
.map(|entry| entry.unwrap().path())
.collect();
assert_eq!(runs.len(), 1);
let applied: AppliedMigrationReceipt = read_json(&runs[0].join("applied.json")).unwrap();
assert!(applied.attested, "{:?}", applied.attestation_errors);
assert!(applied.effects.legacy_v15_tables_converted);
assert!(!applied.effects.dependency_type_key_added);
applied
}
fn raw_legacy_payloads(db_path: &Path) -> Vec<String> {
let conn = open_read_only(db_path).unwrap();
let rows = [
"SELECT rowid, id, title, content_hash FROM issues ORDER BY rowid",
"SELECT rowid, * FROM dependencies ORDER BY rowid",
"SELECT rowid, * FROM comments ORDER BY rowid",
"SELECT rowid, * FROM events ORDER BY rowid",
"SELECT rowid, issue_id, marked_at FROM dirty_issues ORDER BY rowid",
"SELECT rowid, * FROM export_hashes ORDER BY rowid",
"SELECT rowid, parent_id, last_child FROM child_counters ORDER BY rowid",
"SELECT rowid, * FROM labels ORDER BY rowid",
"SELECT rowid, * FROM operator_notes ORDER BY rowid",
"SELECT rowid, * FROM \"operator'opaque\" ORDER BY rowid",
"SELECT * FROM operator_without_rowid ORDER BY key",
]
.into_iter()
.map(|sql| format!("{:?}", conn.query(sql).unwrap()))
.collect();
close_connection(conn).unwrap();
rows
}
#[test]
fn reviewed_legacy_v15_plan_apply_and_undo_preserve_projected_data_and_source_bytes() {
let (_temp, migration) = legacy_v15_migration_context();
let payloads = raw_legacy_payloads(&migration.db_path);
let applied = apply_legacy_v15_upgrade(&migration);
assert_eq!(raw_legacy_payloads(&migration.db_path), payloads);
let conn = open_read_only(&migration.db_path).unwrap();
assert_eq!(
conn.query("SELECT name, seq FROM sqlite_sequence WHERE name IN ('comments', 'events') ORDER BY name")
.unwrap()
.into_iter()
.map(|row| row.values().to_vec())
.collect::<Vec<_>>(),
vec![
vec![SqliteValue::from("comments"), SqliteValue::from(197_i64)],
vec![SqliteValue::from("events"), SqliteValue::from(79_i64)],
]
);
assert!(conn.query("PRAGMA foreign_key_check").unwrap().is_empty());
assert_eq!(integrity_check_messages(&conn).unwrap(), vec!["ok"]);
close_connection(conn).unwrap();
execute_undo(
&DoctorMigrateSchemaUndoArgs {
run_id: applied.run_id,
dry_run: false,
json: false,
},
&migration,
)
.unwrap();
assert_eq!(
raw_family_witness(&migration.db_path).unwrap(),
applied.raw_before
);
assert_eq!(
logical_witness(&migration.db_path).unwrap(),
applied.logical_before
);
assert_eq!(raw_legacy_payloads(&migration.db_path), payloads);
}
#[test]
fn legacy_upgrade_plan_and_undo_bind_hidden_rowids_even_when_values_are_equal() {
for mutation in [
"UPDATE dependencies SET rowid = 2001 WHERE rowid = 11",
"UPDATE \"operator'opaque\" SET rowid = 2002 WHERE rowid = 991",
] {
let (_temp, migration) = legacy_v15_migration_context();
let applied = apply_legacy_v15_upgrade(&migration);
let logical_before = logical_witness(&migration.db_path).unwrap();
let conn = Connection::open(migration.db_path.to_string_lossy().into_owned()).unwrap();
conn.execute(mutation).unwrap();
close_connection(conn).unwrap();
assert_eq!(logical_witness(&migration.db_path).unwrap(), logical_before);
let bytes_before = fs::read(&migration.db_path).unwrap();
let error = execute_undo(
&DoctorMigrateSchemaUndoArgs {
run_id: applied.run_id.clone(),
dry_run: false,
json: false,
},
&migration,
)
.expect_err("hidden rowid changes are newer work, despite equal ordinary values");
assert!(
error.to_string().contains("live database has changed"),
"{error}"
);
assert_eq!(fs::read(&migration.db_path).unwrap(), bytes_before);
assert!(
!migration_runs_root(&migration.beads_dir)
.join(applied.run_id)
.join("undo-prepared.json")
.exists()
);
}
let (_temp, migration) = legacy_v15_migration_context();
let plan_before = build_plan(&migration.db_path).unwrap();
let conn = Connection::open(migration.db_path.to_string_lossy().into_owned()).unwrap();
conn.execute("UPDATE dependencies SET rowid = 2003 WHERE rowid = 11")
.unwrap();
close_connection(conn).unwrap();
let plan_after = build_plan(&migration.db_path).unwrap();
assert_eq!(plan_before.logical_witness, plan_after.logical_witness);
assert_ne!(plan_before.plan_token, plan_after.plan_token);
}
#[test]
fn legacy_upgrade_refuses_stale_operator_constraints_even_with_equal_rows() {
for (before, changed) in [
("a * (b + c) > 0", "a * b + c > 0"),
("\"null\" IS NULL", "null IS NULL"),
("\"1\" > 0", "1 > 0"),
("\"ALLOWED\" = 'ALLOWED'", "\"allowed\" = 'ALLOWED'"),
] {
let (_temp, migration) = legacy_v15_migration_context();
let declaration = |predicate| {
format!(
"CREATE TABLE operator_constraint (a, b, c, \"null\", \"1\", CHECK({predicate}))"
)
};
let conn = Connection::open(migration.db_path.to_string_lossy().into_owned()).unwrap();
conn.execute(&declaration(before)).unwrap();
close_connection(conn).unwrap();
let plan_before = match build_plan(&migration.db_path) {
Ok(plan) => plan,
Err(error)
if before.contains("\"ALLOWED\"")
&& error.to_string().contains("no such column: ALLOWED") =>
{
continue;
}
Err(error) => panic!("plan for {before}: {error}"),
};
let conn = Connection::open(migration.db_path.to_string_lossy().into_owned()).unwrap();
conn.execute("DROP TABLE operator_constraint").unwrap();
conn.execute(&declaration(changed)).unwrap();
close_connection(conn).unwrap();
let plan_after = build_plan(&migration.db_path).unwrap();
assert_eq!(plan_before.logical_witness, plan_after.logical_witness);
assert_ne!(plan_before.plan_token, plan_after.plan_token);
let bytes_before = fs::read(&migration.db_path).unwrap();
execute_apply(
&DoctorMigrateSchemaApplyArgs {
plan_token: plan_before.plan_token.unwrap(),
json: false,
},
&migration,
)
.expect_err("an old plan cannot authorize different operator constraints");
assert_eq!(fs::read(&migration.db_path).unwrap(), bytes_before);
assert!(!migration_runs_root(&migration.beads_dir).exists());
}
}
fn apply_prerequisite_upgrade(migration: &MigrationContext) -> AppliedMigrationReceipt {
let plan = build_plan(&migration.db_path).expect("plan v17 upgrade");
assert!(plan.eligible);
assert_eq!(
(plan.from_version, plan.to_version),
(17, current_schema_version().unwrap())
);
let forecast = plan.forecast.as_ref().expect("migration forecast");
assert!(forecast.prerequisites_column_added);
assert!(forecast.dependency_type_key_added);
assert_eq!(forecast.content_hash_rows_rebuilt, 0);
assert!(!forecast.gate_result_history_created);
execute_apply(
&DoctorMigrateSchemaApplyArgs {
plan_token: plan.plan_token.expect("plan token"),
json: false,
},
migration,
)
.expect("apply v17 upgrade");
let runs = fs::read_dir(migration_runs_root(&migration.beads_dir))
.expect("read runs")
.map(|entry| entry.expect("run entry").path())
.collect::<Vec<_>>();
assert_eq!(runs.len(), 1);
let applied: AppliedMigrationReceipt =
read_json(&runs[0].join("applied.json")).expect("applied receipt");
assert!(applied.attested, "{:?}", applied.attestation_errors);
assert!(applied.effects.prerequisites_column_added);
assert!(applied.effects.dependency_type_key_added);
assert_eq!(applied.effects.content_hash_rows_rebuilt, 0);
applied
}
fn seed_v17_auxiliary_rows(conn: &Connection) {
crate::storage::schema::execute_batch(
conn,
"INSERT INTO labels (issue_id, label) VALUES ('bd-schema-rehearsal', 'handoff');
INSERT INTO comments (issue_id, author, text)
VALUES ('bd-schema-rehearsal', 'reviewer', 'Preserve this audit trail');
INSERT INTO events (issue_id, event_type, actor, new_value, harness)
VALUES ('bd-schema-rehearsal', 'created', 'owner', 'open', 'codex');
INSERT INTO gate_results (issue_id, gate, provider, passed)
VALUES ('bd-schema-rehearsal', 'reviewed', 'reviewer', 1);
INSERT INTO gate_result_history
(issue_id, from_status, to_status, status_revision, gate, provider, passed)
VALUES ('bd-schema-rehearsal', 'open', 'in_progress', 0, 'reviewed', 'reviewer', 1);
INSERT INTO capacity_exemptions
(issue_id, capacity_kind, capacity_name, provider, reason, granted_by)
VALUES ('bd-schema-rehearsal', 'status', 'open', 'lead', 'external review', 'owner');
INSERT INTO capacity_exemption_history
(issue_id, capacity_kind, capacity_name, action, provider, reason, actor)
VALUES ('bd-schema-rehearsal', 'status', 'open', 'grant', 'lead', 'external review', 'owner');
INSERT INTO capacity_occupancy (issue_id, actor, harness, session)
VALUES ('bd-schema-rehearsal', 'owner', 'codex', 'original-session');
INSERT INTO config (key, value) VALUES ('issue_prefix', 'bd');
INSERT INTO metadata (key, value) VALUES ('needs_flush', 'true');
INSERT INTO child_counters (parent_id, last_child) VALUES ('bd-schema-rehearsal', 7);",
)
.expect("seed v17 auxiliary state and history");
}
#[test]
fn typed_key_plan_refuses_operator_foreign_keys_without_changing_source() {
let (_temp, migration) = reviewed_source_migration_context("beads.db", 18);
let conn = Connection::open(migration.db_path.to_string_lossy().into_owned()).unwrap();
crate::storage::schema::execute_batch(
&conn,
r#"
INSERT INTO issues (id, title) VALUES ('bd-target', 'Target');
INSERT INTO dependencies (issue_id, depends_on_id)
VALUES ('bd-schema-rehearsal', 'bd-target');
CREATE TABLE "operator'notes" (
source TEXT, target TEXT, note TEXT,
FOREIGN KEY (source, target)
REFERENCES "DEPENDENCIES" (issue_id, depends_on_id) ON DELETE CASCADE
);
INSERT INTO "operator'notes" VALUES
('bd-schema-rehearsal', 'bd-target', 'preserve this relationship');
"#,
)
.unwrap();
close_connection(conn).unwrap();
let raw_before = raw_family_witness(&migration.db_path).unwrap();
let logical_before = logical_witness(&migration.db_path).unwrap();
let error =
build_plan(&migration.db_path).expect_err("pair foreign key cannot survive widening");
assert!(error.to_string().contains("operator'notes"), "{error}");
assert!(
error
.to_string()
.contains("foreign key referencing dependencies"),
"{error}"
);
assert_eq!(raw_family_witness(&migration.db_path).unwrap(), raw_before);
assert_eq!(logical_witness(&migration.db_path).unwrap(), logical_before);
assert!(!migration_runs_root(&migration.beads_dir).exists());
}
#[test]
fn reviewed_v18_typed_key_preserves_data_and_restores_source_bytes() {
let (_temp, migration) = reviewed_source_migration_context("beads.db", 18);
let conn = Connection::open(migration.db_path.to_string_lossy().into_owned()).unwrap();
seed_v17_auxiliary_rows(&conn);
crate::storage::schema::execute_batch(
&conn,
r#"
UPDATE issues SET prerequisites = '- [x] original input',
acceptance_criteria = '- [ ] final outcome', content_hash = 'original-hash';
INSERT INTO issues (id, title) VALUES ('bd-target', 'Dependency target');
INSERT INTO dependencies
(rowid, issue_id, depends_on_id, type, created_at, created_by, metadata, thread_id)
VALUES (41, 'bd-schema-rehearsal', 'bd-target', 'blocks',
'2026-01-30 04:07:21', 'original-actor', '{"keep":"payload"}', 'thread-19');
DROP INDEX idx_dependencies_thread;
"#,
)
.unwrap();
close_connection(conn).unwrap();
let plan =
build_plan(&migration.db_path).expect("plan canonical v18 with a missing owned index");
assert!(plan.eligible);
assert_eq!(
(plan.from_version, plan.to_version),
(18, current_schema_version().unwrap())
);
let forecast = plan.forecast.unwrap();
assert!(forecast.dependency_type_key_added);
assert!(!forecast.prerequisites_column_added);
assert_eq!(forecast.content_hash_rows_rebuilt, 0);
assert!(!forecast.gate_result_history_created);
execute_apply(
&DoctorMigrateSchemaApplyArgs {
plan_token: plan.plan_token.unwrap(),
json: false,
},
&migration,
)
.expect("apply typed-key migration and repair owned index");
let runs = fs::read_dir(migration_runs_root(&migration.beads_dir))
.unwrap()
.map(|entry| entry.unwrap().path())
.collect::<Vec<_>>();
assert_eq!(runs.len(), 1);
let applied: AppliedMigrationReceipt = read_json(&runs[0].join("applied.json")).unwrap();
assert!(applied.attested, "{:?}", applied.attestation_errors);
assert!(applied.effects.dependency_type_key_added);
assert!(!applied.effects.prerequisites_column_added);
assert_eq!(
applied.logical_after.as_ref().unwrap().tables,
applied.logical_before.tables
);
let conn = open_read_only(&migration.db_path).unwrap();
let rows = conn
.query("SELECT rowid, metadata, thread_id FROM dependencies")
.unwrap();
assert_eq!(rows.len(), 1);
assert_eq!(rows[0].get(0).and_then(SqliteValue::as_integer), Some(41));
assert_eq!(
rows[0].get(1).and_then(SqliteValue::as_text),
Some(r#"{"keep":"payload"}"#)
);
assert_eq!(
rows[0].get(2).and_then(SqliteValue::as_text),
Some("thread-19")
);
assert_eq!(
conn.query("SELECT name FROM sqlite_schema WHERE name = 'idx_dependencies_thread'")
.unwrap()
.len(),
1
);
close_connection(conn).unwrap();
execute_undo(
&DoctorMigrateSchemaUndoArgs {
run_id: applied.run_id,
dry_run: false,
json: false,
},
&migration,
)
.expect("undo v18 upgrade");
assert_eq!(
raw_family_witness(&migration.db_path).unwrap(),
applied.raw_before
);
assert_eq!(
logical_witness(&migration.db_path).unwrap(),
applied.logical_before
);
}
#[test]
fn reviewed_v17_prerequisite_upgrade_preserves_rows_and_undo_bytes() {
let (_temp, migration) = reviewed_source_migration_context("beads.db", 17);
let conn = Connection::open(migration.db_path.to_string_lossy().into_owned())
.expect("open v17 source");
conn.execute(
"UPDATE issues SET acceptance_criteria = '- [ ] final outcome',
content_hash = 'preserved-source-hash' WHERE id = 'bd-schema-rehearsal'",
)
.expect("seed distinct acceptance criteria");
conn.execute(
"INSERT INTO dirty_issues (issue_id, marked_at)
VALUES ('bd-schema-rehearsal', '2026-09-09T01:00:00Z')",
)
.expect("seed dirty bookkeeping");
conn.execute(
"INSERT INTO export_hashes (issue_id, content_hash, exported_at)
VALUES ('bd-schema-rehearsal', 'previous-export', '2026-09-09T00:00:00Z')",
)
.expect("seed export bookkeeping");
seed_v17_auxiliary_rows(&conn);
let source_row = conn.query_row("SELECT * FROM issues").expect("source row");
close_connection(conn).expect("close source");
let applied = apply_prerequisite_upgrade(&migration);
let after = applied
.logical_after
.as_ref()
.expect("attested after state");
for table in applied
.logical_before
.tables
.iter()
.filter(|t| t.name != "issues")
{
assert_eq!(
after.tables.iter().find(|t| t.name == table.name),
Some(table)
);
}
let conn = open_read_only(&migration.db_path).expect("open migrated database");
let row = conn
.query_row("SELECT * FROM issues")
.expect("migrated row");
for index in 0..38 {
assert_eq!(row.get(index), source_row.get(index), "column {index}");
}
assert_eq!(row.get(38).and_then(SqliteValue::as_text), Some(""));
close_connection(conn).expect("close migrated database");
execute_undo(
&DoctorMigrateSchemaUndoArgs {
run_id: applied.run_id,
dry_run: false,
json: false,
},
&migration,
)
.expect("undo v17 migration");
assert_eq!(
raw_family_witness(&migration.db_path).expect("restored bytes"),
applied.raw_before
);
assert_eq!(
logical_witness(&migration.db_path).expect("restored contents"),
applied.logical_before
);
}
#[test]
fn reviewed_v17_prerequisite_write_stales_undo_and_preserves_acceptance() {
let (_temp, migration) = reviewed_source_migration_context("beads.db", 17);
let applied = apply_prerequisite_upgrade(&migration);
let conn = Connection::open(migration.db_path.to_string_lossy().into_owned())
.expect("open migrated database");
conn.execute(
"UPDATE issues SET acceptance_criteria = '- [ ] outcome',
prerequisites = '- [x] input ready' WHERE id = 'bd-schema-rehearsal'",
)
.expect("write distinct prerequisite content");
close_connection(conn).expect("close write");
let before_refusal = logical_witness(&migration.db_path).expect("post-write witness");
let error = execute_undo(
&DoctorMigrateSchemaUndoArgs {
run_id: applied.run_id,
dry_run: false,
json: false,
},
&migration,
)
.expect_err("undo must preserve subsequent prerequisite writes");
assert!(
error.to_string().contains("live database has changed"),
"{error}"
);
assert_eq!(
logical_witness(&migration.db_path).expect("after refusal"),
before_refusal
);
let conn = open_read_only(&migration.db_path).expect("read refused state");
let row = conn
.query_row("SELECT acceptance_criteria, prerequisites FROM issues")
.expect("read independent fields");
assert_eq!(
row.get(0).and_then(SqliteValue::as_text),
Some("- [ ] outcome")
);
assert_eq!(
row.get(1).and_then(SqliteValue::as_text),
Some("- [x] input ready")
);
close_connection(conn).expect("close read");
}
#[test]
fn prerequisite_upgrade_plan_rejects_false_version_and_malformed_column() {
for (statements, expected_error) in [
(
vec!["PRAGMA user_version = 19"],
"runtime shape is not canonical",
),
(
vec!["ALTER TABLE issues ADD COLUMN prerequisites TEXT NOT NULL DEFAULT ''"],
"unsupported historical shape for core table issues",
),
(
vec![
"ALTER TABLE issues ADD COLUMN prerequisites TEXT DEFAULT ''",
"PRAGMA user_version = 19",
],
"runtime shape is not canonical",
),
] {
let (_temp, migration) = reviewed_source_migration_context("beads.db", 17);
let conn = Connection::open(migration.db_path.to_string_lossy().into_owned())
.expect("open source");
for statement in statements {
conn.execute(statement)
.expect("install malformed declaration");
}
close_connection(conn).expect("close malformed source");
let before = logical_witness(&migration.db_path).expect("before witness");
let error = build_plan(&migration.db_path).expect_err("refuse malformed source");
assert!(error.to_string().contains(expected_error), "{error}");
assert_eq!(
logical_witness(&migration.db_path).expect("after witness"),
before
);
assert!(!migration_runs_root(&migration.beads_dir).exists());
}
}
#[test]
fn reviewed_migration_preserves_extension_unique_index_ddl_and_enforcement() {
const EXTENSION_INDEX_NAME: &str = "extension_records_external_key_unique";
const EXTENSION_INDEX_DDL: &str = "CREATE UNIQUE INDEX extension_records_external_key_unique ON extension_records(external_key)";
let (_temp, migration) = reviewed_v14_migration_context();
let source_conn =
Connection::open(migration.db_path.to_string_lossy().into_owned()).expect("open db");
source_conn
.execute(
"CREATE TABLE extension_records (
external_key TEXT NOT NULL,
payload TEXT NOT NULL
)",
)
.expect("create extension table");
source_conn
.execute(EXTENSION_INDEX_DDL)
.expect("create extension unique index");
source_conn
.execute(
"INSERT INTO extension_records (external_key, payload)
VALUES ('stable-key', 'before-migration')",
)
.expect("seed extension row");
let source_index_sql = source_conn
.query_row(&format!(
"SELECT sql FROM sqlite_master WHERE type = 'index' AND name = \
'{EXTENSION_INDEX_NAME}'"
))
.expect("read source extension index DDL")
.get(0)
.and_then(SqliteValue::as_text)
.expect("extension index DDL is text")
.to_string();
assert_eq!(source_index_sql, EXTENSION_INDEX_DDL);
close_connection(source_conn).expect("close extension fixture");
let plan = build_plan(&migration.db_path).expect("build reviewed migration plan");
assert_eq!(
plan.from_version, 14,
"fixture must exercise a real migration"
);
assert!(plan.to_version > plan.from_version);
execute_apply(
&DoctorMigrateSchemaApplyArgs {
plan_token: plan.plan_token.expect("plan token"),
json: false,
},
&migration,
)
.expect("apply reviewed migration with extension index");
let migrated_conn =
Connection::open(migration.db_path.to_string_lossy().into_owned()).expect("open db");
let migrated_index_sql = migrated_conn
.query_row(&format!(
"SELECT sql FROM sqlite_master WHERE type = 'index' AND name = \
'{EXTENSION_INDEX_NAME}'"
))
.expect("extension index survives migration")
.get(0)
.and_then(SqliteValue::as_text)
.expect("migrated extension index DDL is text")
.to_string();
assert_eq!(
migrated_index_sql, source_index_sql,
"reviewed migration must preserve non-br index DDL exactly"
);
let duplicate_error = migrated_conn
.execute(
"INSERT INTO extension_records (external_key, payload)
VALUES ('stable-key', 'duplicate')",
)
.expect_err("extension unique index must remain enforced");
let duplicate_message = duplicate_error.to_string().to_ascii_lowercase();
assert!(
duplicate_message.contains("unique") || duplicate_message.contains("constraint"),
"expected unique/constraint failure, got: {duplicate_error}"
);
let row_count = migrated_conn
.query_row("SELECT COUNT(*) FROM extension_records")
.expect("count extension rows")
.get(0)
.and_then(SqliteValue::as_integer)
.expect("extension row count is integer");
assert_eq!(row_count, 1, "failed duplicate insert must not add a row");
close_connection(migrated_conn).expect("close migrated extension fixture");
}
#[test]
#[allow(
clippy::too_many_lines,
reason = "the end-to-end apply-and-undo proof is clearest as one ordered scenario"
)]
fn reviewed_plan_apply_and_undo_round_trip_exact_logical_state() {
let (_temp, migration) = reviewed_v14_migration_context();
let conn = Connection::open(migration.db_path.to_string_lossy().into_owned())
.expect("open source to remove a known index");
conn.execute("DROP INDEX idx_issues_ready")
.expect("remove known index");
close_connection(conn).expect("close source with missing index");
let plan = build_plan(&migration.db_path).expect("build plan");
assert!(plan.eligible);
assert_eq!(plan.from_version, 14);
assert_eq!(
plan.to_version,
crate::storage::schema::CURRENT_SCHEMA_VERSION as u32
);
let original_logical = plan.logical_witness.clone();
let token = plan.plan_token.expect("plan token");
execute_apply(
&DoctorMigrateSchemaApplyArgs {
plan_token: token,
json: false,
},
&migration,
)
.expect("apply reviewed migration");
let runs_root = migration_runs_root(&migration.beads_dir);
let run_ids: Vec<String> = fs::read_dir(&runs_root)
.expect("read runs")
.map(|entry| {
entry
.expect("run entry")
.file_name()
.to_string_lossy()
.into_owned()
})
.collect();
assert_eq!(run_ids.len(), 1, "one migration run should be recorded");
let run_id = &run_ids[0];
let applied: AppliedMigrationReceipt =
read_json(&runs_root.join(run_id).join("applied.json")).expect("applied receipt");
assert_eq!(
applied
.logical_after
.as_ref()
.expect("attested logical state")
.user_version,
crate::storage::schema::CURRENT_SCHEMA_VERSION as u32
);
assert!(applied.attested);
assert!(applied.forecast.post_migration_maintenance);
assert!(applied.effects.post_migration_maintenance_completed);
let marker: CommitReadyMigrationReceipt =
read_json(&runs_root.join(run_id).join("commit-ready.json"))
.expect("commit-ready receipt");
let mut malformed_applied = applied.clone();
malformed_applied
.raw_after
.as_mut()
.expect("raw committed witness")
.components
.pop();
let malformed_error = validate_applied_against_commit_ready(
&malformed_applied,
&marker,
&runs_root.join(run_id),
)
.expect_err("malformed raw committed witness must be rejected");
assert!(
malformed_error.to_string().contains("raw witness"),
"causal raw-witness validation error must remain visible: {malformed_error}"
);
assert_eq!(
applied
.logical_after
.as_ref()
.expect("attested logical state")
.integrity_check,
"ok"
);
assert!(runs_root.join(run_id).join("before/beads.db").is_file());
execute_undo(
&DoctorMigrateSchemaUndoArgs {
run_id: run_id.clone(),
dry_run: false,
json: false,
},
&migration,
)
.expect("undo reviewed migration");
assert_eq!(
logical_witness(&migration.db_path).expect("restored witness"),
original_logical
);
assert!(runs_root.join(run_id).join("undone.json").is_file());
let quarantine = runs_root.join(run_id).join("undo-quarantine");
assert!(
fs::read_dir(quarantine)
.expect("read quarantine")
.next()
.is_some(),
"undo must retain the displaced applied state"
);
let completed: UndoReceipt =
read_json(&runs_root.join(run_id).join("undone.json")).expect("completed undo receipt");
let mut inconsistent = completed.clone();
inconsistent.logical_expected_before.user_version += 1;
let inconsistent_error = validate_completed_undo_receipt(
&inconsistent,
&DoctorMigrateSchemaUndoArgs {
run_id: run_id.clone(),
dry_run: false,
json: false,
},
&migration,
)
.expect_err("completed undo receipt must remain bound to the applied receipt");
assert!(
inconsistent_error
.to_string()
.contains("internally inconsistent"),
"causal completed-receipt error must remain visible: {inconsistent_error}"
);
execute_undo(
&DoctorMigrateSchemaUndoArgs {
run_id: run_id.clone(),
dry_run: false,
json: false,
},
&migration,
)
.expect("completed undo must be idempotent");
}
#[cfg(any(target_os = "linux", target_os = "android", target_vendor = "apple"))]
#[test]
fn undo_atomically_displaces_the_authority_held_main_with_hardlink_aliases() {
let (_temp, migration) = reviewed_v14_migration_context();
let plan = build_plan(&migration.db_path).expect("build plan");
execute_apply(
&DoctorMigrateSchemaApplyArgs {
plan_token: plan.plan_token.expect("plan token"),
json: false,
},
&migration,
)
.expect("apply reviewed migration");
let run_dir = fs::read_dir(migration_runs_root(&migration.beads_dir))
.expect("read migration runs")
.next()
.expect("one run")
.expect("run entry")
.path();
let run_id = run_dir
.file_name()
.and_then(|value| value.to_str())
.expect("UTF-8 run id")
.to_string();
let applied_alias = migration.beads_dir.join("applied-generation-alias.db");
fs::hard_link(&migration.db_path, &applied_alias).expect("create applied hardlink alias");
let applied_identity = fs::metadata(&applied_alias).expect("alias metadata");
execute_undo(
&DoctorMigrateSchemaUndoArgs {
run_id,
dry_run: false,
json: false,
},
&migration,
)
.expect("undo through atomic prelocked exchange");
let quarantine_root = run_dir.join("undo-quarantine");
let quarantine_dir = fs::read_dir(&quarantine_root)
.expect("read quarantine root")
.next()
.expect("one quarantine")
.expect("quarantine entry")
.path();
let quarantined_main = backup_component_path(&quarantine_dir, &migration.db_path, "")
.expect("quarantine main");
let quarantined_identity = fs::metadata(&quarantined_main).expect("quarantined metadata");
let restored_identity = fs::metadata(&migration.db_path).expect("restored metadata");
assert!(
same_file_identity(&applied_identity, &quarantined_identity),
"the alias-visible applied inode must be retained, not copied or overwritten"
);
assert!(
!same_file_identity(&applied_identity, &restored_identity),
"the canonical path must atomically adopt the prelocked original inode"
);
migration
.write_authority
.verify_database_authority()
.expect("restored canonical inode remains authoritative");
}
#[test]
fn reviewed_apply_refuses_stale_plan_before_creating_a_run() {
let (_temp, migration) = reviewed_v14_migration_context();
let plan = build_plan(&migration.db_path).expect("build plan");
let token = plan.plan_token.expect("plan token");
let conn =
Connection::open(migration.db_path.to_string_lossy().into_owned()).expect("open db");
conn.execute(
"UPDATE issues SET title = 'Changed after plan' WHERE id = 'bd-schema-rehearsal'",
)
.expect("change planned state");
close_connection(conn).expect("close changed db");
let error = execute_apply(
&DoctorMigrateSchemaApplyArgs {
plan_token: token,
json: false,
},
&migration,
)
.expect_err("stale token must be refused");
assert!(error.to_string().contains("plan token is stale"), "{error}");
assert!(
!migration_runs_root(&migration.beads_dir).exists(),
"a stale token must be refused before allocating recovery artifacts"
);
}
#[test]
#[cfg(any(
target_os = "linux",
target_os = "android",
target_vendor = "apple",
windows
))]
fn failed_post_adoption_verification_restores_database_and_authority() {
let (_temp, migration) = reviewed_v14_migration_context();
let raw_before = raw_family_witness(&migration.db_path).expect("original raw witness");
let candidate_path = migration.beads_dir.join("candidate.db");
let locked_stale_path = migration.beads_dir.join("locked-stale-candidate.db");
let displaced_main = migration.beads_dir.join("displaced-original.db");
fs::write(&candidate_path, b"candidate generation").expect("write candidate");
fs::write(&locked_stale_path, b"stale locked generation")
.expect("write stale locked generation");
let replacement_lock = migration
.write_authority
.lock_database_replacement_candidate(&locked_stale_path)
.expect("lock stale candidate inode");
let error = install_compacted_candidate(
&candidate_path,
&migration.db_path,
&displaced_main,
replacement_lock,
&migration.write_authority,
)
.expect_err("post-adoption identity mismatch must fail");
assert!(
error.to_string().contains("authority"),
"failure should identify replacement authority: {error}"
);
assert_eq!(
raw_family_witness(&migration.db_path).expect("restored raw witness"),
raw_before,
"the original database family must be restored exactly"
);
assert_eq!(
fs::read(&candidate_path).expect("restored candidate"),
b"candidate generation",
"the rejected candidate must return to its staging path"
);
assert!(
!displaced_main.exists(),
"failure before original retention must not fabricate a displaced database"
);
migration
.write_authority
.verify_database_authority()
.expect("restored original inode authority");
}
#[test]
#[cfg(any(target_os = "linux", target_os = "android", target_vendor = "apple"))]
fn pre_adoption_fence_rejects_same_byte_foreign_canonical_inode() {
let (_temp, migration) = reviewed_v14_migration_context();
let original_metadata = fs::metadata(&migration.db_path).expect("original metadata");
let foreign = migration.beads_dir.join("same-byte-foreign.db");
let retained_original = migration.beads_dir.join("externally-retained-original.db");
let candidate = migration.beads_dir.join("fenced-candidate.db");
let displaced = migration.beads_dir.join("must-remain-absent.db");
fs::copy(&migration.db_path, &foreign).expect("copy same-byte foreign inode");
fs::write(&candidate, b"candidate generation").expect("write candidate");
let replacement_lock = migration
.write_authority
.lock_database_replacement_candidate(&candidate)
.expect("lock candidate");
rename_path_no_replace(&migration.db_path, &retained_original)
.expect("stage original outside canonical path");
rename_path_no_replace(&foreign, &migration.db_path)
.expect("install same-byte foreign canonical inode");
let foreign_metadata = fs::metadata(&migration.db_path).expect("foreign metadata");
assert!(
!same_file_identity(&original_metadata, &foreign_metadata),
"fixture must use a distinct inode with identical bytes"
);
let failure = install_compacted_candidate(
&candidate,
&migration.db_path,
&displaced,
replacement_lock,
&migration.write_authority,
)
.expect_err("staged-original inode fence must reject a same-byte foreign target");
assert_eq!(
failure.disposition,
CompactedInstallFailureDisposition::LiveStateUncertain
);
assert!(
failure.to_string().contains("generation changed")
|| failure.to_string().contains("authority"),
"causal inode mismatch must remain visible: {failure}"
);
assert_eq!(
fs::read(&candidate).expect("candidate restored by compensating exchange"),
b"candidate generation"
);
assert!(
!displaced.exists(),
"a rejected foreign target must never be accepted as the displaced original"
);
assert!(
retained_original.is_file(),
"the true original stays retained"
);
}
#[test]
#[cfg(any(
target_os = "linux",
target_os = "android",
target_vendor = "apple",
windows
))]
fn post_install_directory_sync_failure_restores_original_main_and_authority() {
let (_temp, migration) = reviewed_v14_migration_context();
let raw_before = raw_family_witness(&migration.db_path).expect("original raw witness");
let candidate_path = migration.beads_dir.join("sync-failure-candidate.db");
let displaced_dir = migration.beads_dir.join("sync-failure-displaced");
fs::create_dir(&displaced_dir).expect("create displaced directory");
let displaced_main = displaced_dir.join("beads.db");
fs::write(&candidate_path, b"candidate generation").expect("write candidate");
let replacement_lock = migration
.write_authority
.lock_database_replacement_candidate(&candidate_path)
.expect("lock candidate inode");
let sync_calls = std::cell::Cell::new(0_u32);
let failure = install_compacted_candidate_with_sync(
&candidate_path,
&migration.db_path,
&displaced_main,
replacement_lock,
&migration.write_authority,
|_| {
let call = sync_calls.get();
sync_calls.set(call + 1);
if call == 0 {
Err(BeadsError::Io(std::io::Error::other(
"forced post-install directory sync failure",
)))
} else {
Ok(())
}
},
)
.expect_err("post-install sync failure must fail the installation");
assert_eq!(
failure.disposition,
CompactedInstallFailureDisposition::OriginalRestored,
"a successful compensating exchange and sync must prove restoration"
);
assert!(
failure
.to_string()
.contains("forced post-install directory sync failure"),
"the causal durability failure must remain visible: {failure}"
);
assert_eq!(
raw_family_witness(&migration.db_path).expect("restored raw witness"),
raw_before,
"the original main and sidecars must remain byte-exact"
);
migration
.write_authority
.verify_database_authority()
.expect("restored original inode authority");
#[cfg(any(target_os = "linux", target_os = "android", target_vendor = "apple"))]
assert_eq!(
fs::read(&displaced_main).expect("retained rejected candidate"),
b"candidate generation",
"the rejected replacement must be retained without overwriting another path"
);
#[cfg(not(any(target_os = "linux", target_os = "android", target_vendor = "apple")))]
assert_eq!(
fs::read(&candidate_path).expect("restored rejected candidate"),
b"candidate generation",
"the rejected replacement must return to its no-replace staging path"
);
}
#[test]
#[cfg(any(
target_os = "linux",
target_os = "android",
target_vendor = "apple",
windows
))]
fn repeated_directory_sync_failure_is_reported_as_uncertain() {
let (_temp, migration) = reviewed_v14_migration_context();
let candidate_path = migration.beads_dir.join("uncertain-sync-candidate.db");
let displaced_dir = migration.beads_dir.join("uncertain-sync-displaced");
fs::create_dir(&displaced_dir).expect("create displaced directory");
let displaced_main = displaced_dir.join("beads.db");
fs::write(&candidate_path, b"candidate generation").expect("write candidate");
let replacement_lock = migration
.write_authority
.lock_database_replacement_candidate(&candidate_path)
.expect("lock candidate inode");
let failure = install_compacted_candidate_with_sync(
&candidate_path,
&migration.db_path,
&displaced_main,
replacement_lock,
&migration.write_authority,
|_| {
Err(BeadsError::Io(std::io::Error::other(
"forced persistent directory sync failure",
)))
},
)
.expect_err("an unproved rollback must fail closed");
assert_eq!(
failure.disposition,
CompactedInstallFailureDisposition::LiveStateUncertain,
"the caller must not restore original sidecars without a durable rollback proof"
);
assert!(
failure.to_string().contains("could not prove"),
"uncertain disposition must explain the missing proof: {failure}"
);
}
#[test]
#[cfg(any(target_os = "linux", target_os = "android", target_vendor = "apple"))]
#[allow(
clippy::too_many_lines,
reason = "the crash-resume fixture keeps each persisted state transition visible in order"
)]
fn commit_ready_marker_resumes_applied_receipt_and_normal_undo() {
let (_temp, migration) = reviewed_v14_migration_context();
let original_logical = logical_witness(&migration.db_path).expect("original witness");
let plan = build_plan(&migration.db_path).expect("build migration plan");
let forecast = plan.forecast.clone().expect("eligible forecast");
let plan_token = plan.plan_token.clone().expect("plan token");
let run_id = allocate_run_id(&migration.beads_dir).expect("allocate run");
let run_dir = migration_runs_root(&migration.beads_dir).join(&run_id);
let before_dir = run_dir.join("before");
ensure_new_directory(&before_dir).expect("create before directory");
copy_family_to_backup(&migration.db_path, &before_dir, &plan.raw_witness)
.expect("copy recovery family");
verify_backup_family(&migration.db_path, &before_dir, &plan.raw_witness)
.expect("verify recovery family");
let marked_at = Utc::now().to_rfc3339();
let prepared = PreparedMigrationReceipt {
schema_version: PREPARED_SCHEMA.to_string(),
run_id: run_id.clone(),
database_path: plan.database_path.clone(),
plan_token: plan_token.clone(),
marked_at: marked_at.clone(),
forecast: forecast.clone(),
raw_before: plan.raw_witness,
logical_before: plan.logical_witness,
};
write_json_new(&run_dir.join("prepared.json"), &prepared).expect("write prepared receipt");
let candidate_path =
maintenance_candidate_path(&migration.db_path, &run_dir).expect("candidate path");
let source_conn = Connection::open(migration.db_path.to_string_lossy().into_owned())
.expect("open source");
let escaped_candidate = candidate_path.to_string_lossy().replace('\'', "''");
source_conn
.execute(&format!("VACUUM INTO '{escaped_candidate}'"))
.expect("build candidate");
close_connection(source_conn).expect("close source");
let candidate_conn = Connection::open(candidate_path.to_string_lossy().into_owned())
.expect("open candidate");
candidate_conn
.execute("BEGIN IMMEDIATE")
.expect("begin migration");
let reviewed_effects = run_reviewed_schema_migration_steps_in_transaction(
&candidate_conn,
forecast.from_version,
forecast.to_version,
&marked_at,
)
.expect("run reviewed migration");
candidate_conn.execute("COMMIT").expect("commit candidate");
candidate_conn
.execute("REINDEX")
.expect("reindex candidate");
candidate_conn
.execute("PRAGMA wal_checkpoint(TRUNCATE)")
.expect("checkpoint candidate");
close_connection(candidate_conn).expect("close candidate");
let candidate_logical = logical_witness(&candidate_path).expect("candidate witness");
let mut effects = ReviewedSchemaMigrationEffectsReceipt::from(reviewed_effects);
effects.post_migration_maintenance_completed = true;
persist_commit_ready_marker(&run_dir, &migration.db_path, &candidate_logical, effects)
.expect("persist commit-ready marker");
let marker: CommitReadyMigrationReceipt =
read_json(&run_dir.join("commit-ready.json")).expect("read marker");
let failed = FailedMigrationReceipt {
schema_version: FAILED_SCHEMA.to_string(),
run_id: marker.run_id.clone(),
database_path: marker.database_path.clone(),
plan_token: marker.plan_token.clone(),
marked_at: marker.marked_at.clone(),
error: "simulated process failure after namespace installation".to_string(),
failed_stage: None,
raw_before: marker.raw_before.clone(),
logical_before: marker.logical_before.clone(),
raw_observed_after_failure: None,
logical_observed_after_failure: None,
};
write_json_new(&run_dir.join("failed.json"), &failed).expect("write failed receipt");
let source_raw = raw_family_witness(&migration.db_path).expect("source raw");
let candidate_raw = raw_family_witness(&candidate_path).expect("candidate raw");
move_present_sidecars_new(
&candidate_path,
&run_dir.join("maintenance-candidate-sidecars"),
&candidate_raw,
"candidate-sidecar-retention",
)
.expect("retain candidate sidecars");
let displaced_dir = run_dir.join("maintenance-displaced");
ensure_new_directory(&displaced_dir).expect("create displaced directory");
move_present_sidecars_new(
&migration.db_path,
&displaced_dir,
&source_raw,
"displace-live-sidecars",
)
.expect("retain source sidecars");
let replacement_lock = migration
.write_authority
.lock_database_replacement_candidate(&candidate_path)
.expect("lock candidate");
exchange_database_paths(&candidate_path, &migration.db_path)
.expect("exchange candidate and live main");
migration
.write_authority
.verify_staged_database_recovery_authority(&candidate_path)
.expect("prove displaced original authority");
migration
.write_authority
.adopt_locked_database_replacement(replacement_lock)
.expect("adopt installed candidate");
let displaced_main =
backup_component_path(&displaced_dir, &migration.db_path, "").expect("displaced main");
rename_path_no_replace(&candidate_path, &displaced_main)
.expect("retain original main without running barriers");
assert!(run_dir.join("commit-ready.json").is_file());
assert!(
!run_dir.join("applied.json").exists(),
"the fixture must stop at the crash boundary before applied receipt publication"
);
let beads_dir = migration.beads_dir.clone();
let db_path = migration.db_path.clone();
drop(migration);
let migration = MigrationContext {
write_authority: Arc::new(
crate::sync::blocking_database_family_write_lock_with_timeout(
&beads_dir,
&db_path,
Some(1000),
)
.expect("reacquire authority after simulated crash"),
),
beads_dir,
db_path,
};
let (applied, resumed_before_dir) = resume_commit_ready_migration(&plan_token, &migration)
.expect("resume commit-ready migration")
.expect("committed generation must materialize applied receipt");
assert_eq!(resumed_before_dir, before_dir);
assert!(applied.attested);
assert!(run_dir.join("applied.json").is_file());
assert!(
run_dir.join("failed.json").is_file(),
"a pre-existing failure receipt must not suppress promotion of the exact live after-state"
);
execute_undo(
&DoctorMigrateSchemaUndoArgs {
run_id,
dry_run: false,
json: false,
},
&migration,
)
.expect("normal undo after resumed receipt");
assert_eq!(
logical_witness(&migration.db_path).expect("restored original witness"),
original_logical
);
}
#[test]
fn pre_install_commit_ready_marker_classifies_original_generation_for_retry() {
let (_temp, migration) = reviewed_v14_migration_context();
let plan = build_plan(&migration.db_path).expect("build migration plan");
let forecast = plan.forecast.clone().expect("eligible forecast");
let plan_token = plan.plan_token.clone().expect("plan token");
let run_id = allocate_run_id(&migration.beads_dir).expect("allocate run");
let run_dir = migration_runs_root(&migration.beads_dir).join(&run_id);
let before_dir = run_dir.join("before");
ensure_new_directory(&before_dir).expect("create before directory");
copy_family_to_backup(&migration.db_path, &before_dir, &plan.raw_witness)
.expect("copy recovery family");
let marked_at = Utc::now().to_rfc3339();
let prepared = PreparedMigrationReceipt {
schema_version: PREPARED_SCHEMA.to_string(),
run_id: run_id.clone(),
database_path: plan.database_path,
plan_token: plan_token.clone(),
marked_at,
forecast: forecast.clone(),
raw_before: plan.raw_witness,
logical_before: plan.logical_witness.clone(),
};
write_json_new(&run_dir.join("prepared.json"), &prepared).expect("write prepared receipt");
let mut expected_after = plan.logical_witness;
expected_after.user_version = forecast.to_version;
expected_after.integrity_check = "ok".to_string();
persist_commit_ready_marker(
&run_dir,
&migration.db_path,
&expected_after,
ReviewedSchemaMigrationEffectsReceipt {
from_version: forecast.from_version,
to_version: forecast.to_version,
content_hash_rows_rebuilt: forecast.content_hash_rows_rebuilt,
gate_result_history_created: forecast.gate_result_history_created,
prerequisites_column_added: forecast.prerequisites_column_added,
dependency_type_key_added: forecast.dependency_type_key_added,
legacy_v15_tables_converted: forecast.legacy_v15_conversion.is_some(),
post_migration_maintenance_completed: true,
},
)
.expect("persist pre-install commit-ready marker");
let resumed = resume_commit_ready_migration(&plan_token, &migration)
.expect("classify interrupted pre-install intent");
assert!(
resumed.is_none(),
"an unchanged original generation must be retried, never called applied"
);
assert!(run_dir.join("failed.json").is_file());
assert!(!run_dir.join("applied.json").exists());
assert_eq!(
logical_witness(&migration.db_path).expect("unchanged original witness"),
prepared.logical_before
);
}
#[test]
#[cfg(any(
target_os = "linux",
target_os = "android",
target_vendor = "apple",
windows
))]
fn interrupted_preinstall_sidecar_move_restores_exact_receipt_family() {
let (_temp, migration) = reviewed_v14_migration_context();
let journal = family_component_path(&migration.db_path, "-journal");
fs::write(&journal, b"receipt-bound journal bytes").expect("write journal fixture");
let expected = raw_family_witness(&migration.db_path).expect("expected raw family");
let run_dir = migration.beads_dir.join("preinstall-sidecar-crash");
let before_dir = run_dir.join("before");
let displaced_dir = run_dir.join("maintenance-displaced");
ensure_new_directory(&before_dir).expect("create before dir");
copy_family_to_backup(&migration.db_path, &before_dir, &expected)
.expect("copy exact family");
ensure_new_directory(&displaced_dir).expect("create displaced dir");
let displaced_journal =
backup_component_path(&displaced_dir, &migration.db_path, "-journal")
.expect("displaced journal path");
rename_path_no_replace(&journal, &displaced_journal)
.expect("simulate sidecar move before main exchange");
assert!(!journal.exists(), "fixture must expose the sidecar gap");
let restored = restore_interrupted_preinstall_family(
&migration.db_path,
&before_dir,
&displaced_dir,
&expected,
&migration.write_authority,
)
.expect("reconcile interrupted sidecar move");
assert_eq!(restored, expected);
assert_eq!(
raw_family_witness(&migration.db_path).expect("live exact family"),
expected
);
migration
.write_authority
.verify_database_authority()
.expect("original main remains authoritative");
}
#[test]
fn reviewed_undo_refuses_a_changed_post_migration_database() {
let (_temp, migration) = reviewed_v14_migration_context();
let plan = build_plan(&migration.db_path).expect("build plan");
execute_apply(
&DoctorMigrateSchemaApplyArgs {
plan_token: plan.plan_token.expect("plan token"),
json: false,
},
&migration,
)
.expect("apply reviewed migration");
let run_dir = fs::read_dir(migration_runs_root(&migration.beads_dir))
.expect("read migration runs")
.next()
.expect("one run")
.expect("run entry")
.path();
let run_id = run_dir
.file_name()
.and_then(|value| value.to_str())
.expect("UTF-8 run id")
.to_string();
let conn =
Connection::open(migration.db_path.to_string_lossy().into_owned()).expect("open db");
conn.execute(
"UPDATE issues SET title = 'New work after migration' \
WHERE id = 'bd-schema-rehearsal'",
)
.expect("mutate post-migration database");
close_connection(conn).expect("close changed db");
let error = execute_undo(
&DoctorMigrateSchemaUndoArgs {
run_id,
dry_run: false,
json: false,
},
&migration,
)
.expect_err("undo must refuse newer tracker work");
assert!(
error.to_string().contains("live database has changed"),
"{error}"
);
assert!(
!run_dir.join("undo-prepared.json").exists(),
"refused undo must not start a recovery state machine"
);
}
#[cfg(unix)]
#[test]
fn reviewed_undo_accepts_raw_layout_churn_when_logical_state_is_exact() {
use std::os::unix::fs::PermissionsExt;
let (_temp, migration) = reviewed_v14_migration_context();
let plan = build_plan(&migration.db_path).expect("build plan");
execute_apply(
&DoctorMigrateSchemaApplyArgs {
plan_token: plan.plan_token.expect("plan token"),
json: false,
},
&migration,
)
.expect("apply reviewed migration");
let run_dir = fs::read_dir(migration_runs_root(&migration.beads_dir))
.expect("read migration runs")
.next()
.expect("one run")
.expect("run entry")
.path();
let run_id = run_dir
.file_name()
.and_then(|value| value.to_str())
.expect("UTF-8 run id")
.to_string();
fs::set_permissions(&migration.db_path, fs::Permissions::from_mode(0o640))
.expect("change only raw file metadata");
execute_undo(
&DoctorMigrateSchemaUndoArgs {
run_id,
dry_run: true,
json: false,
},
&migration,
)
.expect("logical equality must outrank raw layout churn");
assert_eq!(
logical_witness(&migration.db_path)
.expect("post-migration logical witness")
.user_version,
crate::storage::schema::CURRENT_SCHEMA_VERSION as u32
);
}
#[test]
#[cfg(any(target_os = "linux", target_os = "android", target_vendor = "apple"))]
fn reviewed_undo_resumes_after_partial_quarantine() {
let (_temp, migration) = reviewed_v14_migration_context();
let original = build_plan(&migration.db_path)
.expect("build plan")
.logical_witness;
let plan = build_plan(&migration.db_path).expect("build plan");
execute_apply(
&DoctorMigrateSchemaApplyArgs {
plan_token: plan.plan_token.expect("plan token"),
json: false,
},
&migration,
)
.expect("apply reviewed migration");
let run_dir = fs::read_dir(migration_runs_root(&migration.beads_dir))
.expect("read migration runs")
.next()
.expect("one run")
.expect("run entry")
.path();
let run_id = run_dir
.file_name()
.and_then(|value| value.to_str())
.expect("UTF-8 run id")
.to_string();
let applied_path = run_dir.join("applied.json");
let applied: AppliedMigrationReceipt =
read_json(&applied_path).expect("read applied receipt");
let logical_live = logical_witness(&migration.db_path).expect("logical live state");
let raw_live = raw_family_witness(&migration.db_path).expect("raw live state");
let quarantine_dir = run_dir.join("undo-quarantine/undo-resume-fixture");
let receipt = UndoReceipt {
schema_version: UNDO_SCHEMA.to_string(),
run_id: run_id.clone(),
dry_run: false,
database_path: migration.db_path.display().to_string(),
quarantine_path: quarantine_dir.display().to_string(),
applied_receipt_sha256: file_sha256(&applied_path).expect("applied hash"),
raw_expected_before: applied.raw_before.clone(),
logical_expected_before: applied.logical_before.clone(),
raw_live_before_undo: raw_live,
logical_live_before_undo: Some(logical_live),
raw_restored: None,
logical_restored: None,
};
ensure_directory(quarantine_dir.parent().expect("quarantine parent"))
.expect("create quarantine parent");
write_json_new(&run_dir.join("undo-prepared.json"), &receipt).expect("write prepared undo");
ensure_new_directory(&quarantine_dir).expect("create quarantine");
let restored_candidate_dir = run_dir.join("undo-restored-candidate");
ensure_directory(&restored_candidate_dir).expect("create restore candidate dir");
let restored_candidate =
backup_component_path(&restored_candidate_dir, &migration.db_path, "")
.expect("restored candidate path");
let before_main = backup_component_path(&run_dir.join("before"), &migration.db_path, "")
.expect("before main");
let restored_main = component_for_suffix(&applied.raw_before, "").expect("main witness");
copy_regular_file_new(&before_main, &restored_candidate, restored_main.unix_mode)
.expect("stage restored main");
exchange_database_paths(&restored_candidate, &migration.db_path)
.expect("simulate atomic undo exchange before quarantine retention");
let beads_dir = migration.beads_dir.clone();
let db_path = migration.db_path.clone();
drop(migration);
let migration = MigrationContext {
write_authority: Arc::new(
crate::sync::blocking_database_family_write_lock_with_timeout(
&beads_dir,
&db_path,
Some(1000),
)
.expect("reacquire authority after simulated undo crash"),
),
beads_dir,
db_path,
};
execute_undo(
&DoctorMigrateSchemaUndoArgs {
run_id,
dry_run: false,
json: false,
},
&migration,
)
.expect("resume interrupted undo");
assert_eq!(
logical_witness(&migration.db_path).expect("restored logical state"),
original
);
assert!(run_dir.join("undone.json").is_file());
}
#[cfg(unix)]
#[test]
fn migration_recovery_artifacts_are_private() {
use std::os::unix::fs::PermissionsExt;
let (_temp, migration) = reviewed_v14_migration_context();
let plan = build_plan(&migration.db_path).expect("build plan");
execute_apply(
&DoctorMigrateSchemaApplyArgs {
plan_token: plan.plan_token.expect("plan token"),
json: false,
},
&migration,
)
.expect("apply reviewed migration");
let run_dir = fs::read_dir(migration_runs_root(&migration.beads_dir))
.expect("read migration runs")
.next()
.expect("one run")
.expect("run entry")
.path();
for directory in [
&migration.beads_dir.join(".br_recovery"),
&migration_runs_root(&migration.beads_dir),
&run_dir,
&run_dir.join("before"),
] {
assert_eq!(
fs::metadata(directory)
.expect("artifact directory metadata")
.permissions()
.mode()
& 0o777,
0o700
);
}
for file in [
run_dir.join("prepared.json"),
run_dir.join("applied.json"),
run_dir.join("before/beads.db"),
] {
assert_eq!(
fs::metadata(file)
.expect("artifact file metadata")
.permissions()
.mode()
& 0o777,
0o600
);
}
}
#[test]
fn contention_retry_recovers_after_transient_sharing_conflicts() {
let sleeps = std::cell::RefCell::new(Vec::new());
let attempts = std::cell::Cell::new(0_u32);
retry_through_file_contention(
"candidate-sidecar-retention",
Path::new("candidate.db-wal"),
Path::new("retained.db-wal"),
|delay| sleeps.borrow_mut().push(delay),
|_error| true,
|| {
let attempt = attempts.get();
attempts.set(attempt + 1);
if attempt < 2 {
Err(BeadsError::Io(std::io::Error::other(
"injected transient sharing conflict",
)))
} else {
Ok(())
}
},
)
.expect("the attempt after the transient conflicts clear must succeed");
assert_eq!(attempts.get(), 3, "two failures then one success");
assert_eq!(
sleeps.borrow().as_slice(),
&[
FILE_CONTENTION_RETRY_STEP,
FILE_CONTENTION_RETRY_STEP.saturating_mul(2),
],
"backoff must grow linearly with the attempt number"
);
}
#[test]
fn contention_retry_exhaustion_names_stage_and_both_paths() {
let sleeps = std::cell::Cell::new(0_u32);
let attempts = std::cell::Cell::new(0_u32);
let error = retry_through_file_contention(
"install-rename-candidate-to-live",
Path::new("pinned-source.db"),
Path::new("blocked-destination.db"),
|_delay| sleeps.set(sleeps.get() + 1),
|_error| true,
|| {
attempts.set(attempts.get() + 1);
Err(BeadsError::Io(std::io::Error::other(
"injected persistent sharing conflict",
)))
},
)
.expect_err("a persistent sharing conflict must exhaust the retry budget");
assert_eq!(attempts.get(), FILE_CONTENTION_RETRY_ATTEMPTS);
assert_eq!(
sleeps.get(),
FILE_CONTENTION_RETRY_ATTEMPTS - 1,
"no sleep may follow the final attempt"
);
let message = error.to_string();
for needle in [
"install-rename-candidate-to-live",
"pinned-source.db",
"blocked-destination.db",
"injected persistent sharing conflict",
] {
assert!(
message.contains(needle),
"exhaustion error must name {needle}: {message}"
);
}
}
#[test]
fn non_contention_rename_errors_pass_through_without_retry() {
let attempts = std::cell::Cell::new(0_u32);
let error = retry_through_file_contention(
"displace-live-sidecars",
Path::new("source.db-journal"),
Path::new("destination.db-journal"),
|_delay| panic!("a non-contention error must never sleep"),
is_windows_file_contention,
|| {
attempts.set(attempts.get() + 1);
Err(BeadsError::internal("hard rename failure"))
},
)
.expect_err("a non-contention error must surface unchanged");
assert_eq!(attempts.get(), 1, "no retry may follow a hard failure");
assert!(
error.to_string().contains("hard rename failure"),
"the causal error must pass through unchanged: {error}"
);
}
#[test]
fn failed_receipt_stage_marker_is_additive_and_optional() {
let witness = LogicalDatabaseWitness {
user_version: 14,
integrity_check: "ok".to_string(),
schema_sha256: String::new(),
contents_sha256: String::new(),
tables: Vec::new(),
};
let mut receipt = FailedMigrationReceipt {
schema_version: FAILED_SCHEMA.to_string(),
run_id: "run".to_string(),
database_path: "beads.db".to_string(),
plan_token: "token".to_string(),
marked_at: "2026-08-25T00:00:00Z".to_string(),
error: "boom".to_string(),
failed_stage: None,
raw_before: RawFamilyWitness {
components: Vec::new(),
},
logical_before: witness,
raw_observed_after_failure: None,
logical_observed_after_failure: None,
};
let without_stage = serde_json::to_string(&receipt).expect("serialize stage-less receipt");
assert!(
!without_stage.contains("failed_stage"),
"an absent stage must not change the receipt encoding: {without_stage}"
);
let decoded: FailedMigrationReceipt =
serde_json::from_str(&without_stage).expect("pre-marker receipts must deserialize");
assert_eq!(decoded.failed_stage, None);
receipt.failed_stage = Some("candidate-sidecar-retention".to_string());
let with_stage = serde_json::to_string(&receipt).expect("serialize staged receipt");
assert!(
with_stage.contains("\"failed_stage\":\"candidate-sidecar-retention\""),
"a recorded stage must round-trip: {with_stage}"
);
}
#[test]
fn maintenance_failure_records_the_stage_that_was_in_progress() {
let (_temp, migration) = reviewed_v14_migration_context();
let plan = build_plan(&migration.db_path).expect("build plan");
let forecast = plan.forecast.clone().expect("eligible forecast");
let run_id = allocate_run_id(&migration.beads_dir).expect("allocate run");
let run_dir = migration_runs_root(&migration.beads_dir).join(&run_id);
let marked_at = Utc::now().to_rfc3339();
let prepared = PreparedMigrationReceipt {
schema_version: PREPARED_SCHEMA.to_string(),
run_id: run_id.clone(),
database_path: plan.database_path,
plan_token: plan.plan_token.expect("plan token"),
marked_at: marked_at.clone(),
forecast: forecast.clone(),
raw_before: plan.raw_witness,
logical_before: plan.logical_witness,
};
write_json_new(&run_dir.join("prepared.json"), &prepared).expect("write prepared receipt");
let before_dir = run_dir.join("before");
ensure_new_directory(&before_dir).expect("create before backup dir");
copy_family_to_backup(&migration.db_path, &before_dir, &prepared.raw_before)
.expect("retain the before family");
verify_backup_family(&migration.db_path, &before_dir, &prepared.raw_before)
.expect("verify the retained before family");
fs::write(run_dir.join("commit-ready.json"), b"{}").expect("occupy marker path");
let mut failed_stage = None;
run_post_migration_maintenance(
&migration.db_path,
&forecast,
&marked_at,
&run_dir,
&prepared.raw_before,
&migration.write_authority,
&mut failed_stage,
)
.expect_err("an occupied commit-ready marker path must fail the pipeline");
assert_eq!(
failed_stage.as_deref(),
Some("commit-ready-marker"),
"the marker must name the stage that was in progress"
);
}
#[cfg(unix)]
#[test]
fn forced_apply_failure_records_failed_stage_in_failure_receipt() {
let database_name = format!("{}.db", "b".repeat(226));
let (_temp, migration) = reviewed_v14_migration_context_with_database_name(&database_name);
let plan = build_plan(&migration.db_path).expect("build plan");
let runs_root = migration_runs_root(&migration.beads_dir);
let result = execute_apply(
&DoctorMigrateSchemaApplyArgs {
plan_token: plan.plan_token.expect("plan token"),
json: false,
},
&migration,
);
result.expect_err("an overlong VACUUM candidate name must fail the apply");
let run_dir = fs::read_dir(&runs_root)
.expect("read migration runs")
.next()
.expect("one run")
.expect("run entry")
.path();
let failed: FailedMigrationReceipt =
read_json(&run_dir.join("failed.json")).expect("read failure receipt");
assert_eq!(
failed.failed_stage.as_deref(),
Some("vacuum-candidate"),
"the receipt must pinpoint the stage that could not proceed"
);
assert!(
!run_dir.join("maintenance-candidate-sidecars").exists(),
"the failure precedes candidate sidecar retention"
);
}
}