#[cfg(unix)]
use std::os::unix::fs::{MetadataExt, OpenOptionsExt};
use std::{
error::Error,
fmt, fs,
future::Future,
io::{Read, Write},
path::{Path, PathBuf},
process,
sync::{
atomic::{AtomicBool, AtomicU64, Ordering},
Arc, Mutex, MutexGuard,
},
time::{Duration, Instant},
};
use rhiza_archive::{
CheckpointIdentity, CheckpointPublisher, CheckpointPublisherOptions, CheckpointTip,
ObjectArchiveStore, RestoredCheckpoint,
};
#[cfg(any(feature = "graph", feature = "kv"))]
use rhiza_core::SnapshotIdentity;
use rhiza_core::{
ConfigChange, ConfigurationState, ExecutionProfile, LogAnchor, LogEntry, LogHash, LogIndex,
RecoveryAnchor, StopBinding,
};
#[cfg(feature = "graph")]
use rhiza_graph::{
decode_snapshot as decode_graph_snapshot, encode_snapshot as encode_graph_snapshot,
restore_snapshot_file as restore_graph_snapshot_file,
};
#[cfg(feature = "kv")]
use rhiza_kv::{
decode_snapshot as decode_kv_snapshot, encode_snapshot as encode_kv_snapshot,
restore_snapshot_file as restore_kv_snapshot_file, RedbStateMachine,
};
use rhiza_log::{FileLogStore, IndexRange, LogStore};
use rhiza_quepaxa::Membership;
#[cfg(feature = "sql")]
use rhiza_sql::{restore_recovery_snapshot_file, sql_executor_fingerprint};
use serde::{Deserialize, Serialize};
use crate::{Materializer, NodeConfig, NodeRuntime, StopInformation};
const FLUSH_BATCH_ENTRIES: LogIndex = 32;
const SYNC_COMPACTION_POLL_INTERVAL: Duration = Duration::from_millis(100);
const RESTORE_INTENT_FILE: &str = ".rhiza-restore.json";
const RESTORE_STAGING_PREFIX: &str = ".restore-stage-";
const RESTORE_MARKER_TMP_PREFIX: &str = ".restore-marker-tmp-";
const SUCCESSOR_RESTORE_LOCK_FILE: &str = ".successor-restore.lock";
const SUCCESSOR_RESTORE_INTENT_FILE: &str = ".successor-restore.intent";
const SUCCESSOR_RESTORE_COMPLETE_FILE: &str = ".successor-restore.complete";
const SUCCESSOR_PRESTAGE_LOCK_FILE: &str = ".successor-prestage.lock";
pub(crate) const SUCCESSOR_PRESTAGE_INTENT_FILE: &str = ".successor-prestage.intent";
pub(crate) const SUCCESSOR_PRESTAGE_READY_FILE: &str = ".successor-prestage.ready";
const SUCCESSOR_PRESTAGE_PUBLISHED_FILE: &str = ".successor-prestage.published";
const SUCCESSOR_PRESTAGE_FINALIZED_FILE: &str = ".successor-prestage.finalized";
const REPAIR_ARTIFACT_OWNER_FILE: &str = ".rhiza-recovery-owner.json";
pub const LOCAL_CHECKPOINT_IDENTITY_FILE: &str = ".rhiza-checkpoint-identity.json";
static RESTORE_SEQUENCE: AtomicU64 = AtomicU64::new(0);
#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
struct RestoreIntentIdentity {
cluster_id: String,
node_id: String,
execution_profile: ExecutionProfile,
epoch: u64,
config_id: u64,
recovery_generation: u64,
checkpoint_index: LogIndex,
checkpoint_hash: String,
}
#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", content = "identity", rename_all = "snake_case")]
enum RecoveryArtifactIdentity {
Prestage(SuccessorPrestageIdentity),
Restore(RestoreIntentIdentity),
}
#[derive(Clone, Copy, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
enum RepairArtifactRole {
Staging,
Quarantine,
}
#[derive(Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
struct RepairArtifactOwnership {
role: RepairArtifactRole,
name: String,
identity: RecoveryArtifactIdentity,
}
#[derive(Serialize)]
struct SuccessorRestoreIdentity<'a> {
cluster_id: &'a str,
epoch: u64,
target_config_id: u64,
recovery_generation: u64,
node_id: &'a str,
membership_digest: String,
predecessor_config_id: u64,
stop_index: LogIndex,
stop_hash: String,
}
#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
struct SuccessorRestoreReceipt {
cluster_id: String,
epoch: u64,
target_config_id: u64,
recovery_generation: u64,
node_id: String,
membership_digest: String,
predecessor_config_id: u64,
stop_index: LogIndex,
stop_hash: String,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SuccessorPrestageIdentity {
cluster_id: String,
epoch: u64,
predecessor_config_id: u64,
predecessor_membership_digest: String,
predecessor_recovery_generation: u64,
node_id: String,
execution_profile: ExecutionProfile,
target_config_id: u64,
target_membership_digest: String,
seed_index: LogIndex,
seed_hash: String,
}
impl SuccessorPrestageIdentity {
pub fn cluster_id(&self) -> &str {
&self.cluster_id
}
pub const fn epoch(&self) -> u64 {
self.epoch
}
pub const fn predecessor_config_id(&self) -> u64 {
self.predecessor_config_id
}
pub fn predecessor_membership_digest(&self) -> LogHash {
LogHash::from_hex(&self.predecessor_membership_digest)
.expect("validated predecessor membership digest")
}
pub const fn predecessor_recovery_generation(&self) -> u64 {
self.predecessor_recovery_generation
}
pub fn node_id(&self) -> &str {
&self.node_id
}
pub const fn execution_profile(&self) -> ExecutionProfile {
self.execution_profile
}
pub const fn target_config_id(&self) -> u64 {
self.target_config_id
}
pub fn target_membership_digest(&self) -> LogHash {
LogHash::from_hex(&self.target_membership_digest)
.expect("validated successor prestage membership digest")
}
pub fn seed_anchor(&self) -> LogAnchor {
LogAnchor::new(
self.seed_index,
LogHash::from_hex(&self.seed_hash).expect("validated successor prestage seed hash"),
)
}
fn checkpoint_identity(&self) -> CheckpointIdentity {
CheckpointIdentity::new(
self.cluster_id.clone(),
self.epoch,
self.predecessor_config_id,
self.predecessor_recovery_generation,
)
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum SuccessorPrestageState {
Preparing,
Ready,
Published,
Finalized,
}
#[derive(Debug)]
pub struct SuccessorPrestage {
path: PathBuf,
identity: SuccessorPrestageIdentity,
state: SuccessorPrestageState,
_lock: fs::File,
}
impl SuccessorPrestage {
pub fn path(&self) -> &Path {
&self.path
}
pub const fn identity(&self) -> &SuccessorPrestageIdentity {
&self.identity
}
pub const fn state(&self) -> SuccessorPrestageState {
self.state
}
}
pub struct SuccessorRestorePreparation {
tip: CheckpointTip,
data_dir: PathBuf,
identity: Vec<u8>,
requires_recorder_install: bool,
_lock: fs::File,
}
impl SuccessorRestorePreparation {
pub const fn tip(&self) -> CheckpointTip {
self.tip
}
pub const fn requires_recorder_install(&self) -> bool {
self.requires_recorder_install
}
pub fn complete(mut self) -> Result<CheckpointTip, DurabilityError> {
if !self.requires_recorder_install {
return Ok(self.tip);
}
complete_adopted_successor_prestage(&self.data_dir, &self.identity)?;
self.requires_recorder_install = false;
Ok(self.tip)
}
}
pub fn complete_adopted_successor_prestage(
data_dir: &Path,
expected_identity: &[u8],
) -> Result<(), DurabilityError> {
let intent = data_dir.join(SUCCESSOR_RESTORE_INTENT_FILE);
let actual = read_regular_successor_control_file(&intent)?.ok_or_else(|| {
DurabilityError::SnapshotVerification("successor restore intent is missing".into())
})?;
if parse_successor_restore_receipt(&actual).is_none()
|| parse_successor_restore_receipt(expected_identity).is_none()
|| actual != expected_identity
{
return Err(DurabilityError::SnapshotVerification(
"successor restore intent changed before completion".into(),
));
}
let complete = data_dir.join(SUCCESSOR_RESTORE_COMPLETE_FILE);
match fs::symlink_metadata(&complete) {
Ok(_) => {
return Err(DurabilityError::SnapshotVerification(
"successor restore completion target already exists".into(),
));
}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(error) => return Err(error.into()),
}
fs::rename(intent, complete)?;
sync_directory(data_dir)
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum DurabilityMode {
Sync,
Bounded { max_lag: Duration },
Periodic { interval: Duration },
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum DurabilityHealth {
Available,
Unavailable,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum CheckpointRestoreState {
None,
IdentityBound,
}
impl DurabilityMode {
pub fn validate(&self) -> Result<(), DurabilityError> {
match self {
Self::Sync => Ok(()),
Self::Bounded { max_lag } if max_lag.is_zero() => {
Err(DurabilityError::InvalidDuration { mode: "bounded" })
}
Self::Periodic { interval } if interval.is_zero() => {
Err(DurabilityError::InvalidDuration { mode: "periodic" })
}
Self::Bounded { .. } | Self::Periodic { .. } => Ok(()),
}
}
}
#[derive(Debug)]
pub enum DurabilityError {
InvalidDuration {
mode: &'static str,
},
MissingCheckpoint,
Unavailable,
LagExceeded {
committed_index: LogIndex,
durable_index: LogIndex,
max_lag: Duration,
},
ArchiveAheadOfLocal {
durable_index: LogIndex,
local_index: LogIndex,
},
SnapshotRequired {
anchor: Box<RecoveryAnchor>,
},
LocalLogGap {
expected: LogIndex,
actual: Option<LogIndex>,
},
LocalLogConflict {
index: LogIndex,
},
SnapshotVerification(String),
PreconditionFailed,
DataDirNotFresh(PathBuf),
Archive(rhiza_archive::Error),
Log(rhiza_log::Error),
Io(std::io::Error),
}
impl fmt::Display for DurabilityError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::InvalidDuration { mode } => {
write!(f, "{mode} durability duration must be non-zero")
}
Self::MissingCheckpoint => write!(f, "checkpoint manifest is missing"),
Self::Unavailable => write!(f, "sync durability is unavailable"),
Self::LagExceeded {
committed_index,
durable_index,
max_lag,
} => write!(
f,
"checkpoint lag exceeded {max_lag:?}: committed index {committed_index}, durable index {durable_index}"
),
Self::ArchiveAheadOfLocal {
durable_index,
local_index,
} => write!(
f,
"checkpoint tip {durable_index} is ahead of local qlog tip {local_index}"
),
Self::SnapshotRequired { anchor } => write!(
f,
"snapshot restore required at qlog anchor {} before checkpoint flush",
anchor.compacted().index()
),
Self::LocalLogGap { expected, actual } => {
write!(f, "local qlog gap: expected index {expected}, got {actual:?}")
}
Self::LocalLogConflict { index } => {
write!(f, "local qlog hash chain conflicts at index {index}")
}
Self::SnapshotVerification(message) => {
write!(f, "checkpoint snapshot verification failed: {message}")
}
Self::PreconditionFailed => write!(f, "checkpoint precondition failed"),
Self::DataDirNotFresh(path) => write!(
f,
"restore data directory contains existing state: {}",
path.display()
),
Self::Archive(error) => error.fmt(f),
Self::Log(error) => error.fmt(f),
Self::Io(error) => error.fmt(f),
}
}
}
impl Error for DurabilityError {
fn source(&self) -> Option<&(dyn Error + 'static)> {
match self {
Self::Archive(error) => Some(error),
Self::Log(error) => Some(error),
Self::Io(error) => Some(error),
_ => None,
}
}
}
impl From<rhiza_archive::Error> for DurabilityError {
fn from(error: rhiza_archive::Error) -> Self {
Self::Archive(error)
}
}
impl From<rhiza_log::Error> for DurabilityError {
fn from(error: rhiza_log::Error) -> Self {
Self::Log(error)
}
}
impl From<std::io::Error> for DurabilityError {
fn from(error: std::io::Error) -> Self {
Self::Io(error)
}
}
#[derive(Debug)]
enum PendingLag {
New(Instant),
Recovered,
}
#[derive(Debug)]
struct CoordinatorState {
durable_tip: CheckpointTip,
committed_index: LogIndex,
pending_lag: Option<PendingLag>,
health: DurabilityHealth,
}
pub struct CheckpointCoordinator {
store: ObjectArchiveStore,
publisher: CheckpointPublisher,
mode: DurabilityMode,
state: Mutex<CoordinatorState>,
successor_baseline_required: AtomicBool,
publication_attempts: AtomicU64,
}
struct RuntimeCheckpointSnapshot {
anchor: RecoveryAnchor,
archive_bytes: Vec<u8>,
}
#[cfg(any(feature = "graph", feature = "kv"))]
struct EngineSnapshotIdentity<'a> {
cluster_id: &'a str,
epoch: u64,
config_id: u64,
applied_index: LogIndex,
applied_hash: LogHash,
}
impl CheckpointCoordinator {
pub async fn open(
store: ObjectArchiveStore,
mode: DurabilityMode,
) -> Result<Self, DurabilityError> {
Self::open_with_holder(store, mode, "anonymous-node").await
}
pub async fn open_with_holder(
store: ObjectArchiveStore,
mode: DurabilityMode,
holder: impl AsRef<str>,
) -> Result<Self, DurabilityError> {
Self::open_with_holder_and_options(
store,
mode,
holder,
CheckpointPublisherOptions::default(),
)
.await
}
pub async fn open_with_holder_and_options(
store: ObjectArchiveStore,
mode: DurabilityMode,
holder: impl AsRef<str>,
publisher_options: CheckpointPublisherOptions,
) -> Result<Self, DurabilityError> {
mode.validate()?;
store
.load_checkpoint()
.await?
.ok_or(DurabilityError::MissingCheckpoint)?;
let identity = store.checkpoint_identity()?;
let holder = format!(
"checkpoint-coordinator-{}-{}-{}-{}-{}",
identity.cluster_id(),
identity.epoch(),
identity.config_id(),
identity.recovery_generation(),
holder.as_ref()
);
let publisher = store
.open_checkpoint_publisher(holder, publisher_options)
.await?;
let loaded = publisher.cached_checkpoint().await;
let durable_tip = *loaded.manifest().tip();
let restored = store.restore_checkpoint_state().await?;
let restored_tip = *restored.tip();
if restored_tip != durable_tip {
return Err(DurabilityError::Archive(
rhiza_archive::Error::InvalidCheckpoint(
"restored entries changed while verifying the loaded manifest".into(),
),
));
}
Ok(Self {
store,
publisher,
mode,
state: Mutex::new(CoordinatorState {
durable_tip,
committed_index: durable_tip.index(),
pending_lag: None,
health: DurabilityHealth::Available,
}),
successor_baseline_required: AtomicBool::new(false),
publication_attempts: AtomicU64::new(0),
})
}
pub const fn mode(&self) -> &DurabilityMode {
&self.mode
}
pub fn durable_tip(&self) -> CheckpointTip {
self.lock_state().durable_tip
}
pub async fn refresh_durable_tip(&self) -> Result<CheckpointTip, DurabilityError> {
let loaded = self.publisher.observe_checkpoint().await?;
let accepted = self.publisher.cache_observed_checkpoint(loaded).await?;
observe_durable_tip(&self.state, *accepted.manifest().tip())
}
pub fn health(&self) -> DurabilityHealth {
self.lock_state().health
}
#[doc(hidden)]
pub fn checkpoint_publication_attempts(&self) -> u64 {
self.publication_attempts.load(Ordering::Relaxed)
}
pub fn note_committed(&self, index: LogIndex) {
let mut state = self.lock_state();
if index <= state.committed_index {
return;
}
state.committed_index = index;
if index > state.durable_tip.index() && state.pending_lag.is_none() {
state.pending_lag = Some(PendingLag::New(Instant::now()));
}
}
pub fn note_recovered_committed(&self, index: LogIndex) {
let mut state = self.lock_state();
state.committed_index = state.committed_index.max(index);
if state.committed_index > state.durable_tip.index() {
state.pending_lag = Some(PendingLag::Recovered);
}
}
pub fn write_allowed(&self) -> Result<(), DurabilityError> {
if self.successor_baseline_required.load(Ordering::Acquire) {
return Err(DurabilityError::Unavailable);
}
if matches!(self.mode, DurabilityMode::Sync)
&& self.health() == DurabilityHealth::Unavailable
{
return Err(DurabilityError::Unavailable);
}
let DurabilityMode::Bounded { max_lag } = self.mode else {
return Ok(());
};
let state = self.lock_state();
let exceeded = state.committed_index > state.durable_tip.index()
&& match state.pending_lag {
Some(PendingLag::Recovered) => true,
Some(PendingLag::New(pending)) => pending.elapsed() >= max_lag,
None => false,
};
if exceeded {
return Err(DurabilityError::LagExceeded {
committed_index: state.committed_index,
durable_index: state.durable_tip.index(),
max_lag,
});
}
Ok(())
}
#[doc(hidden)]
pub fn require_successor_checkpoint_baseline(&self) {
self.successor_baseline_required
.store(true, Ordering::Release);
}
#[doc(hidden)]
pub fn successor_checkpoint_baseline_required(&self) -> bool {
self.successor_baseline_required.load(Ordering::Acquire)
}
#[doc(hidden)]
pub async fn establish_successor_checkpoint_baseline(
&self,
runtime: &NodeRuntime,
predecessor_stop: LogAnchor,
) -> Result<RecoveryAnchor, DurabilityError> {
let (snapshot, _fence) = {
let _commit = runtime.commit.lock().map_err(|_| {
DurabilityError::SnapshotVerification("commit mutex is poisoned".into())
})?;
runtime
.ensure_ready()
.map_err(|error| DurabilityError::SnapshotVerification(error.to_string()))?;
let configuration = runtime
.configuration_state()
.map_err(|error| DurabilityError::SnapshotVerification(error.to_string()))?;
if !configuration.is_active() {
return Err(DurabilityError::SnapshotVerification(
"successor checkpoint baseline requires the active target configuration".into(),
));
}
let stop_entry = runtime
.config
.predecessor_stop_entry
.as_ref()
.filter(|entry| {
LogAnchor::new(entry.index, entry.hash) == predecessor_stop
&& entry.recompute_hash() == entry.hash
})
.ok_or_else(|| {
DurabilityError::SnapshotVerification(
"successor checkpoint baseline Stop binding changed".into(),
)
})?;
let stopped = runtime
.config
.log_initial_configuration
.validate_entry(stop_entry)
.map_err(|error| DurabilityError::SnapshotVerification(error.to_string()))?;
let root = runtime
.log_root_unlocked()
.map_err(|error| DurabilityError::SnapshotVerification(error.to_string()))?;
let expected_activation_index =
predecessor_stop.index().checked_add(1).ok_or_else(|| {
DurabilityError::SnapshotVerification(
"successor Activate index cannot advance".into(),
)
})?;
if root.index() != expected_activation_index {
return Err(DurabilityError::SnapshotVerification(
"successor checkpoint baseline is not rooted at the Activate entry".into(),
));
}
let activation = runtime
.log_store
.read(root.index())?
.filter(|entry| {
entry.hash == root.hash() && entry.prev_hash == predecessor_stop.hash()
})
.ok_or_else(|| {
DurabilityError::SnapshotVerification(
"successor checkpoint baseline Activate entry is unavailable".into(),
)
})?;
let activated = stopped
.validate_entry(&activation)
.map_err(|error| DurabilityError::SnapshotVerification(error.to_string()))?;
if activated != configuration {
return Err(DurabilityError::SnapshotVerification(
"successor Activate entry does not produce the active target configuration"
.into(),
));
}
if runtime.checkpointing.swap(true, Ordering::AcqRel) {
return Err(DurabilityError::SnapshotVerification(
"checkpoint transition is already in progress".into(),
));
}
let fence = CheckpointFence(&runtime.checkpointing);
let (target, target_hash) = runtime
.ensure_materialized_tip()
.map_err(|error| DurabilityError::SnapshotVerification(error.to_string()))?;
if target != root.index() || target_hash != root.hash() {
return Err(DurabilityError::SnapshotVerification(
"materialized successor state does not match the Activate entry".into(),
));
}
let snapshot =
create_runtime_checkpoint_snapshot(runtime, target, target_hash, &configuration)?;
(snapshot, fence)
};
let expected_tip = CheckpointTip::new(
snapshot.anchor.compacted().index(),
snapshot.anchor.compacted().hash(),
);
let durable_tip = self.durable_tip();
if durable_tip == CheckpointTip::new(0, LogHash::ZERO) {
self.publisher
.publish_initial_checkpoint_snapshot(
snapshot.anchor.clone(),
&snapshot.archive_bytes,
)
.await?;
} else if durable_tip != expected_tip {
return Err(DurabilityError::SnapshotVerification(
"target checkpoint namespace conflicts with the successor baseline".into(),
));
}
let restored = self.store.restore_checkpoint_state().await?;
let published = restored.snapshot().ok_or_else(|| {
DurabilityError::SnapshotVerification(
"successor checkpoint baseline has no snapshot".into(),
)
})?;
if published.anchor() != &snapshot.anchor || published.bytes() != snapshot.archive_bytes {
return Err(DurabilityError::SnapshotVerification(
"successor checkpoint baseline read-back differs from the active runtime".into(),
));
}
{
let _commit = runtime.commit.lock().map_err(|_| {
DurabilityError::SnapshotVerification("commit mutex is poisoned".into())
})?;
runtime.log_store.compact_prefix(&snapshot.anchor)?;
runtime
.compact_embedded_log_before(snapshot.anchor.compacted().index())
.map_err(|error| DurabilityError::SnapshotVerification(error.to_string()))?;
}
self.mark_durable(expected_tip);
self.successor_baseline_required
.store(false, Ordering::Release);
Ok(snapshot.anchor)
}
pub async fn flush_runtime(
&self,
runtime: &NodeRuntime,
target_index: LogIndex,
) -> Result<CheckpointTip, DurabilityError> {
let result = self.flush_runtime_inner(runtime, target_index).await;
if result.is_err() {
self.mark_unavailable();
}
result
}
async fn flush_runtime_inner(
&self,
runtime: &NodeRuntime,
target_index: LogIndex,
) -> Result<CheckpointTip, DurabilityError> {
let log_state = runtime.log_store().logical_state()?;
let local_index = log_state.tip.as_ref().map_or(0, |tip| tip.index());
let mut durable_tip = self.durable_tip();
if durable_tip.index() > local_index {
return Err(DurabilityError::ArchiveAheadOfLocal {
durable_index: durable_tip.index(),
local_index,
});
}
let target_index = target_index.min(local_index);
if target_index <= durable_tip.index() {
return Ok(durable_tip);
}
if let Some(anchor) = log_state.anchor {
if durable_tip.index() < anchor.compacted().index() {
return Err(DurabilityError::SnapshotRequired {
anchor: Box::new(anchor),
});
}
}
let mut next =
durable_tip
.index()
.checked_add(1)
.ok_or_else(|| DurabilityError::LocalLogGap {
expected: durable_tip.index(),
actual: None,
})?;
while next <= target_index {
let end = next
.saturating_add(FLUSH_BATCH_ENTRIES - 1)
.min(target_index);
let entries = runtime
.log_store()
.read_range(IndexRange::new(next, end)?)?;
validate_local_batch(&entries, next, end, durable_tip)?;
self.publication_attempts.fetch_add(1, Ordering::Relaxed);
let published = self.publisher.publish_committed(&entries).await?;
durable_tip = *published.manifest().tip();
self.mark_durable(durable_tip);
if durable_tip.index() >= target_index {
break;
}
next =
durable_tip
.index()
.checked_add(1)
.ok_or_else(|| DurabilityError::LocalLogGap {
expected: durable_tip.index(),
actual: None,
})?;
}
Ok(durable_tip)
}
pub async fn checkpoint_compact(
&self,
runtime: &NodeRuntime,
) -> Result<RecoveryAnchor, DurabilityError> {
self.checkpoint_compact_inner(runtime, None).await
}
pub async fn checkpoint_compact_fenced(
&self,
runtime: &NodeRuntime,
expected_config_id: u64,
expected_recovery_generation: u64,
expected_root: LogAnchor,
) -> Result<RecoveryAnchor, DurabilityError> {
self.checkpoint_compact_inner(
runtime,
Some((
expected_config_id,
expected_recovery_generation,
expected_root,
)),
)
.await
}
async fn checkpoint_compact_inner(
&self,
runtime: &NodeRuntime,
fence: Option<(u64, u64, LogAnchor)>,
) -> Result<RecoveryAnchor, DurabilityError> {
let (target, snapshot, _fence) = {
let _commit = runtime.commit.lock().map_err(|_| {
DurabilityError::SnapshotVerification("commit mutex is poisoned".into())
})?;
runtime
.ensure_ready()
.map_err(|error| DurabilityError::SnapshotVerification(error.to_string()))?;
let configuration = runtime
.configuration_state()
.map_err(|error| DurabilityError::SnapshotVerification(error.to_string()))?;
if !configuration.is_active() && configuration.stop().is_none() {
return Err(DurabilityError::SnapshotVerification(
"runtime configuration is not compactable".into(),
));
}
if let Some((config_id, generation, root)) = fence {
let actual_config_id = configuration.config_id();
let actual_generation = runtime.config.recovery_generation();
let actual_root = runtime.log_root_unlocked().ok();
if actual_config_id != config_id
|| actual_generation != generation
|| actual_root != Some(root)
{
eprintln!(
"checkpoint fence mismatch: config {actual_config_id}/{config_id}, \
generation {actual_generation}/{generation}, root {actual_root:?}/{root:?}"
);
return Err(DurabilityError::PreconditionFailed);
}
}
if runtime
.checkpointing
.swap(true, std::sync::atomic::Ordering::AcqRel)
{
return Err(DurabilityError::SnapshotVerification(
"checkpoint transition is already in progress".into(),
));
}
let fence = CheckpointFence(&runtime.checkpointing);
let (target, target_hash) = runtime
.ensure_materialized_tip()
.map_err(|error| DurabilityError::SnapshotVerification(error.to_string()))?;
let snapshot =
create_runtime_checkpoint_snapshot(runtime, target, target_hash, &configuration)?;
(target, snapshot, fence)
};
self.flush_runtime(runtime, target).await?;
let anchor = snapshot.anchor.clone();
self.publisher
.publish_checkpoint_snapshot(anchor.clone(), &snapshot.archive_bytes)
.await?;
let restored = self.store.restore_checkpoint_state().await?;
let published = restored.snapshot().ok_or_else(|| {
DurabilityError::SnapshotVerification("published checkpoint has no snapshot".into())
})?;
if published.anchor() != &anchor || published.bytes() != snapshot.archive_bytes {
return Err(DurabilityError::SnapshotVerification(
"read-back anchor or snapshot bytes differ".into(),
));
}
{
let _commit = runtime.commit.lock().map_err(|_| {
DurabilityError::SnapshotVerification("commit mutex is poisoned".into())
})?;
runtime.log_store.compact_prefix(&anchor)?;
runtime
.compact_embedded_log_before(anchor.compacted().index())
.map_err(|error| DurabilityError::SnapshotVerification(error.to_string()))?;
}
self.mark_durable(CheckpointTip::new(
anchor.compacted().index(),
anchor.compacted().hash(),
));
Ok(anchor)
}
pub async fn run_background<F>(
self: Arc<Self>,
runtime: Arc<NodeRuntime>,
shutdown: F,
) -> Result<(), DurabilityError>
where
F: Future<Output = ()> + Send,
{
let cadence = match self.mode {
DurabilityMode::Sync => SYNC_COMPACTION_POLL_INTERVAL,
DurabilityMode::Bounded { max_lag } => {
let half = max_lag / 2;
half.min(Duration::from_secs(1))
}
DurabilityMode::Periodic { interval } => interval,
};
tokio::pin!(shutdown);
loop {
tokio::select! {
() = &mut shutdown => return Ok(()),
() = tokio::time::sleep(cadence) => {
if !matches!(self.mode, DurabilityMode::Sync) {
match self.flush_runtime(&runtime, LogIndex::MAX).await {
Ok(_) | Err(DurabilityError::Archive(_) | DurabilityError::Io(_)) => {}
Err(error) => return Err(error),
}
}
if self.publisher.compaction_recommended().await {
match self.checkpoint_compact(&runtime).await {
Ok(_) => {}
Err(DurabilityError::Archive(_) | DurabilityError::Io(_)) => {
let _ = self.publisher.reload().await;
}
Err(error) => return Err(error),
}
}
}
}
}
}
fn mark_durable(&self, durable_tip: CheckpointTip) {
let mut state = self.lock_state();
mark_durable_state(&mut state, durable_tip);
}
fn mark_unavailable(&self) {
let mut state = self.lock_state();
if state.committed_index > state.durable_tip.index() {
state.health = DurabilityHealth::Unavailable;
}
}
fn lock_state(&self) -> MutexGuard<'_, CoordinatorState> {
self.state
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
}
}
fn create_runtime_checkpoint_snapshot(
runtime: &NodeRuntime,
target: LogIndex,
target_hash: LogHash,
configuration: &ConfigurationState,
) -> Result<RuntimeCheckpointSnapshot, DurabilityError> {
#[cfg(not(any(feature = "graph", feature = "kv")))]
let _ = target_hash;
let materializer = runtime
.lock_materializer()
.map_err(|error| DurabilityError::SnapshotVerification(error.to_string()))?;
match &*materializer {
#[cfg(not(any(feature = "sql", feature = "graph", feature = "kv")))]
Materializer::Unavailable => unreachable!("no execution profiles are compiled in"),
#[cfg(feature = "sql")]
Materializer::Sql(state) => {
let snapshot = state
.create_recovery_snapshot(runtime.config().recovery_generation())
.map_err(|error| DurabilityError::SnapshotVerification(error.to_string()))?;
if snapshot.anchor().compacted().index() != target
|| snapshot.anchor().configuration_state() != configuration
{
return Err(DurabilityError::SnapshotVerification(
"SQLite snapshot does not match the compacted runtime state".into(),
));
}
Ok(RuntimeCheckpointSnapshot {
anchor: snapshot.anchor().clone(),
archive_bytes: snapshot.db_bytes().to_vec(),
})
}
#[cfg(feature = "graph")]
Materializer::Graph(state) => {
let snapshot = state
.create_snapshot(target)
.map_err(|error| DurabilityError::SnapshotVerification(error.to_string()))?;
validate_engine_snapshot_identity(
runtime,
configuration,
EngineSnapshotIdentity {
cluster_id: snapshot.cluster_id(),
epoch: snapshot.epoch(),
config_id: snapshot.config_id(),
applied_index: snapshot.applied_index(),
applied_hash: snapshot.applied_hash(),
},
target,
target_hash,
)?;
let archive_bytes = encode_graph_snapshot(&snapshot)
.map_err(|error| DurabilityError::SnapshotVerification(error.to_string()))?;
Ok(RuntimeCheckpointSnapshot {
anchor: engine_recovery_anchor(
runtime,
configuration,
target,
snapshot.applied_hash(),
snapshot.materializer_fingerprint(),
&archive_bytes,
)?,
archive_bytes,
})
}
#[cfg(feature = "kv")]
Materializer::Kv(state) => {
let snapshot = state
.create_snapshot(target)
.map_err(|error| DurabilityError::SnapshotVerification(error.to_string()))?;
validate_engine_snapshot_identity(
runtime,
configuration,
EngineSnapshotIdentity {
cluster_id: snapshot.cluster_id(),
epoch: snapshot.epoch(),
config_id: snapshot.config_id(),
applied_index: snapshot.applied_index(),
applied_hash: snapshot.applied_hash(),
},
target,
target_hash,
)?;
let archive_bytes = encode_kv_snapshot(&snapshot)
.map_err(|error| DurabilityError::SnapshotVerification(error.to_string()))?;
Ok(RuntimeCheckpointSnapshot {
anchor: engine_recovery_anchor(
runtime,
configuration,
target,
snapshot.applied_hash(),
snapshot.materializer_fingerprint(),
&archive_bytes,
)?,
archive_bytes,
})
}
}
}
#[cfg(any(feature = "graph", feature = "kv"))]
fn validate_engine_snapshot_identity(
runtime: &NodeRuntime,
configuration: &ConfigurationState,
snapshot: EngineSnapshotIdentity<'_>,
expected_index: LogIndex,
expected_hash: LogHash,
) -> Result<(), DurabilityError> {
let config = runtime.config();
if snapshot.cluster_id != config.cluster_id()
|| snapshot.epoch != config.epoch()
|| snapshot.config_id != configuration.config_id()
|| snapshot.applied_index == 0
|| snapshot.applied_index != expected_index
|| snapshot.applied_hash != expected_hash
{
return Err(DurabilityError::SnapshotVerification(
"engine snapshot identity does not match the compacted runtime state".into(),
));
}
Ok(())
}
#[cfg(any(feature = "graph", feature = "kv"))]
fn engine_recovery_anchor(
runtime: &NodeRuntime,
configuration: &ConfigurationState,
applied_index: LogIndex,
applied_hash: LogHash,
materializer_fingerprint: LogHash,
archive_bytes: &[u8],
) -> Result<RecoveryAnchor, DurabilityError> {
let size_bytes = u64::try_from(archive_bytes.len()).map_err(|_| {
DurabilityError::SnapshotVerification("snapshot envelope size exceeds u64".into())
})?;
Ok(RecoveryAnchor::new(
runtime.config().cluster_id(),
runtime.config().epoch(),
configuration.clone(),
runtime.config().recovery_generation(),
LogAnchor::new(applied_index, applied_hash),
SnapshotIdentity::new(
format!("snapshot-{applied_index:020}"),
LogHash::digest(&[archive_bytes]),
size_bytes,
materializer_fingerprint,
),
))
}
fn mark_durable_state(state: &mut CoordinatorState, durable_tip: CheckpointTip) {
if durable_tip.index() > state.durable_tip.index() {
state.durable_tip = durable_tip;
}
if state.committed_index <= state.durable_tip.index() {
state.pending_lag = None;
state.health = DurabilityHealth::Available;
}
}
fn observe_durable_tip(
state: &Mutex<CoordinatorState>,
observed: CheckpointTip,
) -> Result<CheckpointTip, DurabilityError> {
let mut state = state
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
let current = state.durable_tip;
if observed.index() < current.index() {
return Err(DurabilityError::SnapshotVerification(format!(
"checkpoint tip rolled back from index {} to {}",
current.index(),
observed.index()
)));
}
if observed.index() == current.index() && observed.hash() != current.hash() {
return Err(DurabilityError::SnapshotVerification(format!(
"checkpoint tip hash changed at index {}",
observed.index()
)));
}
mark_durable_state(&mut state, observed);
Ok(state.durable_tip)
}
struct CheckpointFence<'a>(&'a std::sync::atomic::AtomicBool);
impl Drop for CheckpointFence<'_> {
fn drop(&mut self) {
self.0.store(false, std::sync::atomic::Ordering::Release);
}
}
pub async fn restore_checkpoint_to_fresh_data_dir(
store: ObjectArchiveStore,
data_dir: impl AsRef<Path>,
) -> Result<CheckpointTip, DurabilityError> {
restore_checkpoint_to_fresh_data_dir_with_target(store, data_dir.as_ref(), None, None).await
}
pub async fn restore_checkpoint_to_fresh_data_dir_for_node(
store: ObjectArchiveStore,
data_dir: impl AsRef<Path>,
target_node_id: &str,
) -> Result<CheckpointTip, DurabilityError> {
if target_node_id.is_empty() {
return Err(DurabilityError::SnapshotVerification(
"target node_id is empty".into(),
));
}
restore_checkpoint_to_fresh_data_dir_with_target(
store,
data_dir.as_ref(),
Some(target_node_id),
None,
)
.await
}
pub async fn restore_checkpoint_to_fresh_data_dir_for_node_with_marker(
store: ObjectArchiveStore,
data_dir: impl AsRef<Path>,
target_node_id: &str,
marker_name: &str,
marker_contents: &[u8],
) -> Result<CheckpointTip, DurabilityError> {
if target_node_id.is_empty() {
return Err(DurabilityError::SnapshotVerification(
"target node_id is empty".into(),
));
}
validate_restore_marker_name(marker_name)?;
restore_checkpoint_to_fresh_data_dir_with_target(
store,
data_dir.as_ref(),
Some(target_node_id),
Some((marker_name, marker_contents)),
)
.await
}
fn restore_intent_identity(
identity: &CheckpointIdentity,
node_id: &str,
execution_profile: ExecutionProfile,
checkpoint_root: LogAnchor,
) -> RestoreIntentIdentity {
RestoreIntentIdentity {
cluster_id: identity.cluster_id().to_owned(),
node_id: node_id.to_owned(),
execution_profile,
epoch: identity.epoch(),
config_id: identity.config_id(),
recovery_generation: identity.recovery_generation(),
checkpoint_index: checkpoint_root.index(),
checkpoint_hash: checkpoint_root.hash().to_hex(),
}
}
fn encode_restore_intent(
identity: &CheckpointIdentity,
node_id: &str,
execution_profile: ExecutionProfile,
checkpoint_root: LogAnchor,
) -> Result<Vec<u8>, DurabilityError> {
serde_json::to_vec(&restore_intent_identity(
identity,
node_id,
execution_profile,
checkpoint_root,
))
.map_err(|error| DurabilityError::SnapshotVerification(error.to_string()))
}
fn parse_restore_intent_identity(bytes: &[u8]) -> Option<RestoreIntentIdentity> {
let intent = serde_json::from_slice::<RestoreIntentIdentity>(bytes).ok()?;
(!intent.cluster_id.is_empty()
&& !intent.node_id.is_empty()
&& LogHash::from_hex(&intent.checkpoint_hash).is_some())
.then_some(intent)
}
pub fn checkpoint_restore_in_progress(
data_dir: impl AsRef<Path>,
identity: &CheckpointIdentity,
node_id: &str,
execution_profile: ExecutionProfile,
checkpoint_root: LogAnchor,
) -> Result<CheckpointRestoreState, DurabilityError> {
let data_dir = data_dir.as_ref();
let intent = data_dir.join(RESTORE_INTENT_FILE);
let metadata = match fs::symlink_metadata(&intent) {
Ok(metadata) => Some(metadata),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => None,
Err(error) => return Err(error.into()),
};
let Some(metadata) = metadata else {
return Ok(CheckpointRestoreState::None);
};
if metadata.file_type().is_symlink() || !metadata.is_file() || metadata.len() > 4096 {
return Err(DurabilityError::SnapshotVerification(
"local checkpoint restore intent is invalid".into(),
));
}
let bytes = read_bounded_regular_file(&intent, 4096)?.ok_or_else(|| {
DurabilityError::SnapshotVerification("local checkpoint restore intent disappeared".into())
})?;
let actual: RestoreIntentIdentity = serde_json::from_slice(&bytes).map_err(|_| {
DurabilityError::SnapshotVerification("local checkpoint restore intent is invalid".into())
})?;
let expected = restore_intent_identity(identity, node_id, execution_profile, checkpoint_root);
if actual.cluster_id != expected.cluster_id
|| actual.node_id != expected.node_id
|| actual.execution_profile != expected.execution_profile
|| actual.epoch != expected.epoch
|| actual.config_id != expected.config_id
|| actual.recovery_generation != expected.recovery_generation
|| actual.checkpoint_index != expected.checkpoint_index
|| actual.checkpoint_hash != expected.checkpoint_hash
{
return Err(DurabilityError::SnapshotVerification(
"local checkpoint restore intent does not exactly match this node and checkpoint"
.into(),
));
}
Ok(CheckpointRestoreState::IdentityBound)
}
pub fn validate_local_recovery_view(
data_dir: impl AsRef<Path>,
identity: &CheckpointIdentity,
target_node_id: &str,
execution_profile: ExecutionProfile,
checkpoint_root: LogAnchor,
) -> Result<(), DurabilityError> {
let data_dir = data_dir.as_ref();
if checkpoint_restore_in_progress(
data_dir,
identity,
target_node_id,
execution_profile,
checkpoint_root,
)? != CheckpointRestoreState::None
{
return Err(DurabilityError::SnapshotVerification(
"local recovery view has an incomplete checkpoint restore intent".into(),
));
}
#[cfg(not(feature = "kv"))]
let _ = target_node_id;
if snapshot_profile(identity.cluster_id())? != execution_profile {
return Err(DurabilityError::SnapshotVerification(
"local recovery view profile does not match checkpoint identity".into(),
));
}
let recovery_identity = RecoveryArtifactIdentity::Restore(restore_intent_identity(
identity,
target_node_id,
execution_profile,
checkpoint_root,
));
cleanup_owned_recovery_artifacts(data_dir, &recovery_identity)?;
let validate_qlog = validate_local_materializer_identity(
data_dir,
identity,
target_node_id,
execution_profile,
)?;
if validate_qlog {
validate_local_qlog(data_dir, identity, checkpoint_root)?;
}
Ok(())
}
fn validate_local_materializer_identity(
data_dir: &Path,
identity: &CheckpointIdentity,
target_node_id: &str,
execution_profile: ExecutionProfile,
) -> Result<bool, DurabilityError> {
Ok(match execution_profile {
ExecutionProfile::Sqlite => {
#[cfg(feature = "sql")]
{
let path = data_dir.join("sqlite/db.sqlite");
if !fs::symlink_metadata(&path).is_ok_and(|metadata| metadata.is_file()) {
return Err(DurabilityError::SnapshotVerification(
"SQL materializer is missing or is not a regular file".into(),
));
}
let _state = rhiza_sql::SqliteStateMachine::open(
path,
identity.cluster_id(),
target_node_id,
identity.epoch(),
identity.config_id(),
)
.map_err(|error| DurabilityError::SnapshotVerification(error.to_string()))?;
true
}
#[cfg(not(feature = "sql"))]
return Err(DurabilityError::SnapshotVerification(
"sql execution profile is not compiled in".into(),
));
}
ExecutionProfile::Kv => {
#[cfg(feature = "kv")]
{
let path = data_dir.join("kv/data.redb");
if !fs::symlink_metadata(&path).is_ok_and(|metadata| metadata.is_file()) {
return Err(DurabilityError::SnapshotVerification(
"KV materializer is missing or is not a regular file".into(),
));
}
let _state = RedbStateMachine::open(
&path,
identity.cluster_id(),
target_node_id,
identity.epoch(),
identity.config_id(),
)
.map_err(|error| DurabilityError::SnapshotVerification(error.to_string()))?;
true
}
#[cfg(not(feature = "kv"))]
return Err(DurabilityError::SnapshotVerification(
"kv execution profile is not compiled in".into(),
));
}
ExecutionProfile::Graph => false,
})
}
pub async fn restore_checkpoint_for_rejoin_preserving_recorder(
store: ObjectArchiveStore,
data_dir: impl AsRef<Path>,
target_node_id: &str,
execution_profile: ExecutionProfile,
marker_name: &str,
marker_contents: &[u8],
) -> Result<CheckpointTip, DurabilityError> {
if target_node_id.is_empty() {
return Err(DurabilityError::SnapshotVerification(
"target node_id is empty".into(),
));
}
validate_restore_marker_name(marker_name)?;
let data_dir = data_dir.as_ref();
let identity = store.checkpoint_identity()?.clone();
if snapshot_profile(identity.cluster_id())? != execution_profile
|| execution_profile == ExecutionProfile::Graph
{
return Err(DurabilityError::SnapshotVerification(
"rejoin recovery only replaces matching SQL or KV recovery views".into(),
));
}
let restored = store.restore_checkpoint_state().await?;
let checkpoint_root = LogAnchor::new(restored.tip().index(), restored.tip().hash());
let intent = encode_restore_intent(
&identity,
target_node_id,
execution_profile,
checkpoint_root,
)?;
let recovery_identity = RecoveryArtifactIdentity::Restore(restore_intent_identity(
&identity,
target_node_id,
execution_profile,
checkpoint_root,
));
checkpoint_restore_in_progress(
data_dir,
&identity,
target_node_id,
execution_profile,
checkpoint_root,
)?;
cleanup_owned_recovery_artifacts(data_dir, &recovery_identity)?;
publish_restore_marker(data_dir, RESTORE_INTENT_FILE, &intent)?;
install_restored_checkpoint(
&identity,
&restored,
data_dir,
RestoreInstallOptions {
target_node_id: Some(target_node_id),
remove_generic_intent: true,
replace_rebuildable: true,
recovery_identity: Some(&recovery_identity),
completion_marker: Some((marker_name, marker_contents)),
},
)
}
async fn restore_checkpoint_to_fresh_data_dir_with_target(
store: ObjectArchiveStore,
data_dir: &Path,
target_node_id: Option<&str>,
completion_marker: Option<(&str, &[u8])>,
) -> Result<CheckpointTip, DurabilityError> {
let identity = store.checkpoint_identity()?.clone();
store
.load_checkpoint()
.await?
.ok_or(DurabilityError::MissingCheckpoint)?;
let restored = store.restore_checkpoint_state().await?;
let target_node_id = target_node_id.unwrap_or("<unbound-restore>");
let profile = snapshot_profile(identity.cluster_id())?;
let checkpoint_root = LogAnchor::new(restored.tip().index(), restored.tip().hash());
let intent = encode_restore_intent(&identity, target_node_id, profile, checkpoint_root)?;
let recovery_identity = RecoveryArtifactIdentity::Restore(restore_intent_identity(
&identity,
target_node_id,
profile,
checkpoint_root,
));
prepare_fresh_restore_data_dir(data_dir, completion_marker.map(|(name, _)| name), &intent)?;
publish_restore_marker(data_dir, RESTORE_INTENT_FILE, &intent)?;
install_restored_checkpoint(
&identity,
&restored,
data_dir,
RestoreInstallOptions {
target_node_id: Some(target_node_id),
remove_generic_intent: true,
replace_rebuildable: false,
recovery_identity: Some(&recovery_identity),
completion_marker,
},
)
}
struct RestoreInstallOptions<'a> {
target_node_id: Option<&'a str>,
remove_generic_intent: bool,
replace_rebuildable: bool,
recovery_identity: Option<&'a RecoveryArtifactIdentity>,
completion_marker: Option<(&'a str, &'a [u8])>,
}
struct RestoredCheckpointStaging {
path: PathBuf,
tip: CheckpointTip,
}
fn install_restored_checkpoint(
identity: &CheckpointIdentity,
restored: &RestoredCheckpoint,
data_dir: &Path,
options: RestoreInstallOptions<'_>,
) -> Result<CheckpointTip, DurabilityError> {
let staged = stage_restored_checkpoint(
identity,
restored,
data_dir,
options.target_node_id,
options.recovery_identity,
)?;
let tip = staged.tip;
let staging = staged.path;
let profile = snapshot_profile(identity.cluster_id())?;
let result = (|| -> Result<(), DurabilityError> {
if options.replace_rebuildable {
quarantine_rebuildable_view(data_dir, profile, options.recovery_identity)?;
}
publish_restore_staging(
&staging,
data_dir,
options.remove_generic_intent,
options.completion_marker,
)
})();
if result.is_err() {
let _ = fs::remove_dir_all(&staging);
}
result?;
Ok(tip)
}
fn stage_restored_checkpoint(
identity: &CheckpointIdentity,
restored: &RestoredCheckpoint,
staging_parent: &Path,
target_node_id: Option<&str>,
recovery_identity: Option<&RecoveryArtifactIdentity>,
) -> Result<RestoredCheckpointStaging, DurabilityError> {
let tip = *restored.tip();
let profile = snapshot_profile(identity.cluster_id())?;
validate_restored_suffix(profile, restored.suffix())?;
let staging = create_restore_staging_dir(staging_parent, recovery_identity)?;
let result = (|| -> Result<(), DurabilityError> {
if let Some(snapshot) = restored.snapshot() {
install_profile_snapshot(identity, snapshot, &staging, target_node_id)?;
}
if restored.snapshot().is_some() || !restored.suffix().is_empty() {
let initial_configuration = restored
.snapshot()
.map(|snapshot| snapshot.anchor().configuration_state().clone())
.unwrap_or_else(|| ConfigurationState::active(identity.config_id(), LogHash::ZERO));
let log = FileLogStore::open_with_configuration(
staging.join("consensus/log"),
identity.cluster_id(),
identity.epoch(),
initial_configuration,
)?;
if let Some(snapshot) = restored.snapshot() {
log.install_recovery_anchor(
snapshot.anchor(),
identity.recovery_generation(),
snapshot.anchor().configuration_state(),
)?;
}
for batch in restored.suffix().chunks(FLUSH_BATCH_ENTRIES as usize) {
log.append_batch(batch)?;
}
let installed_tip = log.logical_state()?.tip;
if installed_tip.as_ref().map(|tip| (tip.index(), tip.hash()))
!= Some((tip.index(), tip.hash()))
{
return Err(DurabilityError::SnapshotVerification(
"installed qlog tip does not match checkpoint tip".into(),
));
}
}
sync_directory(&staging)
})();
if result.is_err() {
let _ = fs::remove_dir_all(&staging);
}
result?;
Ok(RestoredCheckpointStaging { path: staging, tip })
}
fn validate_restored_suffix(
profile: ExecutionProfile,
suffix: &[LogEntry],
) -> Result<(), DurabilityError> {
for entry in suffix {
crate::validate_profile_entry_shape(profile, entry)
.map_err(DurabilityError::SnapshotVerification)?;
}
Ok(())
}
fn validate_local_qlog(
data_dir: &Path,
identity: &CheckpointIdentity,
checkpoint_root: LogAnchor,
) -> Result<LogAnchor, DurabilityError> {
validate_local_qlog_with_configuration(
data_dir,
identity,
checkpoint_root,
ConfigurationState::active(identity.config_id(), LogHash::ZERO),
)
}
fn validate_local_qlog_with_configuration(
data_dir: &Path,
identity: &CheckpointIdentity,
checkpoint_root: LogAnchor,
initial_configuration: ConfigurationState,
) -> Result<LogAnchor, DurabilityError> {
let path = data_dir.join("consensus/log");
if !path_has_state(&path)? {
if checkpoint_root == LogAnchor::new(0, LogHash::ZERO) {
return Ok(checkpoint_root);
}
return Err(DurabilityError::SnapshotVerification(
"local qlog is missing or empty".into(),
));
}
let log = FileLogStore::open_with_configuration(
path,
identity.cluster_id(),
identity.epoch(),
initial_configuration,
)?;
let state = log.logical_state()?;
let tip = state
.tip
.ok_or_else(|| DurabilityError::SnapshotVerification("local qlog has no tip".into()))?;
if tip.index() < checkpoint_root.index() {
return Err(DurabilityError::SnapshotVerification(format!(
"local qlog tip {}/{} is behind checkpoint root {}/{}",
tip.index(),
tip.hash().to_hex(),
checkpoint_root.index(),
checkpoint_root.hash().to_hex(),
)));
}
if checkpoint_root.index() == 0 {
if checkpoint_root.hash() != LogHash::ZERO {
return Err(DurabilityError::SnapshotVerification(
"checkpoint genesis hash is not zero".into(),
));
}
return Ok(tip);
}
let included_hash = match state.anchor.as_ref() {
Some(anchor) if anchor.compacted().index() == checkpoint_root.index() => {
Some(anchor.compacted().hash())
}
Some(anchor) if anchor.compacted().index() > checkpoint_root.index() => {
return Err(DurabilityError::SnapshotVerification(
"local qlog compacted past checkpoint root without exact inclusion evidence".into(),
));
}
_ => log.read(checkpoint_root.index())?.map(|entry| entry.hash),
};
if included_hash != Some(checkpoint_root.hash()) {
return Err(DurabilityError::SnapshotVerification(format!(
"local qlog does not include checkpoint root {} with its exact hash",
checkpoint_root.index(),
)));
}
Ok(tip)
}
fn install_profile_snapshot(
identity: &CheckpointIdentity,
snapshot: &rhiza_archive::RestoredCheckpointSnapshot,
staging: &Path,
target_node_id: Option<&str>,
) -> Result<(), DurabilityError> {
match snapshot_profile(identity.cluster_id())? {
ExecutionProfile::Sqlite => {
#[cfg(feature = "sql")]
{
validate_anchor_fingerprint(
snapshot.anchor(),
sql_executor_fingerprint().map_err(|error| {
DurabilityError::SnapshotVerification(error.to_string())
})?,
)?;
let path = staging.join("sqlite/db.sqlite");
let node_id = target_node_id.ok_or_else(|| {
DurabilityError::SnapshotVerification(
"SQLite QWAL snapshot restore requires a target node_id".into(),
)
})?;
restore_recovery_snapshot_file(path, snapshot.bytes(), snapshot.anchor(), node_id)
.map_err(|error| DurabilityError::SnapshotVerification(error.to_string()))
}
#[cfg(not(feature = "sql"))]
Err(DurabilityError::SnapshotVerification(
"sql execution profile is not compiled in".into(),
))
}
ExecutionProfile::Graph => {
#[cfg(feature = "graph")]
{
let decoded = decode_graph_snapshot(snapshot.bytes())
.map_err(|error| DurabilityError::SnapshotVerification(error.to_string()))?;
validate_decoded_snapshot_anchor(
snapshot.anchor(),
decoded.cluster_id(),
decoded.epoch(),
decoded.config_id(),
decoded.applied_index(),
decoded.applied_hash(),
decoded.materializer_fingerprint(),
)?;
let target_node_id = target_node_id.unwrap_or(decoded.created_by());
restore_graph_snapshot_file(
staging.join("ladybug/graph.lbug"),
&decoded,
target_node_id,
)
.map_err(|error| DurabilityError::SnapshotVerification(error.to_string()))
}
#[cfg(not(feature = "graph"))]
Err(DurabilityError::SnapshotVerification(
"graph recovery support is not compiled in".into(),
))
}
ExecutionProfile::Kv => {
#[cfg(feature = "kv")]
{
let decoded = decode_kv_snapshot(snapshot.bytes())
.map_err(|error| DurabilityError::SnapshotVerification(error.to_string()))?;
validate_decoded_snapshot_anchor(
snapshot.anchor(),
decoded.cluster_id(),
decoded.epoch(),
decoded.config_id(),
decoded.applied_index(),
decoded.applied_hash(),
decoded.materializer_fingerprint(),
)?;
let target_node_id = target_node_id.unwrap_or(decoded.created_by());
restore_kv_snapshot_file(staging.join("kv/data.redb"), &decoded, target_node_id)
.map_err(|error| DurabilityError::SnapshotVerification(error.to_string()))
}
#[cfg(not(feature = "kv"))]
Err(DurabilityError::SnapshotVerification(
"KV recovery support is not compiled in".into(),
))
}
}
}
fn snapshot_profile(cluster_id: &str) -> Result<ExecutionProfile, DurabilityError> {
if matches!(cluster_id.strip_prefix("rhiza:graph:"), Some(logical) if !logical.is_empty()) {
Ok(ExecutionProfile::Graph)
} else if matches!(cluster_id.strip_prefix("rhiza:kv:"), Some(logical) if !logical.is_empty()) {
Ok(ExecutionProfile::Kv)
} else if matches!(cluster_id.strip_prefix("rhiza:sql:"), Some(logical) if !logical.is_empty())
{
Ok(ExecutionProfile::Sqlite)
} else {
Err(DurabilityError::SnapshotVerification(
"snapshot checkpoint identity has no canonical execution profile prefix".into(),
))
}
}
fn validate_anchor_fingerprint(
anchor: &RecoveryAnchor,
expected: LogHash,
) -> Result<(), DurabilityError> {
if anchor.executor_fingerprint() != expected {
return Err(DurabilityError::SnapshotVerification(
"snapshot executor fingerprint does not match this binary".into(),
));
}
Ok(())
}
#[cfg(any(feature = "graph", feature = "kv"))]
fn validate_decoded_snapshot_anchor(
anchor: &RecoveryAnchor,
cluster_id: &str,
epoch: u64,
config_id: u64,
applied_index: LogIndex,
applied_hash: LogHash,
materializer_fingerprint: LogHash,
) -> Result<(), DurabilityError> {
validate_anchor_fingerprint(anchor, materializer_fingerprint)?;
if anchor.cluster_id() != cluster_id
|| anchor.epoch() != epoch
|| anchor.config_id() != config_id
|| anchor.compacted().index() != applied_index
|| anchor.compacted().hash() != applied_hash
{
return Err(DurabilityError::SnapshotVerification(
"decoded snapshot identity does not match its recovery anchor".into(),
));
}
Ok(())
}
pub async fn prestage_successor_checkpoint(
store: ObjectArchiveStore,
prestage_dir: impl AsRef<Path>,
predecessor_configuration: ConfigurationState,
target_node_id: &str,
execution_profile: ExecutionProfile,
target_config_id: u64,
target_membership_digest: LogHash,
) -> Result<SuccessorPrestage, DurabilityError> {
if target_node_id.is_empty() {
return Err(DurabilityError::SnapshotVerification(
"successor prestage target node_id is empty".into(),
));
}
let identity = store.checkpoint_identity()?.clone();
if !predecessor_configuration.is_active()
|| predecessor_configuration.config_id() != identity.config_id()
{
return Err(DurabilityError::SnapshotVerification(
"successor prestage predecessor configuration does not match the checkpoint".into(),
));
}
if snapshot_profile(identity.cluster_id())? != execution_profile {
return Err(DurabilityError::SnapshotVerification(
"successor prestage profile does not match checkpoint identity".into(),
));
}
if identity
.config_id()
.checked_add(1)
.filter(|next| *next == target_config_id)
.is_none()
{
return Err(DurabilityError::SnapshotVerification(
"successor prestage target config_id is not the next configuration".into(),
));
}
let loaded = store
.load_checkpoint()
.await?
.ok_or(DurabilityError::MissingCheckpoint)?;
if loaded.manifest().identity() != &identity {
return Err(DurabilityError::SnapshotVerification(
"successor prestage checkpoint identity changed while loading".into(),
));
}
let expected = SuccessorPrestageIdentity {
cluster_id: identity.cluster_id().to_owned(),
epoch: identity.epoch(),
predecessor_config_id: identity.config_id(),
predecessor_membership_digest: predecessor_configuration.digest().to_hex(),
predecessor_recovery_generation: identity.recovery_generation(),
node_id: target_node_id.to_owned(),
execution_profile,
target_config_id,
target_membership_digest: target_membership_digest.to_hex(),
seed_index: loaded.manifest().tip().index(),
seed_hash: loaded.manifest().tip().hash().to_hex(),
};
validate_successor_prestage_identity(&expected)?;
let prestage_dir = prestage_dir.as_ref();
let mut prestage =
prepare_successor_prestage_root(prestage_dir, Some(&expected), &predecessor_configuration)?;
match prestage.state {
SuccessorPrestageState::Ready
| SuccessorPrestageState::Published
| SuccessorPrestageState::Finalized => return Ok(prestage),
SuccessorPrestageState::Preparing => {}
}
cleanup_preparing_successor_prestage(prestage_dir, &expected)?;
let restored = store.restore_checkpoint_state().await?;
if restored.tip() != loaded.manifest().tip() {
return Err(DurabilityError::SnapshotVerification(
"successor prestage checkpoint changed during restore".into(),
));
}
let recovery_identity = RecoveryArtifactIdentity::Prestage(expected.clone());
let staged = stage_restored_checkpoint(
&identity,
&restored,
prestage_dir,
Some(target_node_id),
Some(&recovery_identity),
)?;
if let Err(error) = publish_restore_staging(&staged.path, prestage_dir, false, None) {
let _ = fs::remove_dir_all(&staged.path);
return Err(error);
}
fs::rename(
prestage_dir.join(SUCCESSOR_PRESTAGE_INTENT_FILE),
prestage_dir.join(SUCCESSOR_PRESTAGE_READY_FILE),
)?;
sync_directory(prestage_dir)?;
prestage.state = SuccessorPrestageState::Ready;
Ok(prestage)
}
pub fn inspect_successor_prestage(
prestage_dir: impl AsRef<Path>,
predecessor_configuration: ConfigurationState,
) -> Result<SuccessorPrestage, DurabilityError> {
prepare_successor_prestage_root(prestage_dir.as_ref(), None, &predecessor_configuration)
}
pub fn publish_successor_prestage(
mut prestage: SuccessorPrestage,
data_dir: impl AsRef<Path>,
) -> Result<SuccessorPrestage, DurabilityError> {
let data_dir = data_dir.as_ref();
match prestage.state {
SuccessorPrestageState::Published if prestage.path == data_dir => return Ok(prestage),
SuccessorPrestageState::Ready => {}
_ => return Err(DurabilityError::PreconditionFailed),
}
if prestage.path != data_dir {
match fs::symlink_metadata(data_dir) {
Ok(_) => return Err(DurabilityError::DataDirNotFresh(data_dir.to_path_buf())),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(error) => return Err(error.into()),
}
let source_parent = prestage.path.parent().ok_or_else(|| {
DurabilityError::SnapshotVerification(
"successor prestage path has no parent directory".into(),
)
})?;
let target_parent = data_dir.parent().ok_or_else(|| {
DurabilityError::SnapshotVerification(
"successor data path has no parent directory".into(),
)
})?;
fs::create_dir_all(target_parent)?;
#[cfg(unix)]
if fs::metadata(source_parent)?.dev() != fs::metadata(target_parent)?.dev() {
return Err(DurabilityError::SnapshotVerification(
"successor prestage and final data directory must share a filesystem".into(),
));
}
fs::rename(&prestage.path, data_dir)?;
sync_directory(source_parent)?;
if target_parent != source_parent {
sync_directory(target_parent)?;
}
prestage.path = data_dir.to_path_buf();
}
fs::rename(
prestage.path.join(SUCCESSOR_PRESTAGE_READY_FILE),
prestage.path.join(SUCCESSOR_PRESTAGE_PUBLISHED_FILE),
)?;
sync_directory(&prestage.path)?;
prestage.state = SuccessorPrestageState::Published;
Ok(prestage)
}
fn validate_successor_prestage_stop(
identity: &SuccessorPrestageIdentity,
stop: &StopInformation,
predecessor_membership: &Membership,
) -> Result<rhiza_core::SuccessorDescriptor, DurabilityError> {
if stop.entry.cluster_id != identity.cluster_id()
|| stop.entry.epoch != identity.epoch()
|| stop.entry.config_id != identity.predecessor_config_id()
|| stop.entry.recompute_hash() != stop.entry.hash
{
return Err(DurabilityError::SnapshotVerification(
"successor prestage requires the exact bound Stop".into(),
));
}
let change = ConfigChange::recognize_parts(stop.entry.entry_type, &stop.entry.payload)
.map_err(|_| {
DurabilityError::SnapshotVerification(
"successor prestage Stop entry is not a bound configuration change".into(),
)
})?;
let ConfigChange::BoundStop { successor } = change else {
return Err(DurabilityError::SnapshotVerification(
"successor prestage requires a bound Stop entry".into(),
));
};
if successor.cluster_id() != identity.cluster_id()
|| successor.predecessor_config_id() != identity.predecessor_config_id()
|| successor.predecessor_config_digest() != predecessor_membership.digest()
|| successor.config_id() != identity.target_config_id()
|| successor.digest() != identity.target_membership_digest()
{
return Err(DurabilityError::SnapshotVerification(
"successor prestage Stop binding conflicts with the prestage target".into(),
));
}
stop.proof
.validate_for_cluster(
identity.cluster_id(),
stop.entry.index,
identity.epoch(),
identity.predecessor_config_id(),
predecessor_membership,
)
.map_err(|error| {
DurabilityError::SnapshotVerification(format!(
"successor prestage Stop proof is not quorum-certified: {error:?}"
))
})?;
let command = rhiza_core::StoredCommand::new(stop.entry.entry_type, stop.entry.payload.clone());
let expected_value = rhiza_quepaxa::AcceptedValue::from_command(
identity.cluster_id(),
stop.entry.index,
identity.epoch(),
identity.predecessor_config_id(),
stop.entry.prev_hash,
&command,
);
if stop.proof.proposal().value.as_ref() != Some(&expected_value) {
return Err(DurabilityError::SnapshotVerification(
"successor prestage Stop proof value does not match the exact Stop entry".into(),
));
}
Ok(successor)
}
pub fn finalize_successor_prestage_for_stop(
mut prestage: SuccessorPrestage,
stop: &StopInformation,
predecessor_membership: &Membership,
) -> Result<SuccessorPrestage, DurabilityError> {
if !matches!(
prestage.state,
SuccessorPrestageState::Published | SuccessorPrestageState::Finalized
) {
return Err(DurabilityError::PreconditionFailed);
}
let successor =
validate_successor_prestage_stop(&prestage.identity, stop, predecessor_membership)?;
let expected_stop = LogAnchor::new(stop.entry.index, stop.entry.hash);
let local_tip = validate_local_qlog_with_configuration(
&prestage.path,
&prestage.identity.checkpoint_identity(),
prestage.identity.seed_anchor(),
ConfigurationState::active(
prestage.identity.predecessor_config_id(),
successor.predecessor_config_digest(),
),
)?;
if local_tip != expected_stop {
return Err(DurabilityError::SnapshotVerification(
"successor prestage final qlog tip does not exactly match the bound Stop".into(),
));
}
if prestage.state == SuccessorPrestageState::Published {
fs::rename(
prestage.path.join(SUCCESSOR_PRESTAGE_PUBLISHED_FILE),
prestage.path.join(SUCCESSOR_PRESTAGE_FINALIZED_FILE),
)?;
sync_directory(&prestage.path)?;
prestage.state = SuccessorPrestageState::Finalized;
}
Ok(prestage)
}
pub fn adopt_finalized_successor_prestage(
prestage: SuccessorPrestage,
config: &NodeConfig,
stop: &StopInformation,
predecessor_membership: &Membership,
) -> Result<SuccessorRestorePreparation, DurabilityError> {
if prestage.state != SuccessorPrestageState::Finalized
|| prestage.path.as_path() != config.data_dir().as_path()
|| prestage.identity.cluster_id() != config.cluster_id()
|| prestage.identity.epoch() != config.epoch()
|| prestage.identity.predecessor_recovery_generation() != config.recovery_generation()
|| prestage.identity.node_id() != config.node_id()
|| prestage.identity.execution_profile() != config.execution_profile()
|| prestage.identity.target_membership_digest() != config.membership().digest()
|| stop.entry.cluster_id != config.cluster_id()
|| stop.entry.epoch != config.epoch()
|| stop.entry.config_id != prestage.identity.predecessor_config_id()
|| stop.entry.recompute_hash() != stop.entry.hash
|| config.predecessor_stop_entry.as_ref() != Some(&stop.entry)
{
return Err(DurabilityError::SnapshotVerification(
"finalized successor prestage does not match the target configuration and Stop".into(),
));
}
let successor =
validate_successor_prestage_stop(&prestage.identity, stop, predecessor_membership)?;
let predecessor_digest = successor.predecessor_config_digest();
let expected_stop = LogAnchor::new(stop.entry.index, stop.entry.hash);
if successor.cluster_id() != config.cluster_id()
|| successor.predecessor_config_id() != prestage.identity.predecessor_config_id()
|| successor.config_id() != prestage.identity.target_config_id()
|| successor.digest() != prestage.identity.target_membership_digest()
|| successor.members() != config.membership().members()
|| config.log_initial_configuration()
!= &ConfigurationState::active(
prestage.identity.predecessor_config_id(),
predecessor_digest,
)
|| config.configuration_state()
!= &ConfigurationState::stopped(
prestage.identity.predecessor_config_id(),
predecessor_digest,
expected_stop,
StopBinding::Bound {
successor: successor.clone(),
stop_command_hash: rhiza_core::StoredCommand::new(
stop.entry.entry_type,
stop.entry.payload.clone(),
)
.hash(),
},
)
{
return Err(DurabilityError::SnapshotVerification(
"finalized successor prestage Stop binding conflicts with the target configuration"
.into(),
));
}
let local_tip = validate_local_qlog_with_configuration(
&prestage.path,
&prestage.identity.checkpoint_identity(),
prestage.identity.seed_anchor(),
ConfigurationState::active(
prestage.identity.predecessor_config_id(),
predecessor_digest,
),
)?;
if local_tip != expected_stop {
return Err(DurabilityError::SnapshotVerification(
"finalized successor prestage qlog does not end at the exact Stop".into(),
));
}
let receipt = serde_json::to_vec(&SuccessorRestoreIdentity {
cluster_id: config.cluster_id(),
epoch: config.epoch(),
target_config_id: prestage.identity.target_config_id(),
recovery_generation: config.recovery_generation(),
node_id: config.node_id(),
membership_digest: config.membership().digest().to_hex(),
predecessor_config_id: prestage.identity.predecessor_config_id(),
stop_index: stop.entry.index,
stop_hash: stop.entry.hash.to_hex(),
})
.map_err(|error| DurabilityError::SnapshotVerification(error.to_string()))?;
let successor_lock = fs::OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(prestage.path.join(SUCCESSOR_RESTORE_LOCK_FILE))?;
successor_lock
.try_lock()
.map_err(|_| DurabilityError::PreconditionFailed)?;
let intent = prestage.path.join(SUCCESSOR_RESTORE_INTENT_FILE);
let complete = prestage.path.join(SUCCESSOR_RESTORE_COMPLETE_FILE);
match (
read_regular_successor_control_file(&intent)?,
read_regular_successor_control_file(&complete)?,
) {
(None, None) => {
publish_restore_marker(&prestage.path, SUCCESSOR_RESTORE_INTENT_FILE, &receipt)?
}
(Some(actual), None) if actual == receipt => {}
_ => {
return Err(DurabilityError::DataDirNotFresh(
prestage.path.to_path_buf(),
))
}
}
fs::remove_file(prestage.path.join(SUCCESSOR_PRESTAGE_FINALIZED_FILE))?;
sync_directory(&prestage.path)?;
Ok(SuccessorRestorePreparation {
tip: CheckpointTip::new(stop.entry.index, stop.entry.hash),
data_dir: prestage.path.clone(),
identity: receipt,
requires_recorder_install: true,
_lock: successor_lock,
})
}
fn write_repair_artifact_ownership(
artifact: &Path,
role: RepairArtifactRole,
identity: &RecoveryArtifactIdentity,
) -> Result<(), DurabilityError> {
let name = artifact
.file_name()
.and_then(|name| name.to_str())
.ok_or_else(|| {
DurabilityError::SnapshotVerification(
"repair artifact path must have a UTF-8 final component".into(),
)
})?
.to_owned();
let contents = serde_json::to_vec(&RepairArtifactOwnership {
role,
name,
identity: identity.clone(),
})
.map_err(|error| DurabilityError::SnapshotVerification(error.to_string()))?;
let owner = artifact.join(REPAIR_ARTIFACT_OWNER_FILE);
let mut file = fs::OpenOptions::new()
.write(true)
.create_new(true)
.open(owner)?;
file.write_all(&contents)?;
file.sync_all()?;
sync_directory(artifact)
}
fn create_restore_staging_dir(
data_dir: &Path,
recovery_identity: Option<&RecoveryArtifactIdentity>,
) -> Result<PathBuf, DurabilityError> {
fs::create_dir_all(data_dir)?;
for _ in 0..128 {
let sequence = RESTORE_SEQUENCE.fetch_add(1, Ordering::Relaxed);
let staging = data_dir.join(format!(
"{RESTORE_STAGING_PREFIX}{}-{sequence}",
process::id()
));
match fs::create_dir(&staging) {
Ok(()) => {
if let Some(identity) = recovery_identity {
if let Err(error) = write_repair_artifact_ownership(
&staging,
RepairArtifactRole::Staging,
identity,
) {
let _ = fs::remove_dir_all(&staging);
return Err(error);
}
}
return Ok(staging);
}
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => continue,
Err(error) => return Err(error.into()),
}
}
Err(std::io::Error::new(
std::io::ErrorKind::AlreadyExists,
"could not allocate restore staging directory",
)
.into())
}
fn publish_restore_staging(
staging: &Path,
data_dir: &Path,
remove_generic_intent: bool,
completion_marker: Option<(&str, &[u8])>,
) -> Result<(), DurabilityError> {
sync_directory(staging)?;
for name in ["sqlite", "ladybug", "kv", "consensus"] {
let source = staging.join(name);
if source.exists() {
fs::rename(&source, data_dir.join(name))?;
}
}
fs::remove_dir_all(staging)?;
sync_directory(data_dir)?;
if let Some((marker_name, marker_contents)) = completion_marker {
publish_restore_marker(data_dir, marker_name, marker_contents)?;
}
if remove_generic_intent {
fs::remove_file(data_dir.join(RESTORE_INTENT_FILE))?;
}
sync_directory(data_dir)
}
fn quarantine_rebuildable_view(
data_dir: &Path,
profile: ExecutionProfile,
recovery_identity: Option<&RecoveryArtifactIdentity>,
) -> Result<Option<PathBuf>, DurabilityError> {
let materializer = match profile {
ExecutionProfile::Sqlite => "sqlite",
ExecutionProfile::Kv => "kv",
ExecutionProfile::Graph => {
return Err(DurabilityError::SnapshotVerification(
"graph recovery view replacement is outside this recovery path".into(),
))
}
};
let names = [materializer, "consensus"];
let mut has_rebuildable_view = false;
for name in names {
match fs::symlink_metadata(data_dir.join(name)) {
Ok(metadata) if metadata.file_type().is_symlink() || !metadata.is_dir() => {
return Err(DurabilityError::SnapshotVerification(
"rebuildable recovery view is not a regular directory".into(),
));
}
Ok(_) => has_rebuildable_view = true,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(error) => return Err(error.into()),
}
}
if !has_rebuildable_view {
return Ok(None);
}
for _ in 0..128 {
let sequence = RESTORE_SEQUENCE.fetch_add(1, Ordering::Relaxed);
let quarantine = data_dir.join(format!(
".rebuildable-quarantine-{}-{sequence}",
process::id()
));
match fs::create_dir(&quarantine) {
Ok(()) => {
if let Some(identity) = recovery_identity {
if let Err(error) = write_repair_artifact_ownership(
&quarantine,
RepairArtifactRole::Quarantine,
identity,
) {
let _ = fs::remove_dir_all(&quarantine);
return Err(error);
}
}
for name in names {
let source = data_dir.join(name);
match fs::symlink_metadata(&source) {
Ok(_) => fs::rename(source, quarantine.join(name))?,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(error) => return Err(error.into()),
}
}
sync_directory(&quarantine)?;
sync_directory(data_dir)?;
return Ok(Some(quarantine));
}
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => continue,
Err(error) => return Err(error.into()),
}
}
Err(std::io::Error::new(
std::io::ErrorKind::AlreadyExists,
"could not allocate rebuildable recovery quarantine",
)
.into())
}
fn publish_restore_marker(
data_dir: &Path,
marker_name: &str,
contents: &[u8],
) -> Result<(), DurabilityError> {
validate_restore_marker_name(marker_name)?;
fs::create_dir_all(data_dir)?;
let sequence = RESTORE_SEQUENCE.fetch_add(1, Ordering::Relaxed);
let temporary = data_dir.join(format!(
"{RESTORE_MARKER_TMP_PREFIX}{}-{sequence}",
process::id()
));
let mut file = fs::OpenOptions::new()
.write(true)
.create_new(true)
.open(&temporary)?;
file.write_all(contents)?;
file.sync_all()?;
fs::rename(temporary, data_dir.join(marker_name))?;
sync_directory(data_dir)
}
fn validate_restore_marker_name(marker_name: &str) -> Result<(), DurabilityError> {
if marker_name.is_empty()
|| matches!(marker_name, "." | "..")
|| marker_name.contains(std::path::MAIN_SEPARATOR)
{
return Err(DurabilityError::SnapshotVerification(
"restore marker name must be one local file name".into(),
));
}
Ok(())
}
fn is_owned_generated_recovery_name(name: &str, prefix: &str) -> bool {
let Some(suffix) = name.strip_prefix(prefix) else {
return false;
};
let mut parts = suffix.split('-');
let (Some(process_id), Some(sequence), None) = (parts.next(), parts.next(), parts.next())
else {
return false;
};
process_id.parse::<u32>().is_ok_and(|id| id > 0) && sequence.parse::<u64>().is_ok()
}
fn is_safe_restore_marker_tmp(path: &Path, name: &str) -> Result<bool, DurabilityError> {
if !is_owned_generated_recovery_name(name, RESTORE_MARKER_TMP_PREFIX) {
return Ok(false);
}
Ok(read_bounded_regular_file(path, 16384)?.is_some())
}
fn is_owned_recovery_directory(
path: &Path,
allowed_children: &[&str],
expected_role: RepairArtifactRole,
expected_identity: &RecoveryArtifactIdentity,
) -> Result<bool, DurabilityError> {
let metadata = fs::symlink_metadata(path)?;
if metadata.file_type().is_symlink() || !metadata.is_dir() {
return Ok(false);
}
let owner = path.join(REPAIR_ARTIFACT_OWNER_FILE);
let Some(owner_bytes) = read_bounded_regular_file(&owner, 16384)? else {
return Ok(false);
};
let Ok(ownership) = serde_json::from_slice::<RepairArtifactOwnership>(&owner_bytes) else {
return Ok(false);
};
if ownership.role != expected_role
|| ownership.identity != *expected_identity
|| path.file_name().and_then(|name| name.to_str()) != Some(ownership.name.as_str())
{
return Ok(false);
}
for entry in fs::read_dir(path)? {
let entry = entry?;
let name = entry.file_name();
let name = name.to_string_lossy();
if name == REPAIR_ARTIFACT_OWNER_FILE {
continue;
}
let metadata = fs::symlink_metadata(entry.path())?;
if !allowed_children.contains(&name.as_ref())
|| metadata.file_type().is_symlink()
|| !metadata.is_dir()
{
return Ok(false);
}
}
Ok(true)
}
fn cleanup_owned_recovery_artifacts(
data_dir: &Path,
identity: &RecoveryArtifactIdentity,
) -> Result<(), DurabilityError> {
let mut cleanup = Vec::new();
for entry in fs::read_dir(data_dir)? {
let entry = entry?;
let name = entry.file_name();
let name = name.to_string_lossy();
let has_staging_prefix = name.starts_with(RESTORE_STAGING_PREFIX);
let has_quarantine_prefix = name.starts_with(".rebuildable-quarantine-");
if (has_staging_prefix && !is_owned_generated_recovery_name(&name, RESTORE_STAGING_PREFIX))
|| (has_quarantine_prefix
&& !is_owned_generated_recovery_name(&name, ".rebuildable-quarantine-"))
{
return Err(DurabilityError::DataDirNotFresh(data_dir.to_path_buf()));
}
let artifact = if has_staging_prefix {
Some((
["sqlite", "ladybug", "kv", "consensus"].as_slice(),
RepairArtifactRole::Staging,
))
} else if has_quarantine_prefix {
Some((
["sqlite", "ladybug", "kv", "consensus"].as_slice(),
RepairArtifactRole::Quarantine,
))
} else {
None
};
let Some((allowed_children, role)) = artifact else {
continue;
};
if !is_owned_recovery_directory(&entry.path(), allowed_children, role, identity)? {
return Err(DurabilityError::DataDirNotFresh(data_dir.to_path_buf()));
}
cleanup.push(entry.path());
}
for path in cleanup.iter() {
fs::remove_dir_all(path)?;
}
if !cleanup.is_empty() {
sync_directory(data_dir)?;
}
Ok(())
}
#[cfg(any(target_os = "linux", target_os = "android"))]
const O_NOFOLLOW_FLAG: i32 = 0o400000;
#[cfg(any(
target_os = "macos",
target_os = "ios",
target_os = "freebsd",
target_os = "openbsd",
target_os = "netbsd",
target_os = "dragonfly"
))]
const O_NOFOLLOW_FLAG: i32 = 0x0100;
fn open_recovery_file_no_follow(path: &Path) -> Result<fs::File, DurabilityError> {
let mut options = fs::OpenOptions::new();
options.read(true);
#[cfg(any(
target_os = "linux",
target_os = "android",
target_os = "macos",
target_os = "ios",
target_os = "freebsd",
target_os = "openbsd",
target_os = "netbsd",
target_os = "dragonfly"
))]
options.custom_flags(O_NOFOLLOW_FLAG);
Ok(options.open(path)?)
}
fn read_bounded_regular_file(
path: &Path,
max_bytes: u64,
) -> Result<Option<Vec<u8>>, DurabilityError> {
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 metadata.file_type().is_symlink() || !metadata.is_file() || metadata.len() > max_bytes {
return Err(DurabilityError::SnapshotVerification(
"recovery file is not a bounded regular file".into(),
));
}
let mut file = open_recovery_file_no_follow(path)?;
let opened = file.metadata()?;
if !opened.is_file() || opened.len() > max_bytes {
return Err(DurabilityError::SnapshotVerification(
"recovery file changed to an invalid form before read".into(),
));
}
#[cfg(unix)]
if metadata.dev() != opened.dev()
|| metadata.ino() != opened.ino()
|| metadata.len() != opened.len()
{
return Err(DurabilityError::SnapshotVerification(
"recovery file changed before no-follow open".into(),
));
}
let mut contents = Vec::with_capacity(usize::try_from(opened.len()).unwrap_or(0));
Read::by_ref(&mut file)
.take(max_bytes + 1)
.read_to_end(&mut contents)?;
if contents.len() as u64 > max_bytes || contents.len() as u64 != opened.len() {
return Err(DurabilityError::SnapshotVerification(
"recovery file changed during bounded read".into(),
));
}
Ok(Some(contents))
}
fn read_regular_successor_control_file(path: &Path) -> Result<Option<Vec<u8>>, DurabilityError> {
read_bounded_regular_file(path, 16384)
}
fn parse_successor_restore_receipt(bytes: &[u8]) -> Option<SuccessorRestoreReceipt> {
let receipt = serde_json::from_slice::<SuccessorRestoreReceipt>(bytes).ok()?;
(!receipt.cluster_id.is_empty()
&& !receipt.node_id.is_empty()
&& LogHash::from_hex(&receipt.membership_digest).is_some()
&& LogHash::from_hex(&receipt.stop_hash).is_some())
.then_some(receipt)
}
fn successor_receipt_matches_finalized_prestage(
receipt: &SuccessorRestoreReceipt,
identity: &SuccessorPrestageIdentity,
) -> bool {
receipt.cluster_id == identity.cluster_id
&& receipt.epoch == identity.epoch
&& receipt.target_config_id == identity.target_config_id
&& receipt.recovery_generation == identity.predecessor_recovery_generation
&& receipt.node_id == identity.node_id
&& receipt.membership_digest == identity.target_membership_digest
&& receipt.predecessor_config_id == identity.predecessor_config_id
}
fn parse_successor_prestage_identity(bytes: &[u8]) -> Option<SuccessorPrestageIdentity> {
let identity = serde_json::from_slice::<SuccessorPrestageIdentity>(bytes).ok()?;
validate_successor_prestage_identity(&identity)
.is_ok()
.then_some(identity)
}
fn validate_successor_prestage_identity(
identity: &SuccessorPrestageIdentity,
) -> Result<(), DurabilityError> {
let valid = !identity.cluster_id.is_empty()
&& !identity.node_id.is_empty()
&& snapshot_profile(&identity.cluster_id)? == identity.execution_profile
&& identity
.predecessor_config_id
.checked_add(1)
.is_some_and(|next| next == identity.target_config_id)
&& LogHash::from_hex(&identity.predecessor_membership_digest).is_some()
&& LogHash::from_hex(&identity.target_membership_digest).is_some()
&& LogHash::from_hex(&identity.seed_hash).is_some();
if !valid {
return Err(DurabilityError::SnapshotVerification(
"successor prestage identity is invalid".into(),
));
}
Ok(())
}
fn prepare_successor_prestage_root(
prestage_dir: &Path,
expected_identity: Option<&SuccessorPrestageIdentity>,
predecessor_configuration: &ConfigurationState,
) -> Result<SuccessorPrestage, DurabilityError> {
if expected_identity.is_some() {
fs::create_dir_all(prestage_dir)?;
}
let metadata = fs::symlink_metadata(prestage_dir)
.map_err(|_| DurabilityError::DataDirNotFresh(prestage_dir.to_path_buf()))?;
if metadata.file_type().is_symlink() || !metadata.is_dir() {
return Err(DurabilityError::DataDirNotFresh(prestage_dir.to_path_buf()));
}
let lock_path = prestage_dir.join(SUCCESSOR_PRESTAGE_LOCK_FILE);
let mut lock_options = fs::OpenOptions::new();
lock_options.read(true).write(true);
if expected_identity.is_some() {
lock_options.create(true).truncate(false);
}
let lock = lock_options
.open(&lock_path)
.map_err(|_| DurabilityError::DataDirNotFresh(prestage_dir.to_path_buf()))?;
lock.try_lock()
.map_err(|_| DurabilityError::PreconditionFailed)?;
let marker_files = [
(
SUCCESSOR_PRESTAGE_INTENT_FILE,
SuccessorPrestageState::Preparing,
),
(SUCCESSOR_PRESTAGE_READY_FILE, SuccessorPrestageState::Ready),
(
SUCCESSOR_PRESTAGE_PUBLISHED_FILE,
SuccessorPrestageState::Published,
),
(
SUCCESSOR_PRESTAGE_FINALIZED_FILE,
SuccessorPrestageState::Finalized,
),
];
let mut marker = None;
for (name, state) in marker_files {
let Some(bytes) = read_bounded_regular_file(&prestage_dir.join(name), 16384)
.map_err(|_| DurabilityError::DataDirNotFresh(prestage_dir.to_path_buf()))?
else {
continue;
};
if marker.is_some() {
return Err(DurabilityError::DataDirNotFresh(prestage_dir.to_path_buf()));
}
let identity = parse_successor_prestage_identity(&bytes)
.ok_or_else(|| DurabilityError::DataDirNotFresh(prestage_dir.to_path_buf()))?;
marker = Some((name, state, identity));
}
let (marker_name, state, identity) = match marker {
Some(marker) => marker,
None => {
let expected = expected_identity
.ok_or_else(|| DurabilityError::DataDirNotFresh(prestage_dir.to_path_buf()))?;
for entry in fs::read_dir(prestage_dir)? {
let entry = entry?;
let name = entry.file_name();
let name = name.to_string_lossy();
if name == SUCCESSOR_PRESTAGE_LOCK_FILE {
continue;
}
if is_safe_restore_marker_tmp(&entry.path(), &name)? {
fs::remove_file(entry.path())?;
continue;
}
return Err(DurabilityError::DataDirNotFresh(prestage_dir.to_path_buf()));
}
let bytes = serde_json::to_vec(expected)
.map_err(|error| DurabilityError::SnapshotVerification(error.to_string()))?;
publish_restore_marker(prestage_dir, SUCCESSOR_PRESTAGE_INTENT_FILE, &bytes)?;
(
SUCCESSOR_PRESTAGE_INTENT_FILE,
SuccessorPrestageState::Preparing,
expected.clone(),
)
}
};
if expected_identity.is_some_and(|expected| expected != &identity) {
return Err(DurabilityError::DataDirNotFresh(prestage_dir.to_path_buf()));
}
if !predecessor_configuration.is_active()
|| predecessor_configuration.config_id() != identity.predecessor_config_id()
|| predecessor_configuration.digest() != identity.predecessor_membership_digest()
{
return Err(DurabilityError::SnapshotVerification(
"successor prestage predecessor configuration does not match its identity".into(),
));
}
let recovery_identity = RecoveryArtifactIdentity::Prestage(identity.clone());
for entry in fs::read_dir(prestage_dir)? {
let entry = entry?;
let name = entry.file_name();
let name = name.to_string_lossy();
if name == SUCCESSOR_PRESTAGE_LOCK_FILE || name == marker_name {
continue;
}
if state == SuccessorPrestageState::Finalized && name == SUCCESSOR_RESTORE_INTENT_FILE {
let bytes = read_regular_successor_control_file(&entry.path())?
.ok_or_else(|| DurabilityError::DataDirNotFresh(prestage_dir.to_path_buf()))?;
let receipt = parse_successor_restore_receipt(&bytes)
.ok_or_else(|| DurabilityError::DataDirNotFresh(prestage_dir.to_path_buf()))?;
if successor_receipt_matches_finalized_prestage(&receipt, &identity) {
continue;
}
return Err(DurabilityError::DataDirNotFresh(prestage_dir.to_path_buf()));
}
if state == SuccessorPrestageState::Finalized
&& name == SUCCESSOR_RESTORE_LOCK_FILE
&& fs::symlink_metadata(entry.path())
.is_ok_and(|metadata| metadata.is_file() && !metadata.file_type().is_symlink())
{
continue;
}
if is_safe_restore_marker_tmp(&entry.path(), &name)? {
if state == SuccessorPrestageState::Preparing {
fs::remove_file(entry.path())?;
continue;
}
return Err(DurabilityError::DataDirNotFresh(prestage_dir.to_path_buf()));
}
if ["sqlite", "ladybug", "kv", "consensus"].contains(&name.as_ref()) {
let metadata = fs::symlink_metadata(entry.path())?;
if metadata.file_type().is_symlink() || !metadata.is_dir() {
return Err(DurabilityError::DataDirNotFresh(prestage_dir.to_path_buf()));
}
continue;
}
if name.starts_with(RESTORE_STAGING_PREFIX)
&& state == SuccessorPrestageState::Preparing
&& is_owned_generated_recovery_name(&name, RESTORE_STAGING_PREFIX)
&& is_owned_recovery_directory(
&entry.path(),
&["sqlite", "ladybug", "kv", "consensus"],
RepairArtifactRole::Staging,
&recovery_identity,
)?
{
continue;
}
return Err(DurabilityError::DataDirNotFresh(prestage_dir.to_path_buf()));
}
if !matches!(
state,
SuccessorPrestageState::Preparing | SuccessorPrestageState::Finalized
) {
validate_local_qlog_with_configuration(
prestage_dir,
&identity.checkpoint_identity(),
identity.seed_anchor(),
predecessor_configuration.clone(),
)?;
}
Ok(SuccessorPrestage {
path: prestage_dir.to_path_buf(),
identity,
state,
_lock: lock,
})
}
fn cleanup_preparing_successor_prestage(
prestage_dir: &Path,
identity: &SuccessorPrestageIdentity,
) -> Result<(), DurabilityError> {
let recovery_identity = RecoveryArtifactIdentity::Prestage(identity.clone());
for entry in fs::read_dir(prestage_dir)? {
let entry = entry?;
let name = entry.file_name();
let name = name.to_string_lossy();
let owned_component = ["sqlite", "ladybug", "kv", "consensus"].contains(&name.as_ref());
let owned_staging = name.starts_with(RESTORE_STAGING_PREFIX)
&& is_owned_generated_recovery_name(&name, RESTORE_STAGING_PREFIX)
&& is_owned_recovery_directory(
&entry.path(),
&["sqlite", "ladybug", "kv", "consensus"],
RepairArtifactRole::Staging,
&recovery_identity,
)?;
if owned_component || owned_staging {
fs::remove_dir_all(entry.path())?;
}
}
sync_directory(prestage_dir)
}
fn sync_directory(path: &Path) -> Result<(), DurabilityError> {
fs::File::open(path)?.sync_all()?;
Ok(())
}
fn validate_local_batch(
entries: &[rhiza_core::LogEntry],
start: LogIndex,
end: LogIndex,
durable_tip: CheckpointTip,
) -> Result<(), DurabilityError> {
let expected_len =
usize::try_from(end - start + 1).map_err(|_| DurabilityError::LocalLogGap {
expected: start,
actual: entries.first().map(|entry| entry.index),
})?;
if entries.len() != expected_len {
let actual = entries
.iter()
.zip(start..=end)
.find_map(|(entry, expected)| (entry.index != expected).then_some(entry.index));
return Err(DurabilityError::LocalLogGap {
expected: start + entries.len() as u64,
actual,
});
}
let mut expected_hash = durable_tip.hash();
for (expected_index, entry) in (start..).zip(entries) {
if entry.index != expected_index {
return Err(DurabilityError::LocalLogGap {
expected: expected_index,
actual: Some(entry.index),
});
}
if entry.prev_hash != expected_hash || entry.recompute_hash() != entry.hash {
return Err(DurabilityError::LocalLogConflict { index: entry.index });
}
expected_hash = entry.hash;
}
Ok(())
}
fn prepare_fresh_restore_data_dir(
data_dir: &Path,
completion_marker_name: Option<&str>,
expected_intent: &[u8],
) -> Result<(), DurabilityError> {
if !path_has_state(data_dir)? {
return Ok(());
}
let intent = data_dir.join(RESTORE_INTENT_FILE);
let intent_metadata = match fs::symlink_metadata(&intent) {
Ok(metadata) => Some(metadata),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => None,
Err(error) => return Err(error.into()),
};
let (active_intent, recovery_identity) = if let Some(metadata) = intent_metadata {
if metadata.file_type().is_symlink()
|| !metadata.is_file()
|| read_bounded_regular_file(&intent, 4096)?.as_deref() != Some(expected_intent)
{
return Err(DurabilityError::DataDirNotFresh(data_dir.to_path_buf()));
}
(
&intent,
Some(RecoveryArtifactIdentity::Restore(
parse_restore_intent_identity(expected_intent)
.ok_or_else(|| DurabilityError::DataDirNotFresh(data_dir.to_path_buf()))?,
)),
)
} else {
let entries = fs::read_dir(data_dir)?.collect::<Result<Vec<_>, _>>()?;
if entries.iter().all(|entry| {
let name = entry.file_name();
let name = name.to_string_lossy();
is_safe_restore_marker_tmp(&entry.path(), &name).unwrap_or(false)
}) {
for entry in entries {
fs::remove_file(entry.path())?;
}
sync_directory(data_dir)?;
return Ok(());
}
return Err(DurabilityError::DataDirNotFresh(data_dir.to_path_buf()));
};
for entry in fs::read_dir(data_dir)? {
let entry = entry?;
let name = entry.file_name();
let name = name.to_string_lossy();
let marker_tmp = is_safe_restore_marker_tmp(&entry.path(), &name)?;
if name.starts_with(RESTORE_MARKER_TMP_PREFIX) && !marker_tmp {
return Err(DurabilityError::DataDirNotFresh(data_dir.to_path_buf()));
}
let is_staging = name.starts_with(RESTORE_STAGING_PREFIX);
if is_staging
&& (!is_owned_generated_recovery_name(&name, RESTORE_STAGING_PREFIX)
|| !recovery_identity.as_ref().is_some_and(|identity| {
is_owned_recovery_directory(
&entry.path(),
&["sqlite", "ladybug", "kv", "consensus"],
RepairArtifactRole::Staging,
identity,
)
.unwrap_or(false)
}))
{
return Err(DurabilityError::DataDirNotFresh(data_dir.to_path_buf()));
}
let owned = entry.path() == active_intent.as_path()
|| completion_marker_name.is_some_and(|marker| name == marker)
|| name == "sqlite"
|| name == "ladybug"
|| name == "kv"
|| name == "consensus"
|| marker_tmp
|| is_staging;
if !owned {
return Err(DurabilityError::DataDirNotFresh(data_dir.to_path_buf()));
}
}
for name in ["sqlite", "ladybug", "kv", "consensus"] {
let path = data_dir.join(name);
if path.exists() {
fs::remove_dir_all(path)?;
}
}
for entry in fs::read_dir(data_dir)? {
let entry = entry?;
let name = entry.file_name();
let name = name.to_string_lossy();
if name.starts_with(RESTORE_STAGING_PREFIX) {
fs::remove_dir_all(entry.path())?;
} else if is_safe_restore_marker_tmp(&entry.path(), &name)? {
fs::remove_file(entry.path())?;
}
}
fs::remove_file(active_intent)?;
sync_directory(data_dir)?;
Ok(())
}
fn path_has_state(path: &Path) -> Result<bool, std::io::Error> {
let metadata = match fs::symlink_metadata(path) {
Ok(metadata) => metadata,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false),
Err(error) => return Err(error),
};
if !metadata.is_dir() {
return Ok(true);
}
fs::read_dir(path)?
.next()
.transpose()
.map(|entry| entry.is_some())
}
#[cfg(test)]
#[path = "durability/prestage_tests.rs"]
mod prestage_tests;
#[cfg(test)]
mod tests {
use super::{
mark_durable_state, observe_durable_tip, snapshot_profile, validate_local_qlog,
validate_restored_suffix, write_repair_artifact_ownership, CheckpointTip, CoordinatorState,
DurabilityError, DurabilityHealth, ExecutionProfile, LogAnchor, LogHash, PendingLag,
RecoveryArtifactIdentity, RepairArtifactRole, SuccessorRestorePreparation,
RESTORE_INTENT_FILE, SUCCESSOR_RESTORE_COMPLETE_FILE, SUCCESSOR_RESTORE_INTENT_FILE,
SUCCESSOR_RESTORE_LOCK_FILE,
};
#[cfg(feature = "kv")]
use crate::{KvCommandV1, NodeConfig, NodeRuntime};
use rhiza_archive::CheckpointIdentity;
#[cfg(feature = "kv")]
use rhiza_archive::ObjectArchiveStore;
use rhiza_core::{EntryType, LogEntry};
use rhiza_log::{FileLogStore, LogStore};
#[cfg(feature = "kv")]
use rhiza_obj_store::{ObjStore, ObjStoreConfig};
#[cfg(feature = "kv")]
use rhiza_quepaxa::ThreeNodeConsensus;
use std::sync::{Arc, Barrier, Mutex};
#[test]
fn snapshot_profile_requires_a_canonical_effective_cluster_identity() {
assert_eq!(
snapshot_profile("rhiza:sql:cluster-a").unwrap(),
ExecutionProfile::Sqlite
);
assert_eq!(
snapshot_profile("rhiza:graph:cluster-a").unwrap(),
ExecutionProfile::Graph
);
assert_eq!(
snapshot_profile("rhiza:kv:cluster-a").unwrap(),
ExecutionProfile::Kv
);
assert!(matches!(
snapshot_profile("cluster-a"),
Err(DurabilityError::SnapshotVerification(_))
));
assert!(snapshot_profile("rhiza:graph:").is_err());
}
#[test]
fn generic_restore_prepare_keeps_prefix_spoofed_staging_and_fails_closed() {
let root = tempfile::tempdir().unwrap();
let identity = CheckpointIdentity::new("rhiza:sql:cluster-a", 1, 1, 1);
let intent = super::encode_restore_intent(
&identity,
"node-1",
ExecutionProfile::Sqlite,
LogAnchor::new(0, LogHash::ZERO),
)
.unwrap();
std::fs::write(root.path().join(RESTORE_INTENT_FILE), &intent).unwrap();
let staging = root.path().join(".restore-stage-4242-1");
std::fs::create_dir_all(staging.join("sqlite")).unwrap();
assert!(matches!(
super::prepare_fresh_restore_data_dir(root.path(), None, &intent),
Err(DurabilityError::DataDirNotFresh(_))
));
assert!(staging.join("sqlite").is_dir());
}
#[test]
fn bounded_regular_reader_rejects_file_larger_than_its_limit() {
let root = tempfile::tempdir().unwrap();
let path = root.path().join("oversized");
std::fs::write(&path, b"12345").unwrap();
assert!(super::read_bounded_regular_file(&path, 4).is_err());
assert_eq!(std::fs::read(&path).unwrap(), b"12345");
}
#[test]
fn rejoin_artifact_cleanup_removes_only_owned_stale_stage_and_quarantine() {
let root = tempfile::tempdir().unwrap();
let checkpoint = LogAnchor::new(0, LogHash::ZERO);
let identity = RecoveryArtifactIdentity::Restore(super::restore_intent_identity(
&CheckpointIdentity::new("rhiza:sql:cluster-a", 1, 1, 1),
"node-1",
ExecutionProfile::Sqlite,
checkpoint,
));
let stage = root.path().join(".restore-stage-4242-1");
std::fs::create_dir_all(stage.join("sqlite")).unwrap();
write_repair_artifact_ownership(&stage, RepairArtifactRole::Staging, &identity).unwrap();
let quarantine = root.path().join(".rebuildable-quarantine-4242-2");
std::fs::create_dir_all(quarantine.join("sqlite")).unwrap();
write_repair_artifact_ownership(&quarantine, RepairArtifactRole::Quarantine, &identity)
.unwrap();
super::cleanup_owned_recovery_artifacts(root.path(), &identity).unwrap();
assert!(!stage.exists());
assert!(!quarantine.exists());
}
#[test]
fn rejoin_artifact_cleanup_keeps_foreign_prefix_artifact_without_mutation() {
let root = tempfile::tempdir().unwrap();
let spoof = root.path().join(".restore-stage-foreign");
std::fs::create_dir(&spoof).unwrap();
let identity = RecoveryArtifactIdentity::Restore(super::restore_intent_identity(
&CheckpointIdentity::new("rhiza:sql:cluster-a", 1, 1, 1),
"node-1",
ExecutionProfile::Sqlite,
LogAnchor::new(0, LogHash::ZERO),
));
assert!(matches!(
super::cleanup_owned_recovery_artifacts(root.path(), &identity),
Err(DurabilityError::DataDirNotFresh(_))
));
assert!(spoof.is_dir());
}
#[cfg(unix)]
#[test]
fn successor_completion_keeps_existing_complete_symlink_and_intent() {
use std::os::unix::fs::symlink;
let root = tempfile::tempdir().unwrap();
let receipt = br#"{"cluster_id":"rhiza:sql:cluster-a","epoch":1,"target_config_id":2,"recovery_generation":1,"node_id":"node-1","membership_digest":"0000000000000000000000000000000000000000000000000000000000000000","predecessor_config_id":1,"stop_index":0,"stop_hash":"0000000000000000000000000000000000000000000000000000000000000000"}"#;
let intent = root.path().join(SUCCESSOR_RESTORE_INTENT_FILE);
std::fs::write(&intent, receipt).unwrap();
let target = root.path().join("target");
std::fs::write(&target, b"do not replace").unwrap();
let complete = root.path().join(SUCCESSOR_RESTORE_COMPLETE_FILE);
symlink(&target, &complete).unwrap();
let lock = std::fs::OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(root.path().join(SUCCESSOR_RESTORE_LOCK_FILE))
.unwrap();
let preparation = SuccessorRestorePreparation {
tip: CheckpointTip::new(0, LogHash::ZERO),
data_dir: root.path().to_path_buf(),
identity: receipt.to_vec(),
requires_recorder_install: true,
_lock: lock,
};
assert!(preparation.complete().is_err());
assert_eq!(std::fs::read(&target).unwrap(), b"do not replace");
assert_eq!(std::fs::read(&intent).unwrap(), receipt);
assert!(std::fs::symlink_metadata(&complete)
.unwrap()
.file_type()
.is_symlink());
}
#[test]
fn sqlite_restore_suffix_rejects_noncanonical_commands_during_preflight() {
let payload = b"put\tnoncanonical\tkey\tvalue".to_vec();
let entry = LogEntry {
cluster_id: "rhiza:sql:cluster-a".into(),
epoch: 1,
config_id: 1,
index: 1,
entry_type: EntryType::Command,
prev_hash: LogHash::ZERO,
hash: LogEntry::calculate_hash(
"rhiza:sql:cluster-a",
1,
1,
1,
EntryType::Command,
LogHash::ZERO,
&payload,
),
payload,
};
assert!(matches!(
validate_restored_suffix(ExecutionProfile::Sqlite, &[entry]),
Err(DurabilityError::SnapshotVerification(message)) if message.contains("QWAL")
));
}
#[test]
fn local_qlog_accepts_ahead_tip_when_checkpoint_entry_is_retained() {
let root = tempfile::tempdir().unwrap();
let identity = CheckpointIdentity::new("rhiza:sql:cluster-a", 1, 1, 1);
let log = FileLogStore::open(
root.path().join("consensus/log"),
identity.cluster_id(),
1,
1,
)
.unwrap();
let entry = |index, previous| {
let hash = LogEntry::calculate_hash(
identity.cluster_id(),
index,
1,
1,
EntryType::Noop,
previous,
&[],
);
LogEntry {
cluster_id: identity.cluster_id().into(),
epoch: 1,
config_id: 1,
index,
entry_type: EntryType::Noop,
payload: Vec::new(),
prev_hash: previous,
hash,
}
};
let first = entry(1, LogHash::ZERO);
let second = entry(2, first.hash);
log.append_batch(&[first.clone(), second.clone()]).unwrap();
assert_eq!(
validate_local_qlog(
root.path(),
&identity,
rhiza_core::LogAnchor::new(first.index, first.hash),
)
.unwrap(),
rhiza_core::LogAnchor::new(2, second.hash)
);
assert!(validate_local_qlog(
root.path(),
&identity,
rhiza_core::LogAnchor::new(2, LogHash::digest(&[b"conflicting"])),
)
.is_err());
assert!(validate_local_qlog(
root.path(),
&identity,
rhiza_core::LogAnchor::new(3, LogHash::digest(&[b"ahead checkpoint"])),
)
.is_err());
}
#[test]
fn local_qlog_treats_absent_log_as_genesis_only() {
let root = tempfile::tempdir().unwrap();
let identity = CheckpointIdentity::new("rhiza:sql:cluster-a", 1, 1, 1);
let genesis = rhiza_core::LogAnchor::new(0, LogHash::ZERO);
assert_eq!(
validate_local_qlog(root.path(), &identity, genesis).unwrap(),
genesis
);
assert!(validate_local_qlog(
root.path(),
&identity,
rhiza_core::LogAnchor::new(1, LogHash::digest(&[b"checkpoint"])),
)
.is_err());
}
#[test]
fn restore_intent_remains_until_completion_marker_is_durable_and_retryable() {
let root = tempfile::tempdir().unwrap();
let data_dir = root.path().join("data");
std::fs::create_dir_all(&data_dir).unwrap();
let intent = super::encode_restore_intent(
&CheckpointIdentity::new("rhiza:sql:cluster-a", 1, 1, 1),
"node-1",
ExecutionProfile::Sqlite,
rhiza_core::LogAnchor::new(0, LogHash::ZERO),
)
.unwrap();
std::fs::write(data_dir.join(RESTORE_INTENT_FILE), &intent).unwrap();
std::fs::create_dir(data_dir.join("identity.json")).unwrap();
let staging = super::create_restore_staging_dir(&data_dir, None).unwrap();
assert!(super::publish_restore_staging(
&staging,
&data_dir,
true,
Some(("identity.json", b"identity-fixture")),
)
.is_err());
assert_eq!(
std::fs::read(data_dir.join(RESTORE_INTENT_FILE)).unwrap(),
intent
);
std::fs::remove_dir(data_dir.join("identity.json")).unwrap();
let retry_staging = super::create_restore_staging_dir(&data_dir, None).unwrap();
super::publish_restore_staging(
&retry_staging,
&data_dir,
true,
Some(("identity.json", b"identity-fixture")),
)
.unwrap();
assert_eq!(
std::fs::read(data_dir.join("identity.json")).unwrap(),
b"identity-fixture"
);
assert!(!data_dir.join(RESTORE_INTENT_FILE).exists());
}
#[cfg(feature = "kv")]
#[tokio::test]
async fn kv_compacted_rejoin_restores_missing_or_corrupt_views_without_touching_recorder() {
let root = tempfile::tempdir().unwrap();
let identity = CheckpointIdentity::new("rhiza:kv:cluster-a", 1, 1, 1);
let archive = ObjectArchiveStore::new_checkpoint_for_single_process(
ObjStore::new(ObjStoreConfig::Local {
root: root.path().join("archive"),
})
.unwrap(),
identity.clone(),
);
archive.initialize_checkpoint().await.unwrap();
let source_dir = root.path().join("source");
let config = NodeConfig::new_embedded(
"cluster-a",
"node-1",
source_dir,
1,
1,
["node-1", "node-2", "node-3"],
)
.unwrap()
.with_execution_profile(ExecutionProfile::Kv)
.unwrap()
.with_recovery_generation(1)
.unwrap();
let consensus = Arc::new(
ThreeNodeConsensus::from_recovered_tip(
"rhiza:kv:cluster-a",
"node-1",
1,
1,
[
root.path().join("recorders/node-1"),
root.path().join("recorders/node-2"),
root.path().join("recorders/node-3"),
],
1,
LogHash::ZERO,
)
.unwrap(),
);
let runtime = NodeRuntime::open(config, consensus, &[]).unwrap();
let coordinator = CheckpointCoordinator::open(archive.clone(), DurabilityMode::Sync)
.await
.unwrap();
let committed = runtime
.mutate_kv(KvCommandV1::put("request-1", b"key".to_vec(), b"value".to_vec()).unwrap())
.unwrap();
coordinator.note_committed(committed.applied_index());
coordinator
.flush_runtime(&runtime, committed.applied_index())
.await
.unwrap();
let checkpoint_root = runtime.checkpoint_compact(&coordinator).await.unwrap();
let target = root.path().join("target");
restore_checkpoint_to_fresh_data_dir_for_node(archive.clone(), &target, "node-1")
.await
.unwrap();
validate_local_recovery_view(
&target,
&identity,
"node-1",
ExecutionProfile::Kv,
*checkpoint_root.compacted(),
)
.unwrap();
std::fs::create_dir_all(target.join("recorder")).unwrap();
std::fs::write(target.join("recorder/sentinel"), b"keep-me").unwrap();
std::fs::remove_dir_all(target.join("consensus")).unwrap();
assert!(validate_local_recovery_view(
&target,
&identity,
"node-1",
ExecutionProfile::Kv,
*checkpoint_root.compacted(),
)
.is_err());
restore_checkpoint_for_rejoin_preserving_recorder(
archive.clone(),
&target,
"node-1",
ExecutionProfile::Kv,
"identity.json",
b"identity-fixture",
)
.await
.unwrap();
std::fs::write(target.join("kv/data.redb"), b"corrupt").unwrap();
assert!(validate_local_recovery_view(
&target,
&identity,
"node-1",
ExecutionProfile::Kv,
*checkpoint_root.compacted(),
)
.is_err());
restore_checkpoint_for_rejoin_preserving_recorder(
archive,
&target,
"node-1",
ExecutionProfile::Kv,
"identity.json",
b"identity-fixture",
)
.await
.unwrap();
assert_eq!(
std::fs::read(target.join("recorder/sentinel")).unwrap(),
b"keep-me"
);
}
#[test]
fn concurrent_flush_completion_cannot_regress_the_durable_tip() {
let newer = CheckpointTip::new(8, LogHash::digest(&[b"newer"]));
let older = CheckpointTip::new(4, LogHash::digest(&[b"older"]));
let mut state = CoordinatorState {
durable_tip: newer,
committed_index: 8,
pending_lag: None,
health: DurabilityHealth::Available,
};
mark_durable_state(&mut state, older);
assert_eq!(state.durable_tip, newer);
}
#[test]
fn checkpoint_observation_rejects_same_index_hash_conflict_without_mutation() {
let current = CheckpointTip::new(8, LogHash::digest(&[b"current"]));
let conflicting = CheckpointTip::new(8, LogHash::digest(&[b"conflicting"]));
let state = Mutex::new(CoordinatorState {
durable_tip: current,
committed_index: 8,
pending_lag: None,
health: DurabilityHealth::Available,
});
assert!(matches!(
observe_durable_tip(&state, conflicting),
Err(DurabilityError::SnapshotVerification(_))
));
assert_eq!(state.lock().unwrap().durable_tip, current);
}
#[test]
fn checkpoint_observation_rejects_remote_rollback_without_mutation() {
let current = CheckpointTip::new(8, LogHash::digest(&[b"current"]));
let older = CheckpointTip::new(7, LogHash::digest(&[b"older"]));
let state = Mutex::new(CoordinatorState {
durable_tip: current,
committed_index: 8,
pending_lag: None,
health: DurabilityHealth::Available,
});
assert!(matches!(
observe_durable_tip(&state, older),
Err(DurabilityError::SnapshotVerification(_))
));
assert_eq!(state.lock().unwrap().durable_tip, current);
}
#[test]
fn concurrent_conflicting_checkpoint_observations_accept_exactly_one_tip() {
let state = Arc::new(Mutex::new(CoordinatorState {
durable_tip: CheckpointTip::new(0, LogHash::ZERO),
committed_index: 0,
pending_lag: None,
health: DurabilityHealth::Available,
}));
let start = Arc::new(Barrier::new(3));
let results = [b"first".as_slice(), b"second".as_slice()]
.into_iter()
.map(|label| {
let state = Arc::clone(&state);
let start = Arc::clone(&start);
std::thread::spawn(move || {
let tip = CheckpointTip::new(1, LogHash::digest(&[label]));
start.wait();
(tip, observe_durable_tip(&state, tip))
})
})
.collect::<Vec<_>>();
start.wait();
let results = results
.into_iter()
.map(|thread| thread.join().unwrap())
.collect::<Vec<_>>();
assert_eq!(
results.iter().filter(|(_, result)| result.is_ok()).count(),
1
);
assert_eq!(
results.iter().filter(|(_, result)| result.is_err()).count(),
1
);
let accepted = results
.iter()
.find_map(|(tip, result)| result.is_ok().then_some(*tip))
.unwrap();
assert_eq!(state.lock().unwrap().durable_tip, accepted);
}
#[test]
fn durable_progress_clears_lag_only_after_reaching_the_committed_index() {
let mut state = CoordinatorState {
durable_tip: CheckpointTip::new(2, LogHash::digest(&[b"two"])),
committed_index: 4,
pending_lag: Some(PendingLag::Recovered),
health: DurabilityHealth::Unavailable,
};
let partial = CheckpointTip::new(3, LogHash::digest(&[b"three"]));
mark_durable_state(&mut state, partial);
assert!(state.pending_lag.is_some());
assert_eq!(state.health, DurabilityHealth::Unavailable);
let complete = CheckpointTip::new(4, LogHash::digest(&[b"four"]));
mark_durable_state(&mut state, complete);
assert_eq!(state.durable_tip, complete);
assert!(state.pending_lag.is_none());
assert_eq!(state.health, DurabilityHealth::Available);
}
}