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    /// A candidate attempted to change whether ingress authentication is required.
105    AuthenticationPolicyChanged,
106    /// Generation identifiers must increase monotonically.
107    StaleGeneration,
108    /// Another reload transaction already owns the coordinator.
109    ReloadInProgress,
110    /// A failed or cancelled generation still owns in-flight requests.
111    RetiringGenerationBusy,
112    /// Health or drain deadlines must be non-zero.
113    InvalidPolicy,
114    /// The candidate or active generation failed its health gate.
115    HealthGateFailed(HttpReloadPhase),
116    /// The outgoing generation did not drain before rollback.
117    DrainTimedOut,
118    /// The failed generation remained active past the rollback drain deadline.
119    RollbackDrainTimedOut,
120}
121
122impl fmt::Display for RuntimeHttpReloadError {
123    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
124        match self {
125            Self::ListenerDisabled => formatter.write_str("HTTP reload listener is disabled"),
126            Self::ListenerAddressChanged => {
127                formatter.write_str("HTTP reload requires a prepared listener generation")
128            }
129            Self::AuthenticationPolicyChanged => {
130                formatter.write_str("HTTP reload cannot change the authentication boundary")
131            }
132            Self::StaleGeneration => formatter.write_str("HTTP routing generation must increase"),
133            Self::ReloadInProgress => formatter.write_str("HTTP reload is already in progress"),
134            Self::RetiringGenerationBusy => {
135                formatter.write_str("HTTP routing generation is still draining")
136            }
137            Self::InvalidPolicy => formatter.write_str("HTTP reload policy is invalid"),
138            Self::HealthGateFailed(phase) => {
139                write!(formatter, "HTTP reload health gate failed during {phase:?}")
140            }
141            Self::DrainTimedOut => formatter.write_str("HTTP routing generation drain timed out"),
142            Self::RollbackDrainTimedOut => {
143                formatter.write_str("HTTP rollback generation drain timed out")
144            }
145        }
146    }
147}
148
149impl std::error::Error for RuntimeHttpReloadError {}
150
151/// Prepared, health-gated candidate that can be consumed by one reload.
152pub struct PreparedRuntimeHttpGeneration {
153    generation: Arc<RoutingGeneration>,
154}
155
156impl fmt::Debug for PreparedRuntimeHttpGeneration {
157    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
158        formatter
159            .debug_struct("PreparedRuntimeHttpGeneration")
160            .field("generation", &self.generation.id())
161            .finish_non_exhaustive()
162    }
163}
164
165/// HTTP host whose stable listener dispatches through one atomic generation.
166pub struct ReloadableRuntimeHttpHost {
167    config: super::HttpApiConfig,
168    authentication_required: bool,
169    routing: Arc<RoutingTable>,
170    reload_in_progress: AtomicBool,
171    successful_reloads: AtomicU64,
172    failed_reloads: AtomicU64,
173    rollbacks: AtomicU64,
174}
175
176impl ReloadableRuntimeHttpHost {
177    /// Creates a reloadable host from an already composed initial host.
178    pub fn new(
179        initial_generation: u64,
180        host: RuntimeHttpHost,
181    ) -> Result<Self, RuntimeHttpReloadError> {
182        if !host.config().enabled {
183            return Err(RuntimeHttpReloadError::ListenerDisabled);
184        }
185        if initial_generation == 0 {
186            return Err(RuntimeHttpReloadError::StaleGeneration);
187        }
188        let config = host.config().clone();
189        let authentication_required = host.authentication_required();
190        Ok(Self {
191            config,
192            authentication_required,
193            routing: Arc::new(RoutingTable::new(initial_generation, host.router())),
194            reload_in_progress: AtomicBool::new(false),
195            successful_reloads: AtomicU64::new(0),
196            failed_reloads: AtomicU64::new(0),
197            rollbacks: AtomicU64::new(0),
198        })
199    }
200
201    /// Validates and owns a candidate without changing live routing.
202    pub fn prepare(
203        &self,
204        generation: u64,
205        host: RuntimeHttpHost,
206    ) -> Result<PreparedRuntimeHttpGeneration, RuntimeHttpReloadError> {
207        let active = self.routing.active();
208        if generation <= active.id() {
209            return Err(RuntimeHttpReloadError::StaleGeneration);
210        }
211        if !host.config().enabled {
212            return Err(RuntimeHttpReloadError::ListenerDisabled);
213        }
214        if host.config().host != self.config.host || host.config().port != self.config.port {
215            return Err(RuntimeHttpReloadError::ListenerAddressChanged);
216        }
217        if host.authentication_required() != self.authentication_required {
218            return Err(RuntimeHttpReloadError::AuthenticationPolicyChanged);
219        }
220        Ok(PreparedRuntimeHttpGeneration {
221            generation: self.routing.generation(generation, host.router()),
222        })
223    }
224
225    #[cfg(test)]
226    pub(super) fn new_for_test(
227        initial_generation: u64,
228        config: super::HttpApiConfig,
229        router: Router,
230    ) -> Self {
231        Self {
232            config,
233            authentication_required: true,
234            routing: Arc::new(RoutingTable::new(initial_generation, router)),
235            reload_in_progress: AtomicBool::new(false),
236            successful_reloads: AtomicU64::new(0),
237            failed_reloads: AtomicU64::new(0),
238            rollbacks: AtomicU64::new(0),
239        }
240    }
241
242    #[cfg(test)]
243    pub(super) fn prepare_router_for_test(
244        &self,
245        generation: u64,
246        router: Router,
247    ) -> PreparedRuntimeHttpGeneration {
248        PreparedRuntimeHttpGeneration {
249            generation: self.routing.generation(generation, router),
250        }
251    }
252
253    /// Returns a router that always dispatches through the active generation.
254    pub fn router(&self) -> Router {
255        dynamic_router(Arc::clone(&self.routing))
256    }
257
258    /// Runs the stable listener until cooperative shutdown is requested.
259    pub fn run_until_shutdown(&self, shutdown: Arc<AtomicBool>) -> io::Result<()> {
260        let address = format!("{}:{}", self.config.host, self.config.port);
261        let runtime = super::build_http_runtime()?;
262        runtime.block_on(async move {
263            let listener = tokio::net::TcpListener::bind(address).await?;
264            super::validate_listener_auth_boundary(&listener, self.authentication_required)?;
265            serve_listener(listener, self.router(), shutdown).await
266        })
267    }
268
269    /// Runs on a listener that the composition root already bound and checked.
270    pub fn run_on_listener_until_shutdown(
271        &self,
272        listener: std::net::TcpListener,
273        shutdown: Arc<AtomicBool>,
274    ) -> io::Result<()> {
275        listener.set_nonblocking(true)?;
276        let router = self.router();
277        let runtime = super::build_http_runtime()?;
278        runtime.block_on(async move {
279            let listener = tokio::net::TcpListener::from_std(listener)?;
280            super::validate_listener_auth_boundary(&listener, self.authentication_required)?;
281            serve_listener(listener, router, shutdown).await
282        })
283    }
284
285    /// Health-checks, atomically activates, and drains one prepared generation.
286    pub async fn reload(
287        &self,
288        prepared: PreparedRuntimeHttpGeneration,
289        policy: HttpReloadPolicy,
290    ) -> Result<(), RuntimeHttpReloadError> {
291        let mut guard = match ReloadGuard::acquire(
292            &self.reload_in_progress,
293            &self.routing,
294            &self.failed_reloads,
295            &self.rollbacks,
296        ) {
297            Ok(guard) => guard,
298            Err(error) => return self.fail(error),
299        };
300        self.routing.release_drained_retiring();
301        if self.routing.generations_snapshot().retiring.is_some() {
302            return self.fail(RuntimeHttpReloadError::RetiringGenerationBusy);
303        }
304        let previous = self.routing.active();
305        if prepared.generation.id() <= previous.id() {
306            return self.fail(RuntimeHttpReloadError::StaleGeneration);
307        }
308        if !probe_health(prepared.generation.router(), policy.health_timeout).await {
309            return self.fail(RuntimeHttpReloadError::HealthGateFailed(
310                HttpReloadPhase::Prepare,
311            ));
312        }
313
314        previous.stop_accepting();
315        if !self.routing.retire(Arc::clone(&previous)) {
316            previous.start_accepting();
317            return self.fail(RuntimeHttpReloadError::RetiringGenerationBusy);
318        }
319        guard.arm(Arc::clone(&previous), Arc::clone(&prepared.generation));
320        self.routing.activate(Arc::clone(&prepared.generation));
321        if !probe_health(prepared.generation.router(), policy.health_timeout).await {
322            guard.disarm();
323            return self
324                .rollback(previous, prepared.generation, policy)
325                .await
326                .and(Err(RuntimeHttpReloadError::HealthGateFailed(
327                    HttpReloadPhase::Switch,
328                )));
329        }
330        if !wait_for_drain(&previous, policy.drain_timeout).await {
331            guard.disarm();
332            return self
333                .rollback(previous, prepared.generation, policy)
334                .await
335                .and(Err(RuntimeHttpReloadError::DrainTimedOut));
336        }
337        self.routing.release_retiring(previous.id());
338        guard.disarm();
339        increment(&self.successful_reloads);
340        Ok(())
341    }
342
343    /// Returns the active generation and bounded operational counters.
344    pub fn snapshot(&self) -> HttpReloadSnapshot {
345        let active = self.routing.active();
346        HttpReloadSnapshot {
347            active_generation: active.id(),
348            active_inflight: active.inflight(),
349            reload_in_progress: self.reload_in_progress.load(Ordering::Acquire),
350            successful_reloads: self.successful_reloads.load(Ordering::Relaxed),
351            failed_reloads: self.failed_reloads.load(Ordering::Relaxed),
352            rollbacks: self.rollbacks.load(Ordering::Relaxed),
353        }
354    }
355
356    /// Returns the active and optional retiring generation without request data.
357    pub fn generation_snapshot(&self) -> HttpRoutingGenerationsSnapshot {
358        self.routing.generations_snapshot()
359    }
360
361    async fn rollback(
362        &self,
363        previous: Arc<RoutingGeneration>,
364        failed: Arc<RoutingGeneration>,
365        policy: HttpReloadPolicy,
366    ) -> Result<(), RuntimeHttpReloadError> {
367        failed.stop_accepting();
368        previous.start_accepting();
369        self.routing.activate(Arc::clone(&previous));
370        self.routing.release_retiring(previous.id());
371        if !self.routing.retire(Arc::clone(&failed)) {
372            return Err(RuntimeHttpReloadError::RetiringGenerationBusy);
373        }
374        increment(&self.rollbacks);
375        increment(&self.failed_reloads);
376        if wait_for_drain(&failed, policy.drain_timeout).await {
377            self.routing.release_retiring(failed.id());
378            Ok(())
379        } else {
380            Err(RuntimeHttpReloadError::RollbackDrainTimedOut)
381        }
382    }
383
384    fn fail<T>(&self, error: RuntimeHttpReloadError) -> Result<T, RuntimeHttpReloadError> {
385        increment(&self.failed_reloads);
386        Err(error)
387    }
388}
389
390struct ReloadGuard<'a> {
391    flag: &'a AtomicBool,
392    routing: &'a RoutingTable,
393    failed_reloads: &'a AtomicU64,
394    rollbacks: &'a AtomicU64,
395    rollback: Option<(Arc<RoutingGeneration>, Arc<RoutingGeneration>)>,
396}
397
398impl<'a> ReloadGuard<'a> {
399    fn acquire(
400        flag: &'a AtomicBool,
401        routing: &'a RoutingTable,
402        failed_reloads: &'a AtomicU64,
403        rollbacks: &'a AtomicU64,
404    ) -> Result<Self, RuntimeHttpReloadError> {
405        flag.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
406            .map_err(|_| RuntimeHttpReloadError::ReloadInProgress)?;
407        Ok(Self {
408            flag,
409            routing,
410            failed_reloads,
411            rollbacks,
412            rollback: None,
413        })
414    }
415
416    fn arm(&mut self, previous: Arc<RoutingGeneration>, candidate: Arc<RoutingGeneration>) {
417        self.rollback = Some((previous, candidate));
418    }
419
420    fn disarm(&mut self) {
421        self.rollback = None;
422    }
423}
424
425impl Drop for ReloadGuard<'_> {
426    fn drop(&mut self) {
427        if let Some((previous, failed)) = self.rollback.take() {
428            failed.stop_accepting();
429            previous.start_accepting();
430            self.routing.activate(Arc::clone(&previous));
431            self.routing.release_retiring(previous.id());
432            if self.routing.retire(failed) {
433                increment(self.rollbacks);
434                increment(self.failed_reloads);
435            }
436        }
437        self.routing.release_drained_retiring();
438        self.flag.store(false, Ordering::Release);
439    }
440}
441
442fn dynamic_router(routing: Arc<RoutingTable>) -> Router {
443    Router::new()
444        .fallback(any(dispatch_active_generation))
445        .with_state(routing)
446}
447
448async fn dispatch_active_generation(
449    State(routing): State<Arc<RoutingTable>>,
450    request: Request,
451) -> Response<Body> {
452    loop {
453        let generation = routing.active();
454        if let Some(_permit) = generation.try_admit() {
455            return generation
456                .router()
457                .clone()
458                .oneshot(request)
459                .await
460                .unwrap_or_else(|never| match never {});
461        }
462        tokio::time::sleep(ROUTING_RETRY_DELAY).await;
463    }
464}
465
466async fn probe_health(router: &Router, timeout: Duration) -> bool {
467    let request = Request::get("/v1/health").body(Body::empty());
468    let Ok(request) = request else {
469        return false;
470    };
471    tokio::time::timeout(timeout, router.clone().oneshot(request))
472        .await
473        .ok()
474        .and_then(Result::ok)
475        .is_some_and(|response| response.status() == StatusCode::OK)
476}
477
478async fn wait_for_drain(generation: &RoutingGeneration, timeout: Duration) -> bool {
479    let deadline = Instant::now() + timeout;
480    loop {
481        if generation.inflight() == 0 {
482            return true;
483        }
484        if Instant::now() >= deadline {
485            return false;
486        }
487        tokio::time::sleep(ROUTING_RETRY_DELAY).await;
488    }
489}
490
491async fn serve_listener(
492    listener: tokio::net::TcpListener,
493    router: Router,
494    shutdown: Arc<AtomicBool>,
495) -> io::Result<()> {
496    axum::serve(super::connection::with_read_timeout(listener), router)
497        .with_graceful_shutdown(super::wait_for_shutdown(shutdown))
498        .await
499}
500
501fn increment(counter: &AtomicU64) {
502    let _ = counter.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
503        Some(current.saturating_add(1))
504    });
505}
appcore_api/http/reload.rs

appcore_api/http/
reload.rs
