use std::fs::OpenOptions;
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
use super::{
assess, phase_for, validate_sample, ControllerConfig, ControllerPhase, ConvergenceSample,
RECOVER_CUTOVER, RESUME_CATCH_UP,
};
use crate::mutation::atomic_file::{
capture, open_regular_file_allow_hard_links, path_exists, promote, sync_directory,
validate_regular_file, validate_workspace,
};
use crate::MemoryError;
const STATE_VERSION: u32 = 1;
const STATE_FILE: &str = "online-migration-controller.json";
const STAGING_FILE: &str = "online-migration-controller.json.tmp";
const MAX_STATE_BYTES: u64 = 64 * 1024;
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub(super) struct ControllerState {
pub(super) version: u32,
pub(super) epoch_id: String,
pub(super) config: ControllerConfig,
pub(super) phase: ControllerPhase,
pub(super) samples: Vec<ConvergenceSample>,
pub(super) last_observation: Option<ConvergenceSample>,
pub(super) last_verdict: Option<super::ConvergenceVerdict>,
pub(super) recovery_action: Option<String>,
#[serde(default)]
pub(super) measured_cutover: Option<std::time::Duration>,
}
pub(super) struct StateStore {
workspace: PathBuf,
path: PathBuf,
}
impl StateStore {
pub(super) fn open(
workspace: &Path,
epoch_id: &str,
config: ControllerConfig,
) -> Result<(Self, ControllerState, bool), MemoryError> {
validate_workspace(workspace, "controller")?;
validate_epoch_id(epoch_id)?;
let store = Self {
workspace: workspace.to_owned(),
path: workspace.join(STATE_FILE),
};
store.recover_staging()?;
let resumed = path_exists(&store.path)?;
let state = if resumed {
store.load(epoch_id, config)?
} else {
let state = ControllerState {
version: STATE_VERSION,
epoch_id: epoch_id.to_owned(),
config,
phase: ControllerPhase::CatchingUp,
samples: Vec::with_capacity(config.observation_window),
last_observation: None,
last_verdict: None,
recovery_action: None,
measured_cutover: None,
};
store.save(&state)?;
state
};
Ok((store, state, resumed))
}
pub(super) fn save(&self, state: &ControllerState) -> Result<(), MemoryError> {
self.recover_staging()?;
let bytes = serde_json::to_vec_pretty(state)
.map_err(|err| capture(format!("cannot encode controller state: {err}")))?;
if bytes.len() as u64 > MAX_STATE_BYTES {
return Err(capture("controller state exceeds 64 KiB safety limit"));
}
let staging = self.workspace.join(STAGING_FILE);
write_synced(&staging, &bytes)?;
promote(&staging, &self.path)
.map_err(|err| capture(format!("cannot publish controller state: {err}")))?;
sync_directory(&self.workspace)
.map_err(|err| capture(format!("cannot sync controller directory: {err}")))
}
fn load(
&self,
epoch_id: &str,
config: ControllerConfig,
) -> Result<ControllerState, MemoryError> {
let bytes = read_state_bytes(&self.path)?;
let state: ControllerState = serde_json::from_slice(&bytes)
.map_err(|err| capture(format!("invalid controller state: {err}")))?;
validate_loaded(&state, epoch_id, config)?;
Ok(state)
}
fn recover_staging(&self) -> Result<(), MemoryError> {
let staging = self.workspace.join(STAGING_FILE);
if !path_exists(&staging)? {
return Ok(());
}
validate_regular_file(&staging, "controller")?;
std::fs::remove_file(&staging)
.map_err(|err| capture(format!("cannot remove controller staging file: {err}")))
}
}
fn read_state_bytes(path: &Path) -> Result<Vec<u8>, MemoryError> {
let mut options = OpenOptions::new();
options.read(true);
let (mut file, metadata) = open_regular_file_allow_hard_links(path, "controller", options)?;
let length = metadata.len();
if length > MAX_STATE_BYTES {
return Err(capture("controller state exceeds 64 KiB safety limit"));
}
let capacity = usize::try_from(length)
.map_err(|_| capture("controller state length does not fit memory"))?;
let mut bytes = Vec::with_capacity(capacity);
file.read_to_end(&mut bytes)
.map_err(|err| capture(format!("cannot read controller state: {err}")))?;
Ok(bytes)
}
fn validate_loaded(
state: &ControllerState,
epoch_id: &str,
config: ControllerConfig,
) -> Result<(), MemoryError> {
validate_header(state, epoch_id, config)?;
validate_samples(&state.samples)?;
if let Some(sample) = state.last_observation.as_ref() {
validate_sample(None, sample)?;
}
validate_audit_fields(state)?;
validate_phase(state)
}
fn validate_header(
state: &ControllerState,
epoch_id: &str,
config: ControllerConfig,
) -> Result<(), MemoryError> {
if state.version != STATE_VERSION {
return Err(capture(format!(
"unsupported controller state version {}",
state.version
)));
}
if state.epoch_id != epoch_id {
return Err(capture("controller epoch ownership mismatch"));
}
if state.config != config || state.samples.len() > config.observation_window {
return Err(capture("controller configuration mismatch"));
}
Ok(())
}
fn validate_audit_fields(state: &ControllerState) -> Result<(), MemoryError> {
if state.last_observation.is_some() != state.last_verdict.is_some() {
return Err(capture("incomplete durable controller audit state"));
}
if let Some(last) = state.samples.last() {
if state.last_observation.as_ref() != Some(last)
|| state.last_verdict != Some(assess(&state.samples, state.config).verdict)
{
return Err(capture(
"controller audit state disagrees with measured window",
));
}
}
Ok(())
}
fn validate_samples(samples: &[ConvergenceSample]) -> Result<(), MemoryError> {
let mut previous = None;
for sample in samples {
validate_sample(previous, sample)?;
previous = Some(sample);
}
Ok(())
}
fn validate_phase(state: &ControllerState) -> Result<(), MemoryError> {
if state.phase != ControllerPhase::Activated && state.measured_cutover.is_some() {
return Err(capture("cutover duration exists before activation"));
}
match state.phase {
ControllerPhase::CatchingUp
| ControllerPhase::CutoverReady
| ControllerPhase::NonConverging => validate_observation_phase(state),
ControllerPhase::Quiescing { deadline } => validate_quiescing(state, deadline),
ControllerPhase::Activated => validate_activated(state),
ControllerPhase::Cancelled => validate_recovery(state, &[None]),
}
}
fn validate_observation_phase(state: &ControllerState) -> Result<(), MemoryError> {
if state.samples.is_empty() && state.phase != ControllerPhase::CatchingUp {
return Err(capture("controller phase requires a measured window"));
}
let resuming = state.phase == ControllerPhase::CatchingUp
&& state.recovery_action.as_deref() == Some(RESUME_CATCH_UP);
if !resuming
&& !state.samples.is_empty()
&& phase_for(assess(&state.samples, state.config).verdict) != state.phase
{
return Err(capture("controller phase disagrees with measured window"));
}
let allowed = if state.phase == ControllerPhase::CatchingUp {
[None, Some(RESUME_CATCH_UP)]
} else {
[None, None]
};
validate_recovery(state, &allowed)
}
fn validate_quiescing(
state: &ControllerState,
deadline: std::time::Duration,
) -> Result<(), MemoryError> {
if !has_cutover_ready_window(state)
|| state
.samples
.last()
.is_some_and(|sample| deadline < sample.observed_at)
{
return Err(capture("invalid durable quiescing state"));
}
validate_recovery(state, &[None, Some(RECOVER_CUTOVER)])
}
fn validate_activated(state: &ControllerState) -> Result<(), MemoryError> {
if !has_cutover_ready_window(state)
|| state
.measured_cutover
.is_none_or(|elapsed| elapsed > state.config.pause_budget)
{
return Err(capture("invalid durable activated state"));
}
validate_recovery(state, &[None, Some(RECOVER_CUTOVER)])
}
fn has_cutover_ready_window(state: &ControllerState) -> bool {
!state.samples.is_empty()
&& phase_for(assess(&state.samples, state.config).verdict) == ControllerPhase::CutoverReady
}
fn validate_recovery(state: &ControllerState, allowed: &[Option<&str>]) -> Result<(), MemoryError> {
if allowed.contains(&state.recovery_action.as_deref()) {
Ok(())
} else {
Err(capture("invalid durable controller recovery action"))
}
}
fn write_synced(path: &Path, bytes: &[u8]) -> Result<(), MemoryError> {
let mut file = OpenOptions::new()
.write(true)
.create_new(true)
.open(path)
.map_err(|err| capture(format!("cannot create controller staging file: {err}")))?;
file.write_all(bytes)
.and_then(|()| file.flush())
.and_then(|()| file.sync_all())
.map_err(|err| capture(format!("cannot sync controller state: {err}")))
}
fn validate_epoch_id(epoch_id: &str) -> Result<(), MemoryError> {
if epoch_id.len() != 32 || !epoch_id.bytes().all(|byte| byte.is_ascii_hexdigit()) {
return Err(capture("epoch id must be 32 hexadecimal characters"));
}
Ok(())
}