appcore_gateway/
runtime.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: runtime.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/08/20 00:00:00 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/08/20 00:00:00 by dnettoRaw
8//      ###########      S: 1.0.1-rc.8
9// =============================================================================
10
11//! Owned listener, task, health, and shutdown lifecycle for one Gateway.
12
13use crate::{
14    make_gateway_router, spawn_heartbeat_pruner, GatewayConfig, GatewayError, GatewayMetrics,
15    GatewayResult, GatewayState,
16};
17use appcore_peer_rpc::{BoundedReplayStore, PeerNonceStore, ReplayStoreConfig};
18use appcore_security::HashTokenProvider;
19use parking_lot::Mutex;
20use std::future::IntoFuture;
21use std::net::{SocketAddr, TcpListener};
22use std::sync::Arc;
23use std::thread::JoinHandle;
24use std::time::{Duration, Instant};
25use tokio::sync::watch;
26
27const MAX_SHUTDOWN_JOIN_RESERVE: Duration = Duration::from_millis(100);
28
29/// Concrete execution state of a Gateway runtime instance.
30#[derive(Debug, Clone, Copy, PartialEq, Eq)]
31pub enum GatewayRuntimeState {
32    /// No listener or task is active.
33    Stopped,
34    /// Listener preparation is in progress.
35    Starting,
36    /// The listener and owned tasks are running.
37    Running,
38    /// Cooperative shutdown is in progress.
39    Stopping,
40    /// The runtime terminated with a controlled failure.
41    Failed,
42    /// The runtime thread failed to honor even the forced shutdown deadline.
43    Orphaned,
44}
45
46/// Safe point-in-time Gateway lifecycle and metric snapshot.
47#[derive(Debug, Clone, PartialEq, Eq)]
48pub struct GatewayRuntimeSnapshot {
49    /// Current execution state.
50    pub state: GatewayRuntimeState,
51    /// Deployment-configured listener address.
52    pub configured_bind_address: SocketAddr,
53    /// Actual bound address while or after an instance has run.
54    pub bound_address: Option<SocketAddr>,
55    /// Active authenticated worker sockets.
56    pub active_workers: u64,
57    /// Active authenticated client sockets.
58    pub active_clients: u64,
59    /// Successfully routed messages since this instance started.
60    pub messages_routed: u64,
61    /// Failed routing attempts since this instance started.
62    pub routing_failures: u64,
63    /// Sanitized lifecycle failure, when present.
64    pub last_error: Option<String>,
65}
66
67struct RunningGateway {
68    shutdown: watch::Sender<Option<Duration>>,
69    handle: JoinHandle<GatewayResult<()>>,
70}
71
72struct RuntimeInner {
73    state: GatewayRuntimeState,
74    running: Option<RunningGateway>,
75    gateway_state: Option<Arc<GatewayState>>,
76    bound_address: Option<SocketAddr>,
77    last_error: Option<String>,
78}
79
80/// Restartable owner of one Gateway listener and all work spawned beneath it.
81pub struct GatewayRuntime {
82    config: GatewayConfig,
83    token_provider: HashTokenProvider,
84    connection_replay: Arc<dyn PeerNonceStore>,
85    inner: Mutex<RuntimeInner>,
86}
87
88impl GatewayRuntime {
89    /// Creates a stopped runtime after validating owner-defined configuration.
90    pub fn new(config: GatewayConfig, token_provider: HashTokenProvider) -> GatewayResult<Self> {
91        Self::with_replay_store(
92            config,
93            token_provider,
94            Arc::new(BoundedReplayStore::new(ReplayStoreConfig::default())),
95        )
96    }
97
98    /// Creates a stopped runtime using an explicit durable or shared replay
99    /// store for one-use Gateway connection credentials.
100    pub fn with_replay_store(
101        config: GatewayConfig,
102        token_provider: HashTokenProvider,
103        connection_replay: Arc<dyn PeerNonceStore>,
104    ) -> GatewayResult<Self> {
105        config.validate()?;
106        Ok(Self {
107            config,
108            token_provider,
109            connection_replay,
110            inner: Mutex::new(RuntimeInner {
111                state: GatewayRuntimeState::Stopped,
112                running: None,
113                gateway_state: None,
114                bound_address: None,
115                last_error: None,
116            }),
117        })
118    }
119
120    /// Binds synchronously and starts the listener on one owned runtime thread.
121    ///
122    /// Bind, runtime construction, and thread creation failures are returned
123    /// before this method reports success.
124    pub fn start(&self) -> GatewayResult<()> {
125        let mut inner = self.inner.lock();
126        refresh_runtime(&mut inner);
127        match inner.state {
128            GatewayRuntimeState::Running => return Ok(()),
129            GatewayRuntimeState::Starting | GatewayRuntimeState::Stopping => {
130                return Err(GatewayError::Transport(
131                    "gateway lifecycle transition already in progress".to_string(),
132                ));
133            }
134            GatewayRuntimeState::Orphaned => {
135                return Err(GatewayError::Transport(
136                    "orphaned gateway instance cannot be restarted".to_string(),
137                ));
138            }
139            GatewayRuntimeState::Stopped | GatewayRuntimeState::Failed => {}
140        }
141        inner.state = GatewayRuntimeState::Starting;
142        match self.prepare_instance() {
143            Ok(prepared) => {
144                inner.bound_address = Some(prepared.bound_address);
145                inner.gateway_state = Some(prepared.state);
146                inner.running = Some(prepared.running);
147                inner.last_error = None;
148                inner.state = GatewayRuntimeState::Running;
149                Ok(())
150            }
151            Err(error) => {
152                inner.state = GatewayRuntimeState::Failed;
153                inner.last_error = Some(error.to_string());
154                Err(error)
155            }
156        }
157    }
158
159    /// Requests graceful shutdown, force-cancels the server future before the
160    /// deadline when needed, and joins all owned listener and task work.
161    pub fn stop(&self, timeout: Duration) -> GatewayResult<()> {
162        let running = {
163            let mut inner = self.inner.lock();
164            refresh_runtime(&mut inner);
165            let Some(running) = inner.running.take() else {
166                if inner.state != GatewayRuntimeState::Orphaned {
167                    inner.state = GatewayRuntimeState::Stopped;
168                }
169                return Ok(());
170            };
171            inner.state = GatewayRuntimeState::Stopping;
172            running
173                .shutdown
174                .send_replace(Some(graceful_shutdown_budget(timeout)));
175            running
176        };
177        let deadline = Instant::now().checked_add(timeout);
178        while !running.handle.is_finished()
179            && deadline.is_none_or(|deadline| Instant::now() < deadline)
180        {
181            std::thread::sleep(Duration::from_millis(10));
182        }
183        if !running.handle.is_finished() {
184            let mut inner = self.inner.lock();
185            inner.running = Some(running);
186            inner.state = GatewayRuntimeState::Orphaned;
187            inner.last_error = Some("gateway shutdown timed out".to_string());
188            return Err(GatewayError::Transport(
189                "gateway shutdown timed out".to_string(),
190            ));
191        }
192        let result = join_runtime(running);
193        let mut inner = self.inner.lock();
194        inner.state = if result.is_ok() {
195            GatewayRuntimeState::Stopped
196        } else {
197            GatewayRuntimeState::Failed
198        };
199        inner.last_error = result.as_ref().err().map(ToString::to_string);
200        result
201    }
202
203    /// Returns lifecycle, listener, and bounded metric state without exposing
204    /// credentials or token material.
205    pub fn snapshot(&self) -> GatewayRuntimeSnapshot {
206        let mut inner = self.inner.lock();
207        refresh_runtime(&mut inner);
208        let metrics = inner
209            .gateway_state
210            .as_ref()
211            .map(|state| Arc::clone(&state.metrics));
212        snapshot_from_parts(&self.config, &inner, metrics.as_deref())
213    }
214
215    fn prepare_instance(&self) -> GatewayResult<PreparedGateway> {
216        let state = Arc::new(GatewayState::with_replay_store(
217            self.config.clone(),
218            self.token_provider.clone(),
219            Arc::clone(&self.connection_replay),
220        )?);
221        let listener = bind_listener(self.config.bind_address)?;
222        let bound_address = listener.local_addr().map_err(transport_error)?;
223        let runtime = tokio::runtime::Builder::new_current_thread()
224            .enable_all()
225            .build()
226            .map_err(transport_error)?;
227        let listener = {
228            let _entered = runtime.enter();
229            tokio::net::TcpListener::from_std(listener).map_err(transport_error)?
230        };
231        let (shutdown, shutdown_request) = watch::channel(None);
232        let thread_state = Arc::clone(&state);
233        let handle = std::thread::Builder::new()
234            .name("appcore-gateway".to_string())
235            .spawn(move || run_gateway(runtime, listener, thread_state, shutdown_request))
236            .map_err(transport_error)?;
237        Ok(PreparedGateway {
238            bound_address,
239            state,
240            running: RunningGateway { shutdown, handle },
241        })
242    }
243}
244
245struct PreparedGateway {
246    bound_address: SocketAddr,
247    state: Arc<GatewayState>,
248    running: RunningGateway,
249}
250
251impl Drop for GatewayRuntime {
252    fn drop(&mut self) {
253        let _ = self.stop(Duration::from_secs(10));
254    }
255}
256
257fn bind_listener(address: SocketAddr) -> GatewayResult<TcpListener> {
258    let listener = TcpListener::bind(address).map_err(|error| {
259        GatewayError::Transport(format!(
260            "failed to bind gateway listener {address}: {error}"
261        ))
262    })?;
263    listener.set_nonblocking(true).map_err(transport_error)?;
264    Ok(listener)
265}
266
267fn run_gateway(
268    runtime: tokio::runtime::Runtime,
269    listener: tokio::net::TcpListener,
270    state: Arc<GatewayState>,
271    mut shutdown: watch::Receiver<Option<Duration>>,
272) -> GatewayResult<()> {
273    runtime.block_on(async move {
274        let pruner = spawn_heartbeat_pruner(
275            Arc::clone(&state),
276            state.config().heartbeat_interval,
277            state.config().heartbeat_timeout,
278        );
279        let router = make_gateway_router(Arc::clone(&state));
280        let graceful_state = Arc::clone(&state);
281        let mut server = Box::pin(
282            axum::serve(listener, router)
283                .with_graceful_shutdown(async move {
284                    graceful_state.wait_for_shutdown().await;
285                })
286                .into_future(),
287        );
288        let exit = tokio::select! {
289            result = server.as_mut() => GatewayServerExit::Completed(result),
290            grace = wait_for_shutdown_request(&mut shutdown) => {
291                state.request_shutdown();
292                match tokio::time::timeout(grace, server.as_mut()).await {
293                    Ok(result) => GatewayServerExit::Completed(result),
294                    Err(_) => GatewayServerExit::Forced,
295                }
296            }
297        };
298        drop(server);
299        state.request_shutdown();
300        let pruner_result = pruner.await.map_err(|error| {
301            GatewayError::Transport(format!("gateway heartbeat pruner failed: {error}"))
302        });
303        let result = match exit {
304            GatewayServerExit::Completed(result) => result.map_err(transport_error),
305            GatewayServerExit::Forced => Err(GatewayError::Transport(
306                "gateway graceful shutdown timed out; forced cancellation completed".to_string(),
307            )),
308        };
309        result.and(pruner_result)
310    })
311}
312
313enum GatewayServerExit {
314    Completed(std::io::Result<()>),
315    Forced,
316}
317
318async fn wait_for_shutdown_request(shutdown: &mut watch::Receiver<Option<Duration>>) -> Duration {
319    loop {
320        if let Some(grace) = *shutdown.borrow() {
321            return grace;
322        }
323        if shutdown.changed().await.is_err() {
324            return Duration::ZERO;
325        }
326    }
327}
328
329fn graceful_shutdown_budget(timeout: Duration) -> Duration {
330    timeout.saturating_sub(timeout.min(MAX_SHUTDOWN_JOIN_RESERVE))
331}
332
333fn refresh_runtime(inner: &mut RuntimeInner) {
334    let finished = inner
335        .running
336        .as_ref()
337        .is_some_and(|running| running.handle.is_finished());
338    if !finished {
339        return;
340    }
341    let Some(running) = inner.running.take() else {
342        return;
343    };
344    let requested = running.shutdown.borrow().is_some();
345    let result = join_runtime(running);
346    inner.state = if requested && result.is_ok() {
347        GatewayRuntimeState::Stopped
348    } else {
349        GatewayRuntimeState::Failed
350    };
351    inner.last_error = result.err().map(|error| error.to_string());
352}
353
354fn join_runtime(running: RunningGateway) -> GatewayResult<()> {
355    running
356        .handle
357        .join()
358        .map_err(|_| GatewayError::Transport("gateway runtime thread panicked".to_string()))?
359}
360
361fn snapshot_from_parts(
362    config: &GatewayConfig,
363    inner: &RuntimeInner,
364    metrics: Option<&GatewayMetrics>,
365) -> GatewayRuntimeSnapshot {
366    GatewayRuntimeSnapshot {
367        state: inner.state,
368        configured_bind_address: config.bind_address,
369        bound_address: inner.bound_address,
370        active_workers: metrics.map_or(0, GatewayMetrics::active_workers),
371        active_clients: metrics.map_or(0, GatewayMetrics::active_clients),
372        messages_routed: metrics.map_or(0, GatewayMetrics::messages_routed),
373        routing_failures: metrics.map_or(0, GatewayMetrics::routing_failures),
374        last_error: inner.last_error.clone(),
375    }
376}
377
378fn transport_error(error: impl std::fmt::Display) -> GatewayError {
379    GatewayError::Transport(error.to_string())
380}
381
382#[cfg(test)]
383mod tests {
384    use super::*;
385    use std::io::Write;
386    use std::net::TcpStream;
387
388    fn runtime(address: SocketAddr) -> GatewayRuntime {
389        GatewayRuntime::new(
390            GatewayConfig::new(address, "gateway.test"),
391            HashTokenProvider::from_secret(vec![7; 32]).unwrap(),
392        )
393        .unwrap()
394    }
395
396    #[test]
397    fn bind_failure_is_synchronous_and_fail_closed() {
398        let occupied = TcpListener::bind("127.0.0.1:0").unwrap();
399        let address = occupied.local_addr().unwrap();
400        let gateway = runtime(address);
401
402        let error = gateway.start().unwrap_err();
403
404        assert!(error
405            .to_string()
406            .contains("failed to bind gateway listener"));
407        assert_eq!(gateway.snapshot().state, GatewayRuntimeState::Failed);
408    }
409
410    #[test]
411    fn shutdown_releases_listener_and_owned_runtime_thread() {
412        let gateway = runtime("127.0.0.1:0".parse().unwrap());
413        gateway.start().unwrap();
414        let address = gateway.snapshot().bound_address.unwrap();
415        assert!(TcpStream::connect_timeout(&address, Duration::from_secs(1)).is_ok());
416
417        gateway.stop(Duration::from_secs(2)).unwrap();
418
419        assert_eq!(gateway.snapshot().state, GatewayRuntimeState::Stopped);
420        assert!(TcpListener::bind(address).is_ok());
421    }
422
423    #[test]
424    fn shutdown_force_closes_an_incomplete_http_connection_before_deadline() {
425        let gateway = runtime("127.0.0.1:0".parse().unwrap());
426        gateway.start().unwrap();
427        let address = gateway.snapshot().bound_address.unwrap();
428        let mut client = TcpStream::connect(address).unwrap();
429        client
430            .write_all(b"GET /v1/mesh-relay HTTP/1.1\r\nHost: gateway.test\r\n")
431            .unwrap();
432        std::thread::sleep(Duration::from_millis(100));
433
434        let started = Instant::now();
435        let _ = gateway.stop(Duration::from_millis(500));
436
437        assert!(started.elapsed() < Duration::from_secs(1));
438        assert_ne!(gateway.snapshot().state, GatewayRuntimeState::Orphaned);
439        drop(client);
440        assert!(TcpListener::bind(address).is_ok());
441    }
442}
appcore_gateway/runtime.rs

appcore_gateway/
runtime.rs
