use std::collections::HashMap;
use std::sync::{Arc, Mutex, PoisonError};
use arc_swap::ArcSwap;
use thiserror::Error;
use crate::compat::{Compatibility, Fingerprint, StagedBundle};
use crate::stager::{
Activator, HealthCheck, Outcome, ReleaseId, SignatureVerifier, StageError, Stager, UpdateSource,
};
pub struct HotConfig<C> {
cell: Arc<ArcSwap<C>>,
}
impl<C> Clone for HotConfig<C> {
fn clone(&self) -> Self {
Self {
cell: Arc::clone(&self.cell),
}
}
}
impl<C> HotConfig<C> {
#[must_use]
pub fn new(initial: C) -> Self {
Self {
cell: Arc::new(ArcSwap::from_pointee(initial)),
}
}
#[must_use]
pub fn from_arc(initial: Arc<C>) -> Self {
Self {
cell: Arc::new(ArcSwap::new(initial)),
}
}
#[must_use]
pub fn current(&self) -> Arc<C> {
self.cell.load_full()
}
pub fn install(&self, next: Arc<C>) {
self.cell.store(next);
}
}
pub struct HotConfigActivator<C> {
live: HotConfig<C>,
staged: Arc<Mutex<HashMap<ReleaseId, Arc<C>>>>,
}
impl<C> Clone for HotConfigActivator<C> {
fn clone(&self) -> Self {
Self {
live: self.live.clone(),
staged: Arc::clone(&self.staged),
}
}
}
impl<C> HotConfigActivator<C> {
#[must_use]
pub fn new(live: HotConfig<C>, initial: ReleaseId) -> Self {
let mut staged = HashMap::new();
staged.insert(initial, live.current());
Self {
live,
staged: Arc::new(Mutex::new(staged)),
}
}
pub fn stage(&self, release: ReleaseId, value: Arc<C>) {
self.lock().insert(release, value);
}
#[must_use]
pub fn live(&self) -> HotConfig<C> {
self.live.clone()
}
fn lock(&self) -> std::sync::MutexGuard<'_, HashMap<ReleaseId, Arc<C>>> {
self.staged.lock().unwrap_or_else(PoisonError::into_inner)
}
}
impl<C> Activator for HotConfigActivator<C> {
fn activate(&self, release: &ReleaseId) -> Result<(), StageError> {
let value = self.lock().get(release).map(Arc::clone);
value.map_or_else(
|| {
Err(StageError::Activate(format!(
"no staged config registered for release {release}"
)))
},
|value| {
self.live.install(value);
Ok(())
},
)
}
}
#[derive(Debug, Error)]
pub enum HotReloadError {
#[error("refused: only a hot config bundle reloads without a restart ({})", verdict_label(.0))]
NotHot(Compatibility),
#[error(transparent)]
Stage(#[from] StageError),
}
const fn verdict_label(verdict: &Compatibility) -> &'static str {
match verdict {
Compatibility::Hot => "hot",
Compatibility::Warm => "a binary change that needs a restart",
Compatibility::Cold => "a format change that needs a coordinated redeploy",
Compatibility::Incompatible(_) => "built for a different runtime",
}
}
pub fn ensure_hot(verdict: &Compatibility) -> Result<(), HotReloadError> {
if *verdict == Compatibility::Hot {
Ok(())
} else {
Err(HotReloadError::NotHot(verdict.clone()))
}
}
pub fn apply_hot_reload<S, V, A, H>(
running: &Fingerprint,
bundle: &StagedBundle,
stager: &mut Stager<S, V, A, H>,
release: &ReleaseId,
) -> Result<Outcome, HotReloadError>
where
S: UpdateSource,
V: SignatureVerifier,
A: Activator,
H: HealthCheck,
{
ensure_hot(&bundle.evaluate(running))?;
Ok(stager.stage_and_apply(release)?)
}
#[cfg(test)]
mod tests {
#![allow(clippy::pedantic, clippy::nursery, missing_docs)]
use std::path::PathBuf;
use super::*;
use crate::compat::RuntimeTarget;
use crate::stager::{Health, StagedArtifact};
#[derive(Debug, Clone, PartialEq, Eq)]
struct Cfg {
tag: &'static str,
}
impl Cfg {
fn new(tag: &'static str) -> Arc<Self> {
Arc::new(Self { tag })
}
}
fn running() -> Fingerprint {
Fingerprint::new(3, 7, "polychrome.uno/v1", "catalog-v1")
}
fn hot_bundle() -> StagedBundle {
StagedBundle::new(running().runtime_target(), "catalog-v2")
}
fn artifact_for(release: &ReleaseId) -> StagedArtifact {
StagedArtifact {
release: release.clone(),
staged_path: PathBuf::from(format!("/var/lib/polychrome/staged/{release}")),
bundle: hot_bundle(),
signed_bytes: format!("bytes-of-{release}").into_bytes(),
signature: vec![0xAB; 4],
signer_public_key: vec![0xCD; 4],
}
}
fn ok_source(release: &ReleaseId) -> Result<StagedArtifact, StageError> {
Ok(artifact_for(release))
}
#[test]
fn a_swap_lands_on_the_next_turn_never_an_in_flight_one() {
let live = HotConfig::new(Cfg { tag: "v1" });
let turn_a = live.current();
live.install(Cfg::new("v2"));
let turn_b = live.current();
assert_eq!(
turn_a.tag, "v1",
"in-flight turn keeps its turn-start snapshot"
);
assert_eq!(turn_b.tag, "v2", "the next turn reads the new bundle");
assert_eq!(live.current().tag, "v2", "the live handle now serves v2");
}
#[test]
fn ensure_hot_admits_only_hot() {
assert!(ensure_hot(&Compatibility::Hot).is_ok());
for verdict in [
Compatibility::Warm,
Compatibility::Cold,
Compatibility::Incompatible(crate::compat::Incompatibility::Wire),
] {
let err = ensure_hot(&verdict).unwrap_err();
assert!(
matches!(err, HotReloadError::NotHot(v) if v == verdict),
"non-hot verdict must be refused: {verdict:?}",
);
}
}
#[test]
fn apply_refuses_a_bundle_built_for_another_runtime_before_touching_anything() {
let live = HotConfig::new(Cfg { tag: "v1" });
let activator = HotConfigActivator::new(live.clone(), ReleaseId::new("v1"));
let bundle = StagedBundle::new(RuntimeTarget::new(4, 7, "polychrome.uno/v1"), "catalog-v2");
let mut stager = Stager::new(
|_: &ReleaseId| -> Result<StagedArtifact, StageError> {
panic!("download must not run for a non-hot bundle")
},
|_: &StagedArtifact| panic!("verify must not run for a non-hot bundle"),
activator,
|| panic!("health check must not run for a non-hot bundle"),
ReleaseId::new("v1"),
);
let err =
apply_hot_reload(&running(), &bundle, &mut stager, &ReleaseId::new("v2")).unwrap_err();
assert!(matches!(err, HotReloadError::NotHot(_)));
assert_eq!(live.current().tag, "v1");
}
#[test]
fn apply_never_swaps_an_unverified_bundle() {
let live = HotConfig::new(Cfg { tag: "v1" });
let activator = HotConfigActivator::new(live.clone(), ReleaseId::new("v1"));
activator.stage(ReleaseId::new("v2"), Cfg::new("v2"));
let mut stager = Stager::new(
ok_source,
|_: &StagedArtifact| false,
activator,
|| panic!("health check must not run for an unverified bundle"),
ReleaseId::new("v1"),
);
let err = apply_hot_reload(
&running(),
&hot_bundle(),
&mut stager,
&ReleaseId::new("v2"),
)
.unwrap_err();
assert!(matches!(err, HotReloadError::Stage(StageError::Unverified)));
assert_eq!(live.current().tag, "v1");
}
#[test]
fn a_verified_hot_reload_swaps_the_config_for_the_next_turn() {
let live = HotConfig::new(Cfg { tag: "v1" });
let activator = HotConfigActivator::new(live.clone(), ReleaseId::new("v1"));
activator.stage(ReleaseId::new("v2"), Cfg::new("v2"));
let mut stager = Stager::new(
ok_source,
|_: &StagedArtifact| true,
activator,
|| Health::Healthy,
ReleaseId::new("v1"),
);
let in_flight = live.current();
let outcome = apply_hot_reload(
&running(),
&hot_bundle(),
&mut stager,
&ReleaseId::new("v2"),
)
.unwrap();
assert_eq!(
outcome,
Outcome::Committed {
version: ReleaseId::new("v2"),
}
);
assert_eq!(
in_flight.tag, "v1",
"in-flight turn completes on the prior bundle"
);
assert_eq!(
live.current().tag,
"v2",
"the next turn reads the reloaded bundle"
);
}
#[test]
fn a_failed_health_check_flips_the_config_back_to_the_previous_bundle() {
let live = HotConfig::new(Cfg { tag: "v1" });
let activator = HotConfigActivator::new(live.clone(), ReleaseId::new("v1"));
activator.stage(ReleaseId::new("v2"), Cfg::new("v2"));
let mut stager = Stager::new(
ok_source,
|_: &StagedArtifact| true,
activator,
|| Health::Unhealthy("readiness probe timed out".to_owned()),
ReleaseId::new("v1"),
);
let outcome = apply_hot_reload(
&running(),
&hot_bundle(),
&mut stager,
&ReleaseId::new("v2"),
)
.unwrap();
assert_eq!(
outcome,
Outcome::RolledBack {
stayed_on: ReleaseId::new("v1"),
reason: "readiness probe timed out".to_owned(),
}
);
assert_eq!(
live.current().tag,
"v1",
"rollback restores the previous bundle"
);
}
#[test]
fn not_hot_error_reads_plainly() {
let msg = HotReloadError::NotHot(Compatibility::Warm).to_string();
assert_eq!(
msg,
"refused: only a hot config bundle reloads without a restart \
(a binary change that needs a restart)",
);
for banned in ["sorry", "please", "unfortunately"] {
assert!(!msg.to_lowercase().contains(banned));
}
}
}