#[cfg(test)]
use std::sync::atomic::{AtomicU8, Ordering};
use std::{
fs,
io::{Error as IoError, ErrorKind, Result as IoResult},
path::{Path, PathBuf},
};
use wyrd_spec::error::WyrdError;
use crate::cards::error::RegistryEngineError;
pub(super) struct BundleWorkspace {
staging: PathBuf,
destination: PathBuf,
}
impl BundleWorkspace {
pub(super) fn prepare(destination: &Path) -> Result<Self, WyrdError> {
let parent = destination.parent().unwrap_or_else(|| Path::new("."));
fs::create_dir_all(parent)
.map_err(RegistryEngineError::from)
.map_err(WyrdError::from)?;
match fs::metadata(destination) {
Ok(metadata) if !metadata.is_dir() => {
return Err(WyrdError::RegistryInvalidCardSpec {
message: "pull destination is not a directory".to_owned(),
details: serde_json::json!({ "destination": destination }),
});
}
Ok(_) => {}
Err(error) if error.kind() == ErrorKind::NotFound => {}
Err(error) => return Err(WyrdError::from(RegistryEngineError::from(error))),
}
let staging = parent.join(format!(
".{}.staging-{}",
destination
.file_name()
.and_then(|name| name.to_str())
.unwrap_or("wyrd-state"),
uuid::Uuid::now_v7()
));
fs::create_dir(&staging)
.map_err(RegistryEngineError::from)
.map_err(WyrdError::from)?;
Ok(Self {
staging,
destination: destination.to_path_buf(),
})
}
pub(super) fn staging(&self) -> &Path {
&self.staging
}
pub(super) fn destination(&self) -> &Path {
&self.destination
}
pub(super) fn abort(&self) {
let _ = fs::remove_dir_all(&self.staging);
}
pub(super) fn publish(&self) -> Result<(), WyrdError> {
if fs::metadata(&self.destination).is_err() {
fs::rename(&self.staging, &self.destination)
.map_err(RegistryEngineError::from)
.map_err(WyrdError::from)?;
return Ok(());
}
let backup = self.backup_path();
fs::rename(&self.destination, &backup)
.map_err(RegistryEngineError::from)
.map_err(WyrdError::from)?;
if let Err(error) = promote(&self.staging, &self.destination) {
return self.restore_after_failure(&backup, error);
}
cleanup_backup(&backup, &self.destination);
Ok(())
}
fn backup_path(&self) -> PathBuf {
let parent = self.destination.parent().unwrap_or_else(|| Path::new("."));
parent.join(format!(
".{}.previous-{}",
self.destination
.file_name()
.and_then(|name| name.to_str())
.unwrap_or("wyrd-state"),
uuid::Uuid::now_v7()
))
}
fn restore_after_failure(
&self,
backup: &Path,
promotion_error: IoError,
) -> Result<(), WyrdError> {
let publication_error = WyrdError::from(RegistryEngineError::from(promotion_error));
if let Err(error) = restore(backup, &self.destination) {
let restoration_error = RegistryEngineError::from(error);
tracing::error!(
error = %restoration_error,
destination = %self.destination.display(),
backup = %backup.display(),
"pull publication failed and rollback failed"
);
return Err(WyrdError::Internal {
message: "pull publication and rollback failed".to_owned(),
details: serde_json::json!({
"publication_error": publication_error.to_string(),
"rollback_error": restoration_error.to_string(),
"destination": self.destination,
"backup": backup,
}),
});
}
Err(publication_error)
}
}
fn promote(staging: &Path, destination: &Path) -> IoResult<()> {
#[cfg(test)]
if take_publish_fault(FAIL_PROMOTION) {
return Err(IoError::other("injected pull promotion failure"));
}
fs::rename(staging, destination)
}
fn restore(backup: &Path, destination: &Path) -> IoResult<()> {
#[cfg(test)]
if take_publish_fault(FAIL_RESTORE) {
return Err(IoError::other("injected pull restoration failure"));
}
fs::rename(backup, destination)
}
fn cleanup_backup(backup: &Path, destination: &Path) {
#[cfg(test)]
let cleanup = if take_publish_fault(FAIL_CLEANUP) {
Err(IoError::other("injected pull backup cleanup failure"))
} else {
fs::remove_dir_all(backup)
};
#[cfg(not(test))]
let cleanup = fs::remove_dir_all(backup);
if let Err(error) = cleanup {
tracing::warn!(
error = %error,
backup = %backup.display(),
destination = %destination.display(),
"published pull bundle but could not remove the previous bundle"
);
}
}
#[cfg(test)]
const FAIL_PROMOTION: u8 = 1;
#[cfg(test)]
const FAIL_RESTORE: u8 = 2;
#[cfg(test)]
const FAIL_CLEANUP: u8 = 4;
#[cfg(test)]
static PUBLISH_FAULTS: AtomicU8 = AtomicU8::new(0);
#[cfg(test)]
fn take_publish_fault(fault: u8) -> bool {
PUBLISH_FAULTS.fetch_and(!fault, Ordering::SeqCst) & fault != 0
}
#[cfg(test)]
mod tests {
use std::path::{Path, PathBuf};
use std::sync::Mutex;
use std::sync::atomic::Ordering;
use tempfile::TempDir;
use super::{BundleWorkspace, FAIL_CLEANUP, FAIL_PROMOTION, FAIL_RESTORE, PUBLISH_FAULTS};
static PUBLISH_FAULT_LOCK: Mutex<()> = Mutex::new(());
fn inject_publish_faults(faults: u8) {
PUBLISH_FAULTS.store(faults, Ordering::SeqCst);
}
fn write_bundle_marker(directory: &Path, marker: &str) {
std::fs::create_dir_all(directory).expect("bundle directory creates");
std::fs::write(directory.join("marker"), marker).expect("bundle marker writes");
}
fn test_workspace(staging: PathBuf, destination: PathBuf) -> BundleWorkspace {
BundleWorkspace {
staging,
destination,
}
}
#[test]
fn failed_replacement_restores_the_previous_bundle() {
let _publish_fault_guard = PUBLISH_FAULT_LOCK
.lock()
.expect("publication fault lock remains available");
inject_publish_faults(0);
let temp = TempDir::new().expect("tempdir creates");
let destination = temp.path().join("bundle");
let staging = temp.path().join("missing-staging");
write_bundle_marker(&destination, "previous");
let error = test_workspace(staging, destination.clone())
.publish()
.expect_err("missing staging must fail replacement");
assert!(
error
.to_string()
.contains("local artifact materialization failed")
);
assert_eq!(
std::fs::read_to_string(destination.join("marker")).expect("previous bundle reads"),
"previous"
);
}
#[test]
fn rollback_failure_is_reported_as_unrecoverable() {
let _publish_fault_guard = PUBLISH_FAULT_LOCK
.lock()
.expect("publication fault lock remains available");
inject_publish_faults(0);
let temp = TempDir::new().expect("tempdir creates");
let destination = temp.path().join("bundle");
let staging = temp.path().join("staging");
write_bundle_marker(&destination, "previous");
write_bundle_marker(&staging, "new");
inject_publish_faults(FAIL_PROMOTION | FAIL_RESTORE);
let error = test_workspace(staging, destination.clone())
.publish()
.expect_err("injected promotion must fail");
assert!(
error
.to_string()
.contains("pull publication and rollback failed")
);
assert!(!destination.exists());
assert!(
temp.path()
.read_dir()
.expect("publication parent reads")
.flatten()
.any(|entry| entry.file_name().to_string_lossy().contains(".previous-"))
);
}
#[test]
fn cleanup_failure_does_not_hide_a_published_bundle() {
let _publish_fault_guard = PUBLISH_FAULT_LOCK
.lock()
.expect("publication fault lock remains available");
inject_publish_faults(0);
let temp = TempDir::new().expect("tempdir creates");
let destination = temp.path().join("bundle");
let staging = temp.path().join("staging");
write_bundle_marker(&destination, "previous");
write_bundle_marker(&staging, "new");
inject_publish_faults(FAIL_CLEANUP);
test_workspace(staging, destination.clone())
.publish()
.expect("cleanup failure is best effort");
assert_eq!(
std::fs::read_to_string(destination.join("marker")).expect("new bundle reads"),
"new"
);
}
}