siderite_core/server/
readiness.rs

1//! Application readiness for probes during startup and resource teardown.
2
3use crate::{ApiError, FromRequestParts};
4use http::request::Parts;
5use std::sync::Arc;
6use std::sync::atomic::{AtomicU8, Ordering};
7use tokio::sync::Notify;
8
9/// Observable phase of the root application lifecycle.
10#[derive(Debug, Clone, Copy, PartialEq, Eq)]
11pub enum ServerPhase {
12    /// Initialization has not completed.
13    Starting,
14    /// Startup has completed and admission is available.
15    Ready,
16    /// Transport/task admission is closing and resources are draining.
17    Draining,
18    /// Cleanup attempts have completed, successfully or with an error.
19    Stopped,
20}
21
22/// Extractor and shareable handle for application readiness probes.
23///
24/// In-process services built without running lifespan remain Starting.
25/// Readiness does not assess database health or prove a load balancer has
26/// removed this server; deployments must configure their probes and grace.
27#[derive(Debug, Clone)]
28pub struct Readiness(Arc<State>);
29
30#[derive(Debug, Default)]
31struct State {
32    phase: AtomicU8,
33    /// Woken when the phase reaches Draining or Stopped.
34    stopping: Notify,
35}
36
37impl Readiness {
38    pub(crate) fn new() -> Self {
39        Self(Arc::default())
40    }
41
42    /// Read the current root application lifecycle phase.
43    ///
44    /// Returns:
45    ///     A snapshot which can change immediately after this call.
46    pub fn phase(&self) -> ServerPhase {
47        match self.0.phase.load(Ordering::Acquire) {
48            0 => ServerPhase::Starting,
49            1 => ServerPhase::Ready,
50            2 => ServerPhase::Draining,
51            _ => ServerPhase::Stopped,
52        }
53    }
54
55    /// Return whether startup completed and shutdown has not begun.
56    ///
57    /// Returns:
58    ///     True only during the Ready phase.
59    pub fn is_ready(&self) -> bool {
60        self.phase() == ServerPhase::Ready
61    }
62
63    /// Wait until shutdown has begun (Draining or Stopped); returns at
64    /// once when it already has. Long-lived responses such as event
65    /// streams end on it so a graceful shutdown need not time out.
66    pub async fn stopping(&self) {
67        loop {
68            let woken = self.0.stopping.notified();
69            tokio::pin!(woken);
70            // Registered before the check, so no change is missed.
71            woken.as_mut().enable();
72            let phase = self.phase();
73            if matches!(phase, ServerPhase::Draining | ServerPhase::Stopped) {
74                return;
75            }
76            woken.await;
77        }
78    }
79
80    pub(crate) fn ready(&self) {
81        self.0.phase.store(1, Ordering::Release);
82    }
83
84    pub(crate) fn draining(&self) {
85        // Repeated cleanup must not move Stopped back into Draining.
86        let _ = self
87            .0
88            .phase
89            .try_update(Ordering::AcqRel, Ordering::Acquire, |phase| {
90                (phase < 2).then_some(2)
91            });
92        self.0.stopping.notify_waiters();
93    }
94
95    pub(crate) fn stopped(&self) {
96        self.0.phase.store(3, Ordering::Release);
97        self.0.stopping.notify_waiters();
98    }
99}
100
101impl FromRequestParts for Readiness {
102    async fn from_request_parts(parts: &mut Parts) -> Result<Self, ApiError> {
103        parts
104            .extensions
105            .get::<Self>()
106            .cloned()
107            .ok_or_else(|| ApiError::internal("readiness is not installed"))
108    }
109}
siderite_core/server/readiness.rs

siderite_core/server/
readiness.rs
