dynamic_config/remote/

watch.rs

//! Stopping a blocking watch.
//!
//! A watch nobody owns is a leak nobody asked for, so the handle stops the
//! loop when it is dropped and `RemoteWatch::detach` is how a caller says
//! *this one really should run forever*. Only blocking loops need this: an
//! async watch is a future, and dropping it is stopping it.

use std::sync::{Arc, Weak};
use std::time::Duration;

use crate::sync::atomic::{AtomicBool, Ordering};

/// A running blocking watch, from the caller's side.
///
/// Dropping it stops the loop — the same contract the file watcher's
/// `WatchHandle` has, for the same reason: a watch nobody owns is a leak nobody
/// asked for. [`detach`](Self::detach) is the way to say *this one really should
/// run forever*.
///
/// Only blocking loops need this. An async watch is a future: drop it and it is
/// cancelled, on any executor.
///
/// ```no_run
/// # use dynamic_config::RemoteWatch;
/// # struct Consul;
/// # impl Consul {
/// #     fn watch(&self, _: dynamic_config::Watching, _: fn(dynamic_config::Fetched) -> Result<(), dynamic_config::Error>) -> Result<(), dynamic_config::Error> { Ok(()) }
/// # }
/// # fn example(consul: Consul) {
/// # fn apply(_: dynamic_config::Fetched) -> Result<(), dynamic_config::Error> { Ok(()) }
/// let watch = RemoteWatch::new();
/// let watching = watch.watching();
///
/// std::thread::spawn(move || consul.watch(watching, apply));
///
/// // ... and later, or by dropping `watch`:
/// watch.stop();
/// # }
/// ```
#[must_use = "dropping the handle stops the watch; bind it, or call `.detach()` \
              to watch for the rest of the process"]
#[derive(Debug)]
pub struct RemoteWatch {
    running: Arc<AtomicBool>,
}

impl RemoteWatch {
    /// A handle for a watch that has not been handed to a loop yet.
    pub fn new() -> Self {
        Self {
            running: Arc::new(AtomicBool::new(true)),
        }
    }

    /// The loop's half of this handle.
    ///
    /// Hand it to the watch; keep the `RemoteWatch` yourself.
    #[must_use]
    pub fn watching(&self) -> Watching {
        Watching {
            running: Arc::downgrade(&self.running),
        }
    }

    /// Stops the loop at its next check.
    ///
    /// *At its next check* is the whole caveat, and it is not small: a loop
    /// parked in a blocking query does not return until the store answers or
    /// the wait expires, so the store's wait time is the worst-case delay. Each
    /// companion crate documents its own.
    pub fn stop(&self) {
        self.running.store(false, Ordering::Release);
    }

    /// Whether the loop has been told to stop.
    #[must_use]
    pub fn is_stopped(&self) -> bool {
        !self.running.load(Ordering::Acquire)
    }

    /// Watches for the remainder of the process.
    ///
    /// Leaks the handle on purpose, exactly as the file watcher's
    /// `WatchHandle::detach` does: a watch that must never stop has no owner to
    /// hold it, and pretending otherwise is how it ends up stopped at the end of
    /// `main`'s first statement.
    pub fn detach(self) {
        std::mem::forget(self);
    }

    /// Watches until `shutdown` completes, then stops.
    ///
    /// The shape a server wants: a watch is not something to remember to
    /// stop, it is something that ends when the process is winding down.
    ///
    /// ```no_run
    /// # async fn shutdown_signal() {}
    /// # async fn run(watch: dynamic_config::RemoteWatch) {
    /// watch.run_until(shutdown_signal()).await;
    /// # }
    /// ```
    ///
    /// No runtime is imposed: the future is driven by whichever executor is
    /// already running the caller, and this adds no second cancellation
    /// mechanism — it sets the same flag [`stop`](Self::stop) does, which is
    /// what every loop already checks. The same caveat therefore applies: a
    /// loop parked in a blocking query returns when the store answers or the
    /// wait expires, so the store's wait time is the worst-case delay.
    ///
    /// Takes `self` so the handle cannot outlive the watch it was ending.
    pub async fn run_until(self, shutdown: impl core::future::Future<Output = ()>) {
        shutdown.await;

        // Through `drop` rather than `stop` — they do the same thing, and
        // going through the destructor keeps one path for "this watch is
        // over" instead of two that could drift.
        drop(self);
    }
}

impl Default for RemoteWatch {
    fn default() -> Self {
        Self::new()
    }
}

impl Drop for RemoteWatch {
    fn drop(&mut self) {
        self.stop();
    }
}

/// The loop's half of a [`RemoteWatch`].
///
/// A `Weak`, so a handle that is dropped without anyone remembering to call
/// `stop` still ends the loop: the upgrade fails and
/// [`keep_going`](Self::keep_going) answers `false`.
#[derive(Debug, Clone)]
pub struct Watching {
    running: Weak<AtomicBool>,
}

impl Watching {
    /// Whether the loop should go round again.
    ///
    /// `false` once the caller called [`RemoteWatch::stop`] or dropped the
    /// handle. Check it before every request, not only after one: a loop that
    /// checks only on the way out issues one more query than it was asked to.
    #[must_use]
    pub fn keep_going(&self) -> bool {
        self.running
            .upgrade()
            .is_some_and(|running| running.load(Ordering::Acquire))
    }

    /// Sleeps for `total`, waking early if the watch is stopped.
    ///
    /// The polling loop every blocking store crate writes: sleep a slice,
    /// check [`keep_going`](Self::keep_going), repeat — so a stopped watch
    /// ends within a quarter second instead of at the end of its interval.
    /// Here once, rather than once per store crate.
    pub fn sleep_for(&self, total: Duration) {
        const SLICE: Duration = Duration::from_millis(250);

        let mut slept = Duration::ZERO;

        while slept < total && self.keep_going() {
            std::thread::sleep(SLICE.min(total - slept));
            slept += SLICE;
        }
    }

    /// Sleeps for `total`, jittered, waking early if the watch is stopped.
    ///
    /// A fleet started by one rollout polls in lockstep otherwise: fifty
    /// replicas with a thirty-second interval become fifty simultaneous
    /// requests every thirty seconds, and the store sees a spike rather than
    /// a trickle. The spread is drawn once per loop from the clock, so two
    /// processes on one machine differ and a restart does not land back in
    /// the same phase.
    pub fn sleep_jittered(&self, total: Duration, pace: &mut Pace) {
        self.sleep_for(pace.spread(total));
    }

    /// A token for a watch that should never stop.
    ///
    /// For a loop the caller genuinely wants to outlive everything, so there is
    /// no handle to hold. Prefer [`RemoteWatch::detach`], which says the same
    /// thing at the point where somebody decided it.
    #[must_use]
    pub fn forever() -> Self {
        // A `Weak` that can never upgrade would stop the loop immediately, so
        // this leaks one live flag — one allocation, once, for the life of the
        // process.
        let running = Box::leak(Box::new(Arc::new(AtomicBool::new(true))));

        Self {
            running: Arc::downgrade(running),
        }
    }
}

/// The waits a watch loop makes.
///
/// Two of them, and a loop needs both: the pause between healthy rounds,
/// spread so a fleet does not poll in lockstep, and the growing pause after
/// a failure, so a store that is down is not hammered by everything that
/// depends on it. A loop that sleeps its plain interval after an error is
/// the shape that turns one outage into two.
///
/// ```
/// # use std::time::Duration;
/// # use dynamic_config::{Pace, Watching};
/// # fn example(watching: &Watching) {
/// let mut pace = Pace::new(Duration::from_secs(30));
///
/// while watching.keep_going() {
///     match fetch() {
///         Ok(()) => pace.succeeded(),
///         Err(()) => pace.failed(),
///     }
///
///     pace.wait(watching);
/// }
/// # }
/// # fn fetch() -> Result<(), ()> { Ok(()) }
/// ```
#[derive(Debug, Clone)]
pub struct Pace {
    interval: Duration,
    ceiling: Duration,
    failures: u32,
    entropy: u64,
}

impl Pace {
    /// The default ceiling a backoff grows to.
    const CEILING: Duration = Duration::from_secs(300);

    /// Rounds `interval` apart when things are going well.
    #[must_use]
    pub fn new(interval: Duration) -> Self {
        Self {
            interval,
            ceiling: Self::CEILING,
            failures: 0,
            entropy: seed(),
        }
    }

    /// Caps how far a backoff grows. Five minutes by default.
    #[must_use]
    pub fn with_ceiling(mut self, ceiling: Duration) -> Self {
        self.ceiling = ceiling;
        self
    }

    /// A round went well: the next wait is the plain interval again.
    pub fn succeeded(&mut self) {
        self.failures = 0;
    }

    /// A round failed: the next wait is longer than the last one.
    pub fn failed(&mut self) {
        self.failures = self.failures.saturating_add(1);
    }

    /// How long to wait before the next round, jittered.
    ///
    /// The interval while things work; doubling from it after each failure,
    /// up to the ceiling.
    ///
    /// # Two kinds of jitter, for two different problems
    ///
    /// A **healthy** wait is spread by a quarter either way. The interval is
    /// a promise about how often a store is read, and a fleet only needs its
    /// members nudged out of lockstep to stop arriving together.
    ///
    /// A wait **after a failure** is drawn from the whole range instead —
    /// anywhere between nothing and the full backoff. That is the shape that
    /// actually decorrelates a fleet, and a recovering store is the one
    /// moment it matters: a thousand agents that failed at the same instant
    /// have been counting the same doubling ever since, and a band a quarter
    /// wide would land them back on the store in a clump.
    ///
    /// The two are not interchangeable. Drawing a healthy wait from the
    /// whole range would halve its mean, which is not a jitter policy — it
    /// is a different interval, and it doubles the read rate of every store
    /// in the fleet.
    #[must_use]
    pub fn next_wait(&mut self) -> Duration {
        if self.failures == 0 {
            let interval = self.interval;

            return self.spread(interval);
        }

        // Doubling, but never past the ceiling and never past what a
        // `Duration` can hold — a loop that has failed for a week must
        // not overflow its way back down to no wait at all.
        let factor = 1u32.checked_shl(self.failures.min(16)).unwrap_or(u32::MAX);

        let backoff = self
            .interval
            .checked_mul(factor)
            .unwrap_or(self.ceiling)
            .min(self.ceiling);

        self.up_to(backoff)
    }

    /// Waits for [`next_wait`](Self::next_wait), waking early if the watch
    /// is stopped.
    pub fn wait(&mut self, watching: &Watching) {
        let wait = self.next_wait();

        watching.sleep_for(wait);
    }

    /// `base`, moved by up to a quarter of itself in either direction.
    ///
    /// Deterministic given the seed, so a test can pin it; drawn from the
    /// clock at construction, so two processes do not share a phase.
    #[must_use]
    pub fn spread(&mut self, base: Duration) -> Duration {
        let draw = self.draw();

        let quarter = base / 4;
        let offset = quarter
            .checked_mul(u32::try_from(draw >> 33 & 0xFF).unwrap_or(0))
            .unwrap_or(quarter)
            / 255;

        if draw & 1 == 0 {
            base.saturating_add(offset)
        } else {
            base.saturating_sub(offset)
        }
    }

    /// Anywhere from nothing up to `base` — full jitter.
    ///
    /// The backoff's own shape. Two processes that failed together stop
    /// being correlated after one draw, rather than staying a quarter-width
    /// band apart for as long as the outage lasts.
    fn up_to(&mut self, base: Duration) -> Duration {
        let draw = self.draw();

        // Eight bits of the word, scaled across the range. The same
        // arithmetic `spread` uses, so the two cannot drift apart in how
        // they handle a `Duration` too large to multiply.
        base.checked_mul(u32::try_from(draw >> 33 & 0xFF).unwrap_or(0))
            .unwrap_or(base)
            / 255
    }

    /// One step of the generator, and the only place it advances.
    ///
    /// An LCG rather than a dependency: the numbers only have to be
    /// uncorrelated between processes, which is a far weaker ask than
    /// anything a random number generator is built for.
    fn draw(&mut self) -> u64 {
        self.entropy = self
            .entropy
            .wrapping_mul(6_364_136_223_846_793_005)
            .wrapping_add(1_442_695_040_888_963_407);

        self.entropy
    }
}

/// Something different per process, without a dependency.
fn seed() -> u64 {
    let since = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map_or(0, |elapsed| elapsed.as_nanos() as u64);

    since ^ u64::from(std::process::id()).wrapping_mul(0x9E37_79B9_7F4A_7C15)
}