use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::{Arc, OnceLock};
use std::time::Instant;
use arc_swap::{ArcSwap, ArcSwapOption};
use crate::error::Error;
use crate::reload::{ConfigStatus, FailureStatus, ReloadEvent, ReloadReason};
type Hook<T> = Arc<dyn Fn(&Arc<T>, &Arc<T>) + Send + Sync>;
type EventHook<T> = Arc<dyn Fn(&ReloadEvent<T>) + Send + Sync>;
enum Callback<T> {
Pair(Hook<T>),
Event(EventHook<T>),
}
impl<T> Clone for Callback<T> {
fn clone(&self) -> Self {
match self {
Self::Pair(hook) => Self::Pair(Arc::clone(hook)),
Self::Event(hook) => Self::Event(Arc::clone(hook)),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub struct SnapshotMeta {
pub generation: u64,
pub loaded_at: Instant,
}
struct Registered<T> {
token: u64,
callback: Callback<T>,
}
impl<T> Clone for Registered<T> {
fn clone(&self) -> Self {
Self {
token: self.token,
callback: self.callback.clone(),
}
}
}
pub struct ConfigCell<T> {
inner: OnceLock<ArcSwap<T>>,
hooks: OnceLock<ArcSwap<Vec<Registered<T>>>>,
next_token: std::sync::atomic::AtomicU64,
meta: ArcSwapOption<SnapshotMeta>,
last_reason: ArcSwapOption<ReloadReason>,
last_failure: ArcSwapOption<FailureStatus>,
consecutive_failures: AtomicU32,
#[cfg(feature = "async")]
notify: crate::asynchronous::Notify,
}
impl<T> ConfigCell<T> {
#[must_use]
#[cfg(not(loom))]
pub const fn new() -> Self {
Self {
inner: OnceLock::new(),
hooks: OnceLock::new(),
next_token: std::sync::atomic::AtomicU64::new(0),
meta: ArcSwapOption::const_empty(),
last_reason: ArcSwapOption::const_empty(),
last_failure: ArcSwapOption::const_empty(),
consecutive_failures: AtomicU32::new(0),
#[cfg(feature = "async")]
notify: crate::asynchronous::Notify::new(),
}
}
#[must_use]
#[cfg(loom)]
pub fn new() -> Self {
Self {
inner: OnceLock::new(),
hooks: OnceLock::new(),
next_token: std::sync::atomic::AtomicU64::new(0),
meta: ArcSwapOption::const_empty(),
last_reason: ArcSwapOption::const_empty(),
last_failure: ArcSwapOption::const_empty(),
consecutive_failures: AtomicU32::new(0),
#[cfg(feature = "async")]
notify: crate::asynchronous::Notify::new(),
}
}
pub fn store(&self, value: T) {
self.store_with(value, ReloadReason::Manual);
}
#[allow(clippy::must_use_candidate)]
pub fn store_with(&self, value: T, reason: ReloadReason) -> Arc<T> {
let value = Arc::new(value);
let slot = self.inner.get_or_init(|| ArcSwap::new(Arc::clone(&value)));
let previous = slot.swap(Arc::clone(&value));
let mut installed = None;
self.meta.rcu(|before| {
let meta = Arc::new(SnapshotMeta {
generation: before.as_ref().map_or(0, |meta| meta.generation) + 1,
loaded_at: Instant::now(),
});
installed = Some(*meta);
meta
});
let meta = installed.expect("`rcu` runs its closure at least once");
self.last_reason.store(Some(Arc::new(reason.clone())));
self.consecutive_failures.store(0, Ordering::Relaxed);
#[cfg(feature = "async")]
self.notify.bump();
let previous = if Arc::ptr_eq(&previous, &value) {
None
} else {
Some(previous)
};
#[cfg(feature = "tracing")]
let _span = crate::telemetry::installed::<T>(&reason, meta.generation);
self.dispatch(previous, &value, reason, meta);
value
}
pub fn record_failure(&self, error: &Error) {
let mut count = self.consecutive_failures.load(Ordering::Relaxed);
while count < u32::MAX {
match self.consecutive_failures.compare_exchange_weak(
count,
count + 1,
Ordering::Relaxed,
Ordering::Relaxed,
) {
Ok(_) => break,
Err(actual) => count = actual,
}
}
self.last_failure
.store(Some(Arc::new(FailureStatus::of(error))));
#[cfg(feature = "tracing")]
crate::telemetry::refused::<T>(error);
}
#[must_use]
pub fn status(&self) -> ConfigStatus {
let meta = self.meta();
ConfigStatus {
generation: meta.map_or(0, |meta| meta.generation),
loaded_at: meta.map(|meta| meta.loaded_at),
last_reason: self.last_reason.load_full().map(|reason| (*reason).clone()),
last_failure: self
.last_failure
.load_full()
.map(|failure| (*failure).clone()),
consecutive_failures: self.consecutive_failures.load(Ordering::Relaxed),
}
}
pub fn on_reload(&self, hook: impl Fn(&Arc<T>, &Arc<T>) + Send + Sync + 'static) {
let _ = self.register(Callback::Pair(Arc::new(hook)));
}
pub fn on_reload_with(&self, hook: impl Fn(&ReloadEvent<T>) + Send + Sync + 'static) {
let _ = self.register(Callback::Event(Arc::new(hook)));
}
#[must_use = "dropping the guard unregisters the hook; bind it for as long \
as the hook should fire, or use `on_reload` for a permanent one"]
pub fn on_reload_scoped(
&'static self,
hook: impl Fn(&Arc<T>, &Arc<T>) + Send + Sync + 'static,
) -> HookGuard<T> {
HookGuard {
token: self.register(Callback::Pair(Arc::new(hook))),
cell: GuardCell::Static(self),
}
}
#[must_use = "dropping the guard unregisters the hook; bind it for as long \
as the hook should fire, or use `on_reload_with` for a \
permanent one"]
pub fn on_reload_with_scoped(
&'static self,
hook: impl Fn(&ReloadEvent<T>) + Send + Sync + 'static,
) -> HookGuard<T> {
HookGuard {
token: self.register(Callback::Event(Arc::new(hook))),
cell: GuardCell::Static(self),
}
}
pub(crate) fn on_reload_scoped_shared(
cell: &Arc<Self>,
hook: impl Fn(&Arc<T>, &Arc<T>) + Send + Sync + 'static,
) -> HookGuard<T> {
HookGuard {
token: cell.register(Callback::Pair(Arc::new(hook))),
cell: GuardCell::Shared(Arc::clone(cell)),
}
}
pub(crate) fn on_reload_with_scoped_shared(
cell: &Arc<Self>,
hook: impl Fn(&ReloadEvent<T>) + Send + Sync + 'static,
) -> HookGuard<T> {
HookGuard {
token: cell.register(Callback::Event(Arc::new(hook))),
cell: GuardCell::Shared(Arc::clone(cell)),
}
}
fn register(&self, callback: Callback<T>) -> u64 {
let token = self
.next_token
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
self.hooks
.get_or_init(|| ArcSwap::from_pointee(Vec::new()))
.rcu(|current| {
let mut next = Vec::with_capacity(current.len() + 1);
next.extend(current.iter().cloned());
next.push(Registered {
token,
callback: callback.clone(),
});
next
});
token
}
fn unregister(&self, token: u64) {
let Some(hooks) = self.hooks.get() else {
return;
};
hooks.rcu(|current| {
current
.iter()
.filter(|registered| registered.token != token)
.cloned()
.collect::<Vec<_>>()
});
}
fn dispatch(
&self,
previous: Option<Arc<T>>,
current: &Arc<T>,
reason: ReloadReason,
meta: SnapshotMeta,
) {
let Some(hooks) = self.hooks.get() else {
return;
};
let hooks = hooks.load();
if hooks.is_empty() {
return;
}
let event = ReloadEvent::new(previous, Arc::clone(current), reason, meta);
for registered in hooks.iter() {
let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
match ®istered.callback {
Callback::Pair(hook) => {
if let Some(previous) = &event.previous {
hook(previous, &event.current);
}
}
Callback::Event(hook) => hook(&event),
}
}));
if outcome.is_err() {
crate::log::warning!(
"a reload hook panicked; it stays registered and the \
remaining hooks still run"
);
}
}
}
pub fn load(&self) -> Option<Arc<T>> {
self.inner.get().map(ArcSwap::load_full)
}
#[must_use]
pub fn generation(&self) -> u64 {
self.meta.load().as_ref().map_or(0, |meta| meta.generation)
}
#[must_use]
pub fn meta(&self) -> Option<SnapshotMeta> {
self.meta.load().as_deref().copied()
}
pub fn get_or_panic(&self, type_name: &str) -> Arc<T> {
self.load().unwrap_or_else(|| {
panic!(
"{type_name} has no snapshot installed; configure and install \
one first: `{type_name}::builder(\"..\")...init()?`"
)
})
}
#[cfg(feature = "async")]
#[cfg_attr(docsrs, doc(cfg(feature = "async")))]
pub fn changes(&'static self) -> crate::Changes<T>
where
T: Send + Sync,
{
crate::Changes::new(self)
}
#[cfg(feature = "async")]
pub(crate) fn notify(&self) -> &crate::asynchronous::Notify {
&self.notify
}
}
#[must_use = "dropping the guard unregisters the hook immediately; bind it for \
as long as the hook should fire, or register a permanent hook \
with `on_reload`"]
pub struct HookGuard<T: 'static> {
cell: GuardCell<T>,
token: u64,
}
enum GuardCell<T: 'static> {
Static(&'static ConfigCell<T>),
Shared(Arc<ConfigCell<T>>),
}
impl<T> Drop for HookGuard<T> {
fn drop(&mut self) {
match &self.cell {
GuardCell::Static(cell) => cell.unregister(self.token),
GuardCell::Shared(cell) => cell.unregister(self.token),
}
}
}
impl<T> std::fmt::Debug for HookGuard<T> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("HookGuard")
.field("token", &self.token)
.finish_non_exhaustive()
}
}
impl<T> Default for ConfigCell<T> {
fn default() -> Self {
Self::new()
}
}
impl<T: std::fmt::Debug> std::fmt::Debug for ConfigCell<T> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self.load() {
Some(value) => f.debug_tuple("ConfigCell").field(&value).finish(),
None => f.write_str("ConfigCell(uninitialized)"),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Mutex;
use std::thread;
#[test]
fn a_fresh_cell_is_empty() {
let cell = ConfigCell::<u16>::new();
assert!(cell.load().is_none());
}
#[test]
fn a_reader_keeps_the_generation_it_took() {
let cell = ConfigCell::new();
cell.store(String::from("first"));
let held = cell.load().unwrap();
cell.store(String::from("second"));
assert_eq!(*held, "first");
assert_eq!(*cell.load().unwrap(), "second");
}
#[test]
fn concurrent_first_writes_do_not_lose_the_cell() {
let cell: &'static ConfigCell<usize> = Box::leak(Box::new(ConfigCell::new()));
let writers: Vec<_> = (0..8)
.map(|value| thread::spawn(move || cell.store(value)))
.collect();
for writer in writers {
writer.join().unwrap();
}
let final_value = *cell.load().expect("some writer must have won");
assert!(final_value < 8);
}
#[test]
fn the_first_store_is_an_initialization_not_a_reload() {
let seen = Arc::new(Mutex::new(Vec::new()));
let cell = ConfigCell::new();
let recorder = Arc::clone(&seen);
cell.on_reload(move |previous, current| {
recorder.lock().unwrap().push((**previous, **current));
});
cell.store(1u16);
assert!(
seen.lock().unwrap().is_empty(),
"there is nothing to compare the first snapshot against"
);
cell.store(2u16);
cell.store(3u16);
assert_eq!(*seen.lock().unwrap(), [(1, 2), (2, 3)]);
}
#[test]
fn every_registered_callback_runs() {
let count = Arc::new(Mutex::new(0usize));
let cell = ConfigCell::new();
for _ in 0..3 {
let counter = Arc::clone(&count);
cell.on_reload(move |_, _| *counter.lock().unwrap() += 1);
}
cell.store(1u16);
cell.store(2u16);
assert_eq!(*count.lock().unwrap(), 3);
}
#[test]
fn a_panicking_hook_silences_neither_the_rest_nor_the_next_reload() {
let count = Arc::new(Mutex::new(0usize));
let cell = ConfigCell::new();
cell.on_reload(|_, _| panic!("a bug in somebody's hook"));
{
let counter = Arc::clone(&count);
cell.on_reload(move |_, _| *counter.lock().unwrap() += 1);
}
cell.store(1u16);
cell.store(2u16);
cell.store(3u16);
assert_eq!(
*count.lock().unwrap(),
2,
"the hook after the panicking one must run on every reload"
);
}
#[test]
fn dropping_the_guard_unregisters_the_hook() {
let count = Arc::new(Mutex::new(0usize));
let cell: &'static ConfigCell<u16> = Box::leak(Box::new(ConfigCell::new()));
cell.store(1);
let guard = {
let counter = Arc::clone(&count);
cell.on_reload_scoped(move |_, _| *counter.lock().unwrap() += 1)
};
cell.store(2);
assert_eq!(*count.lock().unwrap(), 1);
drop(guard);
cell.store(3);
assert_eq!(
*count.lock().unwrap(),
1,
"an unregistered hook must not fire"
);
}
#[test]
fn store_with_hands_back_the_snapshot_it_installed() {
let cell = ConfigCell::new();
let first = cell.store_with(1u16, ReloadReason::Initial);
assert!(Arc::ptr_eq(&first, &cell.load().unwrap()));
let second = cell.store_with(2u16, ReloadReason::Manual);
assert_eq!(*first, 1, "the earlier install's snapshot is unmoved");
assert_eq!(*second, 2);
assert!(Arc::ptr_eq(&second, &cell.load().unwrap()));
}
#[test]
#[should_panic(expected = "`DbConfig::builder(")]
fn get_or_panic_points_at_the_builder() {
ConfigCell::<u16>::new().get_or_panic("DbConfig");
}
#[test]
fn the_event_form_sees_the_first_install_and_the_pair_form_does_not() {
let pairs = Arc::new(Mutex::new(0usize));
let events = Arc::new(Mutex::new(Vec::new()));
let cell = ConfigCell::new();
{
let counter = Arc::clone(&pairs);
cell.on_reload(move |_, _| *counter.lock().unwrap() += 1);
}
{
let recorder = Arc::clone(&events);
cell.on_reload_with(move |event| {
recorder
.lock()
.unwrap()
.push((event.previous.as_deref().copied(), *event.current));
});
}
cell.store(1u16);
assert_eq!(*pairs.lock().unwrap(), 0);
assert_eq!(*events.lock().unwrap(), [(None, 1)]);
cell.store(2u16);
assert_eq!(*pairs.lock().unwrap(), 1);
assert_eq!(*events.lock().unwrap(), [(None, 1), (Some(1), 2)]);
}
#[test]
fn an_event_carries_the_reason_and_the_generation_of_its_own_install() {
let seen = Arc::new(Mutex::new(Vec::new()));
let cell = ConfigCell::new();
{
let recorder = Arc::clone(&seen);
cell.on_reload_with(move |event| {
recorder
.lock()
.unwrap()
.push((event.reason.clone(), event.meta.generation));
});
}
cell.store_with(1u16, ReloadReason::Initial);
cell.store_with(2u16, ReloadReason::RemoteChanged);
cell.store(3u16);
assert_eq!(
*seen.lock().unwrap(),
[
(ReloadReason::Initial, 1),
(ReloadReason::RemoteChanged, 2),
(ReloadReason::Manual, 3),
]
);
}
#[test]
fn a_panicking_event_hook_leaves_the_rest_running() {
let count = Arc::new(Mutex::new(0usize));
let cell = ConfigCell::new();
cell.on_reload_with(|_| panic!("a bug in somebody's hook"));
{
let counter = Arc::clone(&count);
cell.on_reload_with(move |_| *counter.lock().unwrap() += 1);
}
cell.store(1u16);
cell.store(2u16);
assert_eq!(*count.lock().unwrap(), 2);
}
#[test]
fn a_scoped_event_hook_stops_when_its_guard_drops() {
let count = Arc::new(Mutex::new(0usize));
let cell: &'static ConfigCell<u16> = Box::leak(Box::new(ConfigCell::new()));
let guard = {
let counter = Arc::clone(&count);
cell.on_reload_with_scoped(move |_| *counter.lock().unwrap() += 1)
};
cell.store(1);
cell.store(2);
assert_eq!(*count.lock().unwrap(), 2, "the first install counts too");
drop(guard);
cell.store(3);
assert_eq!(*count.lock().unwrap(), 2);
}
#[test]
fn failures_count_up_and_an_install_resets_the_streak() {
let cell = ConfigCell::<u16>::new();
assert_eq!(cell.status().consecutive_failures, 0);
assert!(cell.status().is_healthy());
assert!(cell.status().last_failure.is_none());
assert!(cell.status().last_reason.is_none());
for expected in 1..=3 {
cell.record_failure(&Error::new(
crate::ErrorKind::Parse,
"unexpected end of input",
));
assert_eq!(cell.status().consecutive_failures, expected);
}
assert!(!cell.status().is_healthy());
assert_eq!(
cell.status().last_failure.unwrap().kind,
crate::ErrorKind::Parse
);
cell.store_with(1, ReloadReason::Recovered);
let status = cell.status();
assert_eq!(status.consecutive_failures, 0);
assert!(status.is_healthy());
assert_eq!(status.generation, 1);
assert_eq!(status.last_reason, Some(ReloadReason::Recovered));
assert!(
status.last_failure.is_some(),
"the streak resets; the record of what went wrong does not"
);
assert!(status.loaded_at.is_some());
}
}