use crate::diagnostic_log::warn;
use serde_json::to_string as json_string;
use std::path::{Path, PathBuf};
use std::time::Duration;
pub const PHASE_SIGNAL_VARIABLE: &str = "CLI_GIT_TEST_ONLY_PHASE_SIGNAL";
use crate::wait_ceilings::PHASE_PATIENCE;
const RELEASE_POLL: Duration = Duration::from_millis(20);
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum Phase {
MergeIntent,
MergeApplied,
ServiceStaged,
ServicePrepared,
}
pub const TRANSACTION_TEST_PHASES: [Phase; 4] = [
Phase::MergeIntent,
Phase::MergeApplied,
Phase::ServiceStaged,
Phase::ServicePrepared,
];
impl Phase {
pub fn name(self) -> &'static str {
match self {
Phase::MergeIntent => return "merge-intent",
Phase::MergeApplied => return "merge-applied",
Phase::ServiceStaged => return "service-staged",
Phase::ServicePrepared => return "service-prepared",
}
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum PhaseAction {
Kill,
Pause,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PhaseSignal {
pub phase: Phase,
pub action: PhaseAction,
pub directory: Option<PathBuf>,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PhaseSignalError {
pub value: String,
}
impl std::fmt::Display for PhaseSignalError {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let quoted: String = json_string(self.value.as_str()).unwrap_or_else(|_| format!("{:?}", self.value));
let phases: Vec<&str> = TRANSACTION_TEST_PHASES.iter().map(|phase| { return phase.name(); }).collect();
return write!(
formatter,
"{PHASE_SIGNAL_VARIABLE}={quoted} is not <phase>:kill[:<absolute directory>] or <phase>:pause:<absolute directory> with a phase among {}.",
phases.join(", ")
)
}
}
impl std::error::Error for PhaseSignalError {}
#[derive(Debug)]
pub enum PhaseMarkerError {
Malformed(PhaseSignalError),
Io(std::io::Error),
}
impl std::fmt::Display for PhaseMarkerError {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
PhaseMarkerError::Malformed(error) => return write!(formatter, "{error}"),
PhaseMarkerError::Io(error) => return write!(formatter, "transaction phase marker failed: {error}"),
}
}
}
impl std::error::Error for PhaseMarkerError {}
pub fn parse_phase_signal(value: &str) -> Result<PhaseSignal, PhaseSignalError> {
let refusal = || {
return PhaseSignalError {
value: value.to_string(),
};
};
let Some(phase_end) = value.find(':') else {
return Err(refusal());
};
let rest: &str = &value[phase_end + 1..];
let (action_text, directory_text) = match rest.find(':') {
Some(action_end) => (&rest[..action_end], &rest[action_end + 1..]),
None => (rest, ""),
};
let Some(phase) = TRANSACTION_TEST_PHASES
.iter()
.copied()
.find(|candidate| { return candidate.name() == &value[..phase_end]; })
else {
return Err(refusal());
};
let action: PhaseAction = match action_text {
"kill" => PhaseAction::Kill,
"pause" => PhaseAction::Pause,
_ => return Err(refusal()),
};
if (action == PhaseAction::Pause && directory_text.is_empty())
|| (!directory_text.is_empty() && !Path::new(directory_text).is_absolute())
{
return Err(refusal());
}
return Ok(PhaseSignal {
phase,
action,
directory: if directory_text.is_empty() {
None
} else {
Some(PathBuf::from(directory_text))
},
});
}
#[cfg(unix)]
fn kill_self() -> ! {
unsafe {
libc::kill(libc::getpid(), libc::SIGKILL);
}
std::process::abort()
}
#[cfg(not(unix))]
fn kill_self() -> ! {
std::process::exit(137)
}
fn await_release(directory: &Path, release: &Path) -> Result<(), PhaseMarkerError> {
return await_release_within(directory, release, PHASE_PATIENCE);
}
fn await_release_within(directory: &Path, release: &Path, patience: Duration) -> Result<(), PhaseMarkerError> {
let began: std::time::Instant = std::time::Instant::now();
while !release.exists() {
if began.elapsed() >= patience {
return Err(PhaseMarkerError::Io(std::io::Error::new(
std::io::ErrorKind::TimedOut,
format!("nothing released the pause in {} within {} seconds", directory.display(), patience.as_secs()),
)));
}
if !directory.exists() {
return Err(PhaseMarkerError::Io(std::io::Error::new(
std::io::ErrorKind::NotFound,
format!("the directory {} of the pause is gone, so nothing can release it", directory.display()),
)));
}
std::thread::sleep(RELEASE_POLL);
}
return Ok(());
}
pub(crate) fn reach_with_value(
value: Option<&str>,
phase: Phase,
occurrence: Option<u64>,
) -> Result<(), PhaseMarkerError> {
let Some(armed) = value.filter(|text| { return !text.is_empty(); }) else {
return Ok(());
};
let signal: PhaseSignal = parse_phase_signal(armed).map_err(PhaseMarkerError::Malformed)?;
if signal.phase != phase {
return Ok(());
}
let stem: String = match occurrence {
Some(number) => format!("{}-{number}", phase.name()),
None => phase.name().to_string(),
};
warn(
"transactionPhase",
&format!("{PHASE_SIGNAL_VARIABLE} reached {stem}: {:?}", signal.action),
);
if let Some(directory) = signal.directory.as_deref() {
std::fs::write(directory.join(format!("{stem}.reached")), std::process::id().to_string())
.map_err(PhaseMarkerError::Io)?;
}
match signal.action {
PhaseAction::Kill => kill_self(),
PhaseAction::Pause => {
let directory: &Path = signal.directory.as_deref().unwrap_or(Path::new(""));
return await_release(directory, &directory.join(format!("{stem}.release")));
}
}
}
pub fn reach_transaction_phase(phase: Phase, occurrence: Option<u64>) -> Result<(), PhaseMarkerError> {
let value: Option<String> = std::env::var_os(PHASE_SIGNAL_VARIABLE).map(|text: std::ffi::OsString| { return text.to_string_lossy().into_owned(); });
return reach_with_value(value.as_deref(), phase, occurrence);
}
#[cfg(test)]
#[path = "transaction_phase_tests.rs"]
mod tests;