appcore_api/http/
reload.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: reload.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/08/30 05:00:00 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/08/30 05:00:00 by dnettoRaw
8//      ###########      S: 1.0.2-rc
9// =============================================================================
10
11//! Atomic HTTP routing-generation reload with bounded drain and rollback.
12
13use super::reload_generation::{HttpRoutingGenerationsSnapshot, RoutingGeneration, RoutingTable};
14use super::RuntimeHttpHost;
15use axum::body::Body;
16use axum::extract::{Request, State};
17use axum::http::{Response, StatusCode};
18use axum::routing::any;
19use axum::Router;
20use std::fmt;
21use std::io;
22use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
23use std::sync::Arc;
24use std::time::{Duration, Instant};
25use tower::ServiceExt;
26
27const ROUTING_RETRY_DELAY: Duration = Duration::from_millis(1);
28const MAX_RELOAD_TIMEOUT: Duration = Duration::from_secs(60);
29
30/// Reload phase associated with a controlled failure.
31#[derive(Debug, Clone, Copy, PartialEq, Eq)]
32pub enum HttpReloadPhase {
33    /// Candidate validation and health before activation.
34    Prepare,
35    /// Atomic routing activation and health confirmation.
36    Switch,
37    /// Bounded completion of requests admitted by the old generation.
38    Drain,
39    /// Restoration and drain after a failed activation.
40    Rollback,
41}
42
43/// Bounded health and drain policy for one routing reload.
44#[derive(Debug, Clone, Copy, PartialEq, Eq)]
45pub struct HttpReloadPolicy {
46    health_timeout: Duration,
47    drain_timeout: Duration,
48}
49
50impl HttpReloadPolicy {
51    /// Creates a policy with non-zero health and drain deadlines.
52    pub fn new(
53        health_timeout: Duration,
54        drain_timeout: Duration,
55    ) -> Result<Self, RuntimeHttpReloadError> {
56        if health_timeout.is_zero()
57            || drain_timeout.is_zero()
58            || health_timeout > MAX_RELOAD_TIMEOUT
59            || drain_timeout > MAX_RELOAD_TIMEOUT
60        {
61            return Err(RuntimeHttpReloadError::InvalidPolicy);
62        }
63        Ok(Self {
64            health_timeout,
65            drain_timeout,
66        })
67    }
68
69    /// Returns the candidate health deadline.
70    pub fn health_timeout(self) -> Duration {
71        self.health_timeout
72    }
73
74    /// Returns the old-generation drain deadline.
75    pub fn drain_timeout(self) -> Duration {
76        self.drain_timeout
77    }
78}
79
80/// Payload-free state and counters for a reloadable HTTP host.
81#[derive(Debug, Clone, Copy, PartialEq, Eq)]
82pub struct HttpReloadSnapshot {
83    /// Generation currently selected for new requests.
84    pub active_generation: u64,
85    /// Requests currently executing on the selected generation.
86    pub active_inflight: usize,
87    /// Whether one prepare/switch/drain transaction is active.
88    pub reload_in_progress: bool,
89    /// Reloads that switched and drained successfully.
90    pub successful_reloads: u64,
91    /// Failed reload attempts, including controlled rollbacks.
92    pub failed_reloads: u64,
93    /// Switches restored to the prior generation.
94    pub rollbacks: u64,
95}
96
97/// Controlled, redacted reload failure.
98#[derive(Debug, Clone, Copy, PartialEq, Eq)]
99pub enum RuntimeHttpReloadError {
100    /// Reload requires an enabled listener.
101    ListenerDisabled,
102    /// A candidate attempted to change the listener address in-place.
103    ListenerAddressChanged,
104    /// Generation identifiers must increase monotonically.
105    StaleGeneration,
106    /// Another reload transaction already owns the coordinator.
107    ReloadInProgress,
108    /// A failed or cancelled generation still owns in-flight requests.
109    RetiringGenerationBusy,
110    /// Health or drain deadlines must be non-zero.
111    InvalidPolicy,
112    /// The candidate or active generation failed its health gate.
113    HealthGateFailed(HttpReloadPhase),
114    /// The outgoing generation did not drain before rollback.
115    DrainTimedOut,
116    /// The failed generation remained active past the rollback drain deadline.
117    RollbackDrainTimedOut,
118}
119
120impl fmt::Display for RuntimeHttpReloadError {
121    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
122        match self {
123            Self::ListenerDisabled => formatter.write_str("HTTP reload listener is disabled"),
124            Self::ListenerAddressChanged => {
125                formatter.write_str("HTTP reload requires a prepared listener generation")
126            }
127            Self::StaleGeneration => formatter.write_str("HTTP routing generation must increase"),
128            Self::ReloadInProgress => formatter.write_str("HTTP reload is already in progress"),
129            Self::RetiringGenerationBusy => {
130                formatter.write_str("HTTP routing generation is still draining")
131            }
132            Self::InvalidPolicy => formatter.write_str("HTTP reload policy is invalid"),
133            Self::HealthGateFailed(phase) => {
134                write!(formatter, "HTTP reload health gate failed during {phase:?}")
135            }
136            Self::DrainTimedOut => formatter.write_str("HTTP routing generation drain timed out"),
137            Self::RollbackDrainTimedOut => {
138                formatter.write_str("HTTP rollback generation drain timed out")
139            }
140        }
141    }
142}
143
144impl std::error::Error for RuntimeHttpReloadError {}
145
146/// Prepared, health-gated candidate that can be consumed by one reload.
147pub struct PreparedRuntimeHttpGeneration {
148    generation: Arc<RoutingGeneration>,
149}
150
151impl fmt::Debug for PreparedRuntimeHttpGeneration {
152    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
153        formatter
154            .debug_struct("PreparedRuntimeHttpGeneration")
155            .field("generation", &self.generation.id())
156            .finish_non_exhaustive()
157    }
158}
159
160/// HTTP host whose stable listener dispatches through one atomic generation.
161pub struct ReloadableRuntimeHttpHost {
162    config: super::HttpApiConfig,
163    routing: Arc<RoutingTable>,
164    reload_in_progress: AtomicBool,
165    successful_reloads: AtomicU64,
166    failed_reloads: AtomicU64,
167    rollbacks: AtomicU64,
168}
169
170impl ReloadableRuntimeHttpHost {
171    /// Creates a reloadable host from an already composed initial host.
172    pub fn new(
173        initial_generation: u64,
174        host: RuntimeHttpHost,
175    ) -> Result<Self, RuntimeHttpReloadError> {
176        if !host.config().enabled {
177            return Err(RuntimeHttpReloadError::ListenerDisabled);
178        }
179        if initial_generation == 0 {
180            return Err(RuntimeHttpReloadError::StaleGeneration);
181        }
182        let config = host.config().clone();
183        Ok(Self {
184            config,
185            routing: Arc::new(RoutingTable::new(initial_generation, host.router())),
186            reload_in_progress: AtomicBool::new(false),
187            successful_reloads: AtomicU64::new(0),
188            failed_reloads: AtomicU64::new(0),
189            rollbacks: AtomicU64::new(0),
190        })
191    }
192
193    /// Validates and owns a candidate without changing live routing.
194    pub fn prepare(
195        &self,
196        generation: u64,
197        host: RuntimeHttpHost,
198    ) -> Result<PreparedRuntimeHttpGeneration, RuntimeHttpReloadError> {
199        let active = self.routing.active();
200        if generation <= active.id() {
201            return Err(RuntimeHttpReloadError::StaleGeneration);
202        }
203        if !host.config().enabled {
204            return Err(RuntimeHttpReloadError::ListenerDisabled);
205        }
206        if host.config().host != self.config.host || host.config().port != self.config.port {
207            return Err(RuntimeHttpReloadError::ListenerAddressChanged);
208        }
209        Ok(PreparedRuntimeHttpGeneration {
210            generation: self.routing.generation(generation, host.router()),
211        })
212    }
213
214    #[cfg(test)]
215    pub(super) fn new_for_test(
216        initial_generation: u64,
217        config: super::HttpApiConfig,
218        router: Router,
219    ) -> Self {
220        Self {
221            config,
222            routing: Arc::new(RoutingTable::new(initial_generation, router)),
223            reload_in_progress: AtomicBool::new(false),
224            successful_reloads: AtomicU64::new(0),
225            failed_reloads: AtomicU64::new(0),
226            rollbacks: AtomicU64::new(0),
227        }
228    }
229
230    #[cfg(test)]
231    pub(super) fn prepare_router_for_test(
232        &self,
233        generation: u64,
234        router: Router,
235    ) -> PreparedRuntimeHttpGeneration {
236        PreparedRuntimeHttpGeneration {
237            generation: self.routing.generation(generation, router),
238        }
239    }
240
241    /// Returns a router that always dispatches through the active generation.
242    pub fn router(&self) -> Router {
243        dynamic_router(Arc::clone(&self.routing))
244    }
245
246    /// Runs the stable listener until cooperative shutdown is requested.
247    pub fn run_until_shutdown(&self, shutdown: Arc<AtomicBool>) -> io::Result<()> {
248        let address = format!("{}:{}", self.config.host, self.config.port);
249        let runtime = super::build_http_runtime()?;
250        runtime.block_on(async move {
251            let listener = tokio::net::TcpListener::bind(address).await?;
252            serve_listener(listener, self.router(), shutdown).await
253        })
254    }
255
256    /// Runs on a listener that the composition root already bound and checked.
257    pub fn run_on_listener_until_shutdown(
258        &self,
259        listener: std::net::TcpListener,
260        shutdown: Arc<AtomicBool>,
261    ) -> io::Result<()> {
262        listener.set_nonblocking(true)?;
263        let router = self.router();
264        let runtime = super::build_http_runtime()?;
265        runtime.block_on(async move {
266            let listener = tokio::net::TcpListener::from_std(listener)?;
267            serve_listener(listener, router, shutdown).await
268        })
269    }
270
271    /// Health-checks, atomically activates, and drains one prepared generation.
272    pub async fn reload(
273        &self,
274        prepared: PreparedRuntimeHttpGeneration,
275        policy: HttpReloadPolicy,
276    ) -> Result<(), RuntimeHttpReloadError> {
277        let mut guard = match ReloadGuard::acquire(
278            &self.reload_in_progress,
279            &self.routing,
280            &self.failed_reloads,
281            &self.rollbacks,
282        ) {
283            Ok(guard) => guard,
284            Err(error) => return self.fail(error),
285        };
286        self.routing.release_drained_retiring();
287        if self.routing.generations_snapshot().retiring.is_some() {
288            return self.fail(RuntimeHttpReloadError::RetiringGenerationBusy);
289        }
290        let previous = self.routing.active();
291        if prepared.generation.id() <= previous.id() {
292            return self.fail(RuntimeHttpReloadError::StaleGeneration);
293        }
294        if !probe_health(prepared.generation.router(), policy.health_timeout).await {
295            return self.fail(RuntimeHttpReloadError::HealthGateFailed(
296                HttpReloadPhase::Prepare,
297            ));
298        }
299
300        previous.stop_accepting();
301        if !self.routing.retire(Arc::clone(&previous)) {
302            previous.start_accepting();
303            return self.fail(RuntimeHttpReloadError::RetiringGenerationBusy);
304        }
305        guard.arm(Arc::clone(&previous), Arc::clone(&prepared.generation));
306        self.routing.activate(Arc::clone(&prepared.generation));
307        if !probe_health(prepared.generation.router(), policy.health_timeout).await {
308            guard.disarm();
309            return self
310                .rollback(previous, prepared.generation, policy)
311                .await
312                .and(Err(RuntimeHttpReloadError::HealthGateFailed(
313                    HttpReloadPhase::Switch,
314                )));
315        }
316        if !wait_for_drain(&previous, policy.drain_timeout).await {
317            guard.disarm();
318            return self
319                .rollback(previous, prepared.generation, policy)
320                .await
321                .and(Err(RuntimeHttpReloadError::DrainTimedOut));
322        }
323        self.routing.release_retiring(previous.id());
324        guard.disarm();
325        increment(&self.successful_reloads);
326        Ok(())
327    }
328
329    /// Returns the active generation and bounded operational counters.
330    pub fn snapshot(&self) -> HttpReloadSnapshot {
331        let active = self.routing.active();
332        HttpReloadSnapshot {
333            active_generation: active.id(),
334            active_inflight: active.inflight(),
335            reload_in_progress: self.reload_in_progress.load(Ordering::Acquire),
336            successful_reloads: self.successful_reloads.load(Ordering::Relaxed),
337            failed_reloads: self.failed_reloads.load(Ordering::Relaxed),
338            rollbacks: self.rollbacks.load(Ordering::Relaxed),
339        }
340    }
341
342    /// Returns the active and optional retiring generation without request data.
343    pub fn generation_snapshot(&self) -> HttpRoutingGenerationsSnapshot {
344        self.routing.generations_snapshot()
345    }
346
347    async fn rollback(
348        &self,
349        previous: Arc<RoutingGeneration>,
350        failed: Arc<RoutingGeneration>,
351        policy: HttpReloadPolicy,
352    ) -> Result<(), RuntimeHttpReloadError> {
353        failed.stop_accepting();
354        previous.start_accepting();
355        self.routing.activate(Arc::clone(&previous));
356        self.routing.release_retiring(previous.id());
357        if !self.routing.retire(Arc::clone(&failed)) {
358            return Err(RuntimeHttpReloadError::RetiringGenerationBusy);
359        }
360        increment(&self.rollbacks);
361        increment(&self.failed_reloads);
362        if wait_for_drain(&failed, policy.drain_timeout).await {
363            self.routing.release_retiring(failed.id());
364            Ok(())
365        } else {
366            Err(RuntimeHttpReloadError::RollbackDrainTimedOut)
367        }
368    }
369
370    fn fail<T>(&self, error: RuntimeHttpReloadError) -> Result<T, RuntimeHttpReloadError> {
371        increment(&self.failed_reloads);
372        Err(error)
373    }
374}
375
376struct ReloadGuard<'a> {
377    flag: &'a AtomicBool,
378    routing: &'a RoutingTable,
379    failed_reloads: &'a AtomicU64,
380    rollbacks: &'a AtomicU64,
381    rollback: Option<(Arc<RoutingGeneration>, Arc<RoutingGeneration>)>,
382}
383
384impl<'a> ReloadGuard<'a> {
385    fn acquire(
386        flag: &'a AtomicBool,
387        routing: &'a RoutingTable,
388        failed_reloads: &'a AtomicU64,
389        rollbacks: &'a AtomicU64,
390    ) -> Result<Self, RuntimeHttpReloadError> {
391        flag.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
392            .map_err(|_| RuntimeHttpReloadError::ReloadInProgress)?;
393        Ok(Self {
394            flag,
395            routing,
396            failed_reloads,
397            rollbacks,
398            rollback: None,
399        })
400    }
401
402    fn arm(&mut self, previous: Arc<RoutingGeneration>, candidate: Arc<RoutingGeneration>) {
403        self.rollback = Some((previous, candidate));
404    }
405
406    fn disarm(&mut self) {
407        self.rollback = None;
408    }
409}
410
411impl Drop for ReloadGuard<'_> {
412    fn drop(&mut self) {
413        if let Some((previous, failed)) = self.rollback.take() {
414            failed.stop_accepting();
415            previous.start_accepting();
416            self.routing.activate(Arc::clone(&previous));
417            self.routing.release_retiring(previous.id());
418            if self.routing.retire(failed) {
419                increment(self.rollbacks);
420                increment(self.failed_reloads);
421            }
422        }
423        self.routing.release_drained_retiring();
424        self.flag.store(false, Ordering::Release);
425    }
426}
427
428fn dynamic_router(routing: Arc<RoutingTable>) -> Router {
429    Router::new()
430        .fallback(any(dispatch_active_generation))
431        .with_state(routing)
432}
433
434async fn dispatch_active_generation(
435    State(routing): State<Arc<RoutingTable>>,
436    request: Request,
437) -> Response<Body> {
438    loop {
439        let generation = routing.active();
440        if let Some(_permit) = generation.try_admit() {
441            return generation
442                .router()
443                .clone()
444                .oneshot(request)
445                .await
446                .unwrap_or_else(|never| match never {});
447        }
448        tokio::time::sleep(ROUTING_RETRY_DELAY).await;
449    }
450}
451
452async fn probe_health(router: &Router, timeout: Duration) -> bool {
453    let request = Request::get("/v1/health").body(Body::empty());
454    let Ok(request) = request else {
455        return false;
456    };
457    tokio::time::timeout(timeout, router.clone().oneshot(request))
458        .await
459        .ok()
460        .and_then(Result::ok)
461        .is_some_and(|response| response.status() == StatusCode::OK)
462}
463
464async fn wait_for_drain(generation: &RoutingGeneration, timeout: Duration) -> bool {
465    let deadline = Instant::now() + timeout;
466    loop {
467        if generation.inflight() == 0 {
468            return true;
469        }
470        if Instant::now() >= deadline {
471            return false;
472        }
473        tokio::time::sleep(ROUTING_RETRY_DELAY).await;
474    }
475}
476
477async fn serve_listener(
478    listener: tokio::net::TcpListener,
479    router: Router,
480    shutdown: Arc<AtomicBool>,
481) -> io::Result<()> {
482    axum::serve(super::connection::with_read_timeout(listener), router)
483        .with_graceful_shutdown(super::wait_for_shutdown(shutdown))
484        .await
485}
486
487fn increment(counter: &AtomicU64) {
488    let _ = counter.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
489        Some(current.saturating_add(1))
490    });
491}
appcore_api/http/reload.rs

appcore_api/http/
reload.rs
