saluki_app/
dynamic_api.rs

1//! Dynamic API server.
2//!
3//! Unlike [`APIBuilder`][crate::api::APIBuilder], which constructs its route set once at build time,
4//! `DynamicAPIBuilder` subscribes to runtime notifications via the dataspace registry and dynamically registers and
5//! unregisters routes as they're asserted and retracted.
6
7use std::{
8    convert::Infallible,
9    error::Error,
10    future::Future,
11    net::SocketAddr,
12    panic::{catch_unwind, AssertUnwindSafe},
13    pin::Pin,
14    sync::Arc,
15    task::{Context, Poll},
16};
17
18use arc_swap::ArcSwap;
19use async_trait::async_trait;
20use axum::{body::Body as AxumBody, Router};
21use http::{Request, Response};
22use rcgen::{generate_simple_self_signed, CertifiedKey};
23use rustls::{pki_types::PrivateKeyDer, ServerConfig};
24use rustls_pki_types::PrivatePkcs8KeyDer;
25use saluki_api::{APIHandler, DynamicRoute, EndpointProtocol, EndpointType};
26use saluki_common::{collections::FastIndexMap, sync::shutdown::ShutdownHandle};
27use saluki_core::runtime::{
28    state::{DataspaceRegistry, DataspaceUpdate, Identifier, IdentifierFilter, Subscription},
29    InitializationError, Supervisable, SupervisorFuture,
30};
31use saluki_error::{generic_error, GenericError};
32use saluki_io::net::{
33    listener::ConnectionOrientedListener,
34    server::{http::HttpServer, multiplex_service::MultiplexService},
35    util::hyper::TowerToHyperService,
36    ListenAddress,
37};
38use saluki_tls::ensure_server_config_fips_compliant;
39use tokio::select;
40use tonic::{body::Body as GrpcBody, server::NamedService, service::RoutesBuilder};
41use tower::Service;
42use tracing::{debug, info, warn};
43
44/// The actual bound listen address of a running dynamic API server.
45///
46/// Asserted by dynamic API servers to allow discovering the exact socket address the server is bound to.
47#[derive(Clone, Debug)]
48pub struct BoundApiAddress(pub SocketAddr);
49
50/// A dynamic API server that can add and remove routes at runtime.
51///
52/// `DynamicAPIBuilder` serves HTTP and gRPC on a given address, multiplexing both protocols on a single port. Route
53/// additions and removals are handled by subscribing to assertions/retractions of [`DynamicRoute`] in the
54/// [`DataspaceRegistry`].
55///
56/// ## Adding and removing routes
57///
58/// Any process that wants to dynamically register API routes can simply assert a [`DynamicRoute`] in the
59/// [`DataspaceRegistry`]. Retracting the assertion will remove the route, either when retracted manually or when the
60/// process owning the route assertions exits.
61///
62/// If the dynamic API server is restarted, it will re-register any routes that were previously asserted.
63///
64/// ## Static handlers and services
65///
66/// In addition to dynamic routes, callers can register static HTTP handlers and gRPC services up-front via
67/// [`with_handler`][Self::with_handler], [`with_optional_handler`][Self::with_optional_handler], and
68/// [`with_grpc_service`][Self::with_grpc_service]. These form a base router that's cloned on every rebuild and merged
69/// with the currently asserted dynamic routes. Static routes take precedence on conflicts: a dynamic route whose path
70/// and method overlap with a static route is skipped (with a warning) until the conflict clears.
71///
72/// ## Assertions
73///
74/// - `BoundApiAddress`: the actual listen address bound by the API server. Identifier is `"dynamic-<type>-api"`, where
75///   `type` is the stringified value of `EndpointType::as_str` (for example, `"dynamic-privileged-api"`)
76pub struct DynamicAPIBuilder {
77    endpoint_type: EndpointType,
78    listen_address: ListenAddress,
79    tls_config: Option<ServerConfig>,
80    http_router: Router,
81    grpc_router: RoutesBuilder,
82}
83
84impl DynamicAPIBuilder {
85    /// Creates a new `DynamicAPIBuilder` for the given endpoint type and listen address.
86    pub fn new(endpoint_type: EndpointType, listen_address: ListenAddress) -> Self {
87        Self {
88            endpoint_type,
89            listen_address,
90            tls_config: None,
91            http_router: Router::new(),
92            grpc_router: RoutesBuilder::default(),
93        }
94    }
95
96    /// Adds the given handler as a static HTTP handler.
97    ///
98    /// The handler's initial state and routes are merged into the base router. These routes are always served by the
99    /// API regardless of which dynamic routes are currently asserted.
100    pub fn with_handler<H>(mut self, handler: H) -> Self
101    where
102        H: APIHandler,
103    {
104        let handler_router = handler.generate_routes();
105        let handler_state = handler.generate_initial_state();
106        self.http_router = self.http_router.merge(handler_router.with_state(handler_state));
107        self
108    }
109
110    /// Adds the given optional handler as a static HTTP handler.
111    ///
112    /// If `handler` is `Some`, its initial state and routes are merged into the base router. Otherwise the builder is
113    /// returned unchanged.
114    pub fn with_optional_handler<H>(self, handler: Option<H>) -> Self
115    where
116        H: APIHandler,
117    {
118        if let Some(handler) = handler {
119            self.with_handler(handler)
120        } else {
121            self
122        }
123    }
124
125    /// Adds the given gRPC service as a static service on the base router.
126    pub fn with_grpc_service<S>(mut self, svc: S) -> Self
127    where
128        S: Service<Request<GrpcBody>, Response = Response<GrpcBody>, Error = Infallible>
129            + NamedService
130            + Clone
131            + Send
132            + Sync
133            + 'static,
134        S::Future: Send + 'static,
135        S::Error: Into<Box<dyn Error + Send + Sync>> + Send,
136    {
137        self.grpc_router.add_service(svc);
138        self
139    }
140
141    /// Sets the TLS configuration for the server.
142    pub fn with_tls_config(mut self, config: ServerConfig) -> Self {
143        self.tls_config = Some(config);
144        self
145    }
146
147    /// Sets the TLS configuration for the server based on a dynamically generated, self-signed certificate.
148    pub fn with_self_signed_tls(self) -> Self {
149        self.try_with_self_signed_tls()
150            .expect("self-signed server TLS configuration should build and pass FIPS validation")
151    }
152
153    /// Sets the TLS configuration for the server based on a dynamically generated, self-signed certificate.
154    ///
155    /// # Errors
156    ///
157    /// If the certificate cannot be generated, the TLS configuration cannot be built, or the resulting TLS
158    /// configuration is not FIPS compliant, an error is returned.
159    pub fn try_with_self_signed_tls(self) -> Result<Self, GenericError> {
160        let CertifiedKey { cert, signing_key } = generate_simple_self_signed(["localhost".to_owned()])?;
161        let cert_chain = vec![cert.der().clone()];
162        let key = PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(signing_key.serialize_der()));
163
164        let mut config = ServerConfig::builder()
165            .with_no_client_auth()
166            .with_single_cert(cert_chain, key)?;
167
168        ensure_server_config_fips_compliant(&mut config)?;
169
170        Ok(self.with_tls_config(config))
171    }
172}
173
174#[async_trait]
175impl Supervisable for DynamicAPIBuilder {
176    fn name(&self) -> &str {
177        match self.endpoint_type {
178            EndpointType::Unprivileged => "dynamic-unprivileged-api",
179            EndpointType::Privileged => "dynamic-privileged-api",
180        }
181    }
182
183    async fn initialize(&self, process_shutdown: ShutdownHandle) -> Result<SupervisorFuture, InitializationError> {
184        // Build the static base routers.
185        //
186        // We reset the fallback route of the base gRPC router as Tonic's `unimplemented` fallback handle will collide
187        // when merging additional gRPC service routers together. We do this for every gRPC service router that we merge
188        // and then we re-apply a fallback handler that returns a standard gRPC `UNIMPLEMENTED` response when we have
189        // our final, merged gRPC router.
190        let base_http = self.http_router.clone();
191        let base_grpc = self.grpc_router.clone().routes().into_axum_router().reset_fallback();
192
193        // Create dynamic inner routers for both HTTP and gRPC sides, seeded with the static base so that the static
194        // routes are served even before any dynamic routes are asserted. The gRPC seed gets the unimplemented fallback
195        // applied so unmatched gRPC requests return the correct status from the start.
196        let (inner_http, outer_http) = create_dynamic_router(base_http.clone());
197        let (inner_grpc, outer_grpc) = create_dynamic_router(grpc_post_process(base_grpc.clone()));
198
199        let dataspace = DataspaceRegistry::try_current().ok_or_else(|| generic_error!("Dataspace not available."))?;
200
201        // Bind the HTTP listener immediately so we fail fast on bind errors.
202        let listener = ConnectionOrientedListener::from_listen_address(self.listen_address.clone())
203            .await
204            .map_err(|e| InitializationError::Failed { source: e.into() })?;
205
206        // Get the listen address our listener is bound to and assert it.
207        //
208        // This allows other processes to find out where we've bound to when the port isn't known ahead of time,
209        // such as during tests when binding to ephemeral ports.
210        let bound_addr = listener
211            .local_addr()
212            .map_err(|e| InitializationError::Failed { source: e.into() })?;
213        dataspace.assert(BoundApiAddress(bound_addr), Identifier::named(self.name()));
214
215        let multiplexed_service = TowerToHyperService::new(MultiplexService::new(outer_http, outer_grpc));
216
217        let mut http_server = HttpServer::from_listener(listener, multiplexed_service);
218        if let Some(tls_config) = self.tls_config.clone() {
219            http_server = http_server.with_tls_config(tls_config);
220        }
221        let (server_shutdown_coordinator, error_handle) = http_server.listen();
222
223        let endpoint_type = self.endpoint_type;
224        let listen_address = self.listen_address.clone();
225
226        Ok(Box::pin(async move {
227            info!("Serving {} API on {}.", endpoint_type.name(), listen_address);
228
229            // Subscribe to all dynamic route assertions.
230            let route_assertions = dataspace.subscribe::<DynamicRoute>(IdentifierFilter::All);
231
232            select! {
233                _ = process_shutdown => {
234                    // Trigger the HTTP server to shut down and wait for it to do so gracefully.
235                    debug!(endpoint_type = endpoint_type.name(), "Triggering shutdown of dynamic API endpoint.");
236
237                    server_shutdown_coordinator.shutdown_and_wait().await;
238
239                    Ok(())
240                },
241                maybe_err = error_handle => match maybe_err {
242                    Some(e) => Err(GenericError::from(e)),
243                    None => Ok(()),
244                },
245                result = run_event_loop(inner_http, inner_grpc, base_http, base_grpc, route_assertions, endpoint_type) => result,
246            }
247        }))
248    }
249}
250
251/// A [`tower::Service`] that routes a request based on a dynamically updated [`Router`].
252///
253/// When installed as the fallback service for a top-level [`Router`], `DynamicRouterService` dynamically routing
254/// requests based on the current defined "inner" router, which itself can be hot-swapped at runtime. This allows for
255/// seamless updates to the API endpoint routing without requiring a restart of the HTTP listener or complicated
256/// configuration changes.
257#[derive(Clone)]
258struct DynamicRouterService {
259    inner_router: Arc<ArcSwap<Router>>,
260}
261
262impl DynamicRouterService {
263    fn from_inner(inner_router: &Arc<ArcSwap<Router>>) -> Self {
264        Self {
265            inner_router: Arc::clone(inner_router),
266        }
267    }
268}
269
270impl Service<http::Request<AxumBody>> for DynamicRouterService {
271    type Response = Response<AxumBody>;
272    type Error = Infallible;
273    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
274
275    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
276        Poll::Ready(Ok(()))
277    }
278
279    fn call(&mut self, request: http::Request<AxumBody>) -> Self::Future {
280        let mut router = Arc::unwrap_or_clone(self.inner_router.load_full());
281        Box::pin(async move { router.call(request).await })
282    }
283}
284
285/// Runs the event loop that listens for route assertions/retractions and hot-swaps the inner routers.
286#[allow(clippy::too_many_arguments)]
287async fn run_event_loop(
288    inner_http: Arc<ArcSwap<Router>>, inner_grpc: Arc<ArcSwap<Router>>, base_http: Router, base_grpc: Router,
289    mut route_assertions: Subscription<DynamicRoute>, endpoint_type: EndpointType,
290) -> Result<(), GenericError> {
291    let mut http_handlers = FastIndexMap::default();
292    let mut grpc_handlers = FastIndexMap::default();
293
294    while let Some(update) = route_assertions.recv().await {
295        let mut rebuild_http = false;
296        let mut rebuild_grpc = false;
297
298        match update {
299            DataspaceUpdate::Asserted(id, route) => {
300                if route.endpoint_type() != endpoint_type {
301                    continue;
302                }
303
304                match route.endpoint_protocol() {
305                    EndpointProtocol::Http => {
306                        debug!(?id, "Registering dynamic HTTP handler.");
307                        http_handlers.insert(id, route.into_router());
308
309                        rebuild_http = true;
310                    }
311                    EndpointProtocol::Grpc => {
312                        debug!(?id, "Registering dynamic gRPC handler.");
313                        grpc_handlers.insert(id, route.into_router());
314
315                        rebuild_grpc = true;
316                    }
317                }
318            }
319            DataspaceUpdate::Retracted(id) => {
320                if http_handlers.swap_remove(&id).is_some() {
321                    debug!(?id, "Withdrawing dynamic HTTP handler.");
322                    rebuild_http = true;
323                }
324
325                if grpc_handlers.swap_remove(&id).is_some() {
326                    debug!(?id, "Withdrawing dynamic gRPC handler.");
327                    rebuild_grpc = true;
328                }
329            }
330            // Routes are modeled as assertions; transient messages are not meaningful here.
331            DataspaceUpdate::Message(..) => continue,
332        }
333
334        if rebuild_http {
335            rebuild_router(&inner_http, &base_http, &http_handlers, http_post_process);
336        }
337
338        if rebuild_grpc {
339            rebuild_router(&inner_grpc, &base_grpc, &grpc_handlers, grpc_post_process);
340        }
341    }
342
343    Ok(())
344}
345
346/// Creates a dynamic router pair: a swappable inner router (seeded with `initial`) and an outer router that delegates
347/// to it.
348fn create_dynamic_router(initial: Router) -> (Arc<ArcSwap<Router>>, Router) {
349    let inner = Arc::new(ArcSwap::from_pointee(initial));
350    let outer = Router::new().fallback_service(DynamicRouterService::from_inner(&inner));
351    (inner, outer)
352}
353
354/// Attempts to merge `other` into `base`, returning the merged router on success.
355///
356/// `Router::merge` panics when two routers define overlapping routes (same path and HTTP method) and axum exposes no
357/// fallible alternative. Since `Router` is opaque -- there is no public API to inspect which paths/methods a router
358/// carries -- we can't detect conflicts ahead of time.
359///
360/// To recover from the panic without losing the accumulated router state, we clone `base` before the merge attempt.
361/// The clone is passed into `catch_unwind`: if the merge panics, only the clone is in a partially mutated state and it
362/// is simply dropped. The original `base` remains intact and is returned as-is. `AssertUnwindSafe` is sound here
363/// because:
364///
365/// - The closure captures only the clone (`candidate`) and a clone of `other`. Neither aliases mutable state that
366///   outlives the closure.
367/// - The panic originates from a deterministic format string in axum's `panic_on_err!` macro -- no locks are held and
368///   no resources are leaked in the panic path.
369/// - On panic, `candidate` is dropped without further use, so any internal inconsistency is irrelevant.
370fn try_merge_router(base: &Router, id: &Identifier, other: &Router) -> Result<Router, String> {
371    let candidate = base.clone();
372    match catch_unwind(AssertUnwindSafe(|| candidate.merge(other.clone()))) {
373        Ok(merged) => Ok(merged),
374        Err(payload) => {
375            let reason = payload
376                .downcast_ref::<String>()
377                .map(|s| s.as_str())
378                .or_else(|| payload.downcast_ref::<&str>().copied())
379                .unwrap_or("unknown");
380            Err(format!("failed to merge dynamic handler {id:?}: {reason}"))
381        }
382    }
383}
384
385/// Rebuilds the merged inner router from the static `base` and all currently registered dynamic handlers, applies
386/// `post_process` to the merged router, then stores the result in the [`ArcSwap`].
387fn rebuild_router(
388    inner_router: &Arc<ArcSwap<Router>>, base: &Router, handlers: &FastIndexMap<Identifier, Router>,
389    post_process: fn(Router) -> Router,
390) {
391    let mut merged = base.clone();
392    let mut skipped = 0usize;
393
394    for (id, router) in handlers.iter() {
395        let resetable = router.clone().reset_fallback();
396        match try_merge_router(&merged, id, &resetable) {
397            Ok(new_merged) => merged = new_merged,
398            Err(reason) => {
399                warn!(%reason, "Skipping dynamic handler due to overlapping route.");
400                skipped += 1;
401            }
402        }
403    }
404
405    let merged = post_process(merged);
406    inner_router.store(Arc::new(merged));
407    debug!(handler_count = handlers.len(), skipped, "Rebuilt inner router.");
408}
409
410fn http_post_process(router: Router) -> Router {
411    router
412}
413
414/// Adds a fallback handler that returns a standard gRPC `UNIMPLEMENTED` response when no other handler matches.
415fn grpc_post_process(router: Router) -> Router {
416    router.fallback(grpc_unimplemented)
417}
418
419async fn grpc_unimplemented() -> Response<AxumBody> {
420    tonic::Status::unimplemented("").into_http()
421}
422
423#[cfg(test)]
424mod tests {
425    use std::{net::SocketAddr, time::Duration};
426
427    use async_trait::async_trait;
428    use axum::Router;
429    use http_body_util::{BodyExt as _, Empty};
430    use hyper::{body::Bytes, StatusCode};
431    use hyper_util::{client::legacy::Client, rt::TokioExecutor};
432    use saluki_api::{APIHandler, DynamicRoute, EndpointType};
433    use saluki_core::runtime::{
434        state::{DataspaceRegistry, DataspaceUpdate, Identifier, IdentifierFilter},
435        InitializationError, Supervisable, Supervisor, SupervisorFuture,
436    };
437    use tokio::{
438        pin, select,
439        sync::{mpsc, oneshot},
440        task::JoinHandle,
441        time::{sleep, timeout, Instant},
442    };
443
444    use super::*;
445
446    struct SimpleHandler {
447        path: &'static str,
448        body: &'static str,
449    }
450
451    impl APIHandler for SimpleHandler {
452        type State = ();
453
454        fn generate_initial_state(&self) -> Self::State {}
455
456        fn generate_routes(&self) -> Router<Self::State> {
457            let body = self.body;
458            Router::new().route(self.path, axum::routing::get(move || async move { body }))
459        }
460    }
461
462    enum RouteCommand {
463        Assert { id: Identifier, route: DynamicRoute },
464        Retract { id: Identifier },
465    }
466
467    struct RouteAsserter {
468        commands_rx: std::sync::Mutex<Option<mpsc::Receiver<RouteCommand>>>,
469        addr_tx: std::sync::Mutex<Option<oneshot::Sender<SocketAddr>>>,
470        endpoint_type: EndpointType,
471    }
472
473    #[async_trait]
474    impl Supervisable for RouteAsserter {
475        fn name(&self) -> &str {
476            "route-asserter"
477        }
478
479        async fn initialize(&self, process_shutdown: ShutdownHandle) -> Result<SupervisorFuture, InitializationError> {
480            let mut commands_rx =
481                self.commands_rx
482                    .lock()
483                    .unwrap()
484                    .take()
485                    .ok_or_else(|| InitializationError::Failed {
486                        source: generic_error!("RouteAsserter can only be initialized once"),
487                    })?;
488            let addr_tx = self.addr_tx.lock().unwrap().take();
489            let endpoint_type = self.endpoint_type;
490
491            Ok(Box::pin(async move {
492                let dataspace =
493                    DataspaceRegistry::try_current().ok_or_else(|| generic_error!("Dataspace not available."))?;
494
495                // Wait for the DynamicAPIBuilder to assert its bound address.
496                let bound_addr_name = match endpoint_type {
497                    EndpointType::Unprivileged => "dynamic-unprivileged-api",
498                    EndpointType::Privileged => "dynamic-privileged-api",
499                };
500                let mut addr_sub =
501                    dataspace.subscribe::<BoundApiAddress>(IdentifierFilter::exact(Identifier::named(bound_addr_name)));
502
503                let addr = match addr_sub.recv().await {
504                    Some(DataspaceUpdate::Asserted(_, BoundApiAddress(mut addr))) => {
505                        // Convert 0.0.0.0 to 127.0.0.1 so the test client can connect.
506                        if addr.ip().is_unspecified() {
507                            addr.set_ip(std::net::Ipv4Addr::LOCALHOST.into());
508                        }
509                        addr
510                    }
511                    other => return Err(generic_error!("unexpected bound address update: {:?}", other)),
512                };
513
514                if let Some(tx) = addr_tx {
515                    let _ = tx.send(addr);
516                }
517
518                // Process route commands until shutdown.
519                pin!(process_shutdown);
520
521                loop {
522                    select! {
523                        _ = &mut process_shutdown => break,
524                        cmd = commands_rx.recv() => {
525                            let Some(cmd) = cmd else { break };
526                            match cmd {
527                                RouteCommand::Assert { id, route } => {
528                                    dataspace.assert(route, id);
529                                }
530                                RouteCommand::Retract { id } => {
531                                    dataspace.retract::<DynamicRoute>(id);
532                                }
533                            }
534                        }
535                    }
536                }
537
538                Ok(())
539            }))
540        }
541    }
542
543    struct TestHarness {
544        addr: SocketAddr,
545        commands: mpsc::Sender<RouteCommand>,
546        _shutdown: oneshot::Sender<()>,
547        _handle: JoinHandle<()>,
548    }
549
550    impl TestHarness {
551        async fn assert_route(&self, id: impl Into<Identifier>, route: DynamicRoute) {
552            self.commands
553                .send(RouteCommand::Assert { id: id.into(), route })
554                .await
555                .unwrap();
556        }
557
558        async fn retract_route(&self, id: impl Into<Identifier>) {
559            self.commands
560                .send(RouteCommand::Retract { id: id.into() })
561                .await
562                .unwrap();
563        }
564    }
565
566    async fn setup_test_harness(endpoint_type: EndpointType) -> TestHarness {
567        setup_test_harness_with(endpoint_type, |b| b).await
568    }
569
570    async fn setup_test_harness_with<F>(endpoint_type: EndpointType, configure: F) -> TestHarness
571    where
572        F: FnOnce(DynamicAPIBuilder) -> DynamicAPIBuilder,
573    {
574        let (commands_tx, commands_rx) = mpsc::channel(16);
575        let (addr_tx, addr_rx) = oneshot::channel();
576
577        let api_builder = configure(DynamicAPIBuilder::new(endpoint_type, ListenAddress::any_tcp(0)));
578        let route_asserter = RouteAsserter {
579            commands_rx: std::sync::Mutex::new(Some(commands_rx)),
580            addr_tx: std::sync::Mutex::new(Some(addr_tx)),
581            endpoint_type,
582        };
583
584        let mut sup = Supervisor::new("test-dynamic-api").unwrap();
585        sup.add_worker(api_builder);
586        sup.add_worker(route_asserter);
587
588        let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();
589        let handle = tokio::spawn(async move {
590            let _ = sup.run_with_shutdown(shutdown_rx).await;
591        });
592
593        let addr = timeout(Duration::from_secs(5), addr_rx)
594            .await
595            .expect("timed out waiting for bound address")
596            .expect("addr channel closed");
597
598        TestHarness {
599            addr,
600            commands: commands_tx,
601            _shutdown: shutdown_tx,
602            _handle: handle,
603        }
604    }
605
606    async fn http_get(addr: SocketAddr, path: &str) -> (StatusCode, String) {
607        let client: Client<_, Empty<Bytes>> = Client::builder(TokioExecutor::new()).build_http();
608        let uri = format!("http://{}{}", addr, path);
609        let resp = client.get(uri.parse().unwrap()).await.unwrap();
610        let status = resp.status();
611        let body = resp.into_body().collect().await.unwrap().to_bytes();
612        let body_str = String::from_utf8_lossy(&body).into_owned();
613        (status, body_str)
614    }
615
616    async fn grpc_post(addr: SocketAddr, path: &str) -> (StatusCode, http::HeaderMap) {
617        let client: Client<_, Empty<Bytes>> = Client::builder(TokioExecutor::new()).build_http();
618        let uri: hyper::Uri = format!("http://{}{}", addr, path).parse().unwrap();
619        let req = hyper::Request::builder()
620            .uri(uri)
621            .method(hyper::Method::POST)
622            .header(hyper::header::CONTENT_TYPE, "application/grpc")
623            .body(Empty::<Bytes>::new())
624            .unwrap();
625        let resp = client.request(req).await.unwrap();
626        (resp.status(), resp.headers().clone())
627    }
628
629    async fn assert_status_eventually(addr: SocketAddr, path: &str, expected: StatusCode) -> String {
630        let deadline = Instant::now() + Duration::from_secs(2);
631        loop {
632            let (status, body) = http_get(addr, path).await;
633            if status == expected {
634                return body;
635            }
636            if Instant::now() > deadline {
637                panic!("expected {} for {} but got {}", expected, path, status);
638            }
639            sleep(Duration::from_millis(50)).await;
640        }
641    }
642
643    // -- Tests ---------------------------------------------------------------------------
644
645    #[tokio::test]
646    async fn serves_asserted_http_route() {
647        let harness = setup_test_harness(EndpointType::Unprivileged).await;
648
649        let route = DynamicRoute::http(
650            EndpointType::Unprivileged,
651            SimpleHandler {
652                path: "/health",
653                body: "ok",
654            },
655        );
656        harness.assert_route("health", route).await;
657
658        let body = assert_status_eventually(harness.addr, "/health", StatusCode::OK).await;
659        assert_eq!(body, "ok");
660    }
661
662    #[tokio::test]
663    async fn returns_404_for_unknown_route() {
664        let harness = setup_test_harness(EndpointType::Unprivileged).await;
665        let (status, _) = http_get(harness.addr, "/nonexistent").await;
666        assert_eq!(status, StatusCode::NOT_FOUND);
667    }
668
669    #[tokio::test]
670    async fn route_retraction_removes_route() {
671        let harness = setup_test_harness(EndpointType::Unprivileged).await;
672
673        let route = DynamicRoute::http(
674            EndpointType::Unprivileged,
675            SimpleHandler {
676                path: "/temp",
677                body: "temporary",
678            },
679        );
680        harness.assert_route("temp", route).await;
681        assert_status_eventually(harness.addr, "/temp", StatusCode::OK).await;
682
683        harness.retract_route("temp").await;
684        assert_status_eventually(harness.addr, "/temp", StatusCode::NOT_FOUND).await;
685    }
686
687    #[tokio::test]
688    async fn multiple_routes_independent_lifecycle() {
689        let harness = setup_test_harness(EndpointType::Unprivileged).await;
690
691        let route_a = DynamicRoute::http(
692            EndpointType::Unprivileged,
693            SimpleHandler {
694                path: "/a",
695                body: "alpha",
696            },
697        );
698        let route_b = DynamicRoute::http(
699            EndpointType::Unprivileged,
700            SimpleHandler {
701                path: "/b",
702                body: "bravo",
703            },
704        );
705        harness.assert_route("a", route_a).await;
706        harness.assert_route("b", route_b).await;
707
708        assert_status_eventually(harness.addr, "/a", StatusCode::OK).await;
709        assert_status_eventually(harness.addr, "/b", StatusCode::OK).await;
710
711        // Retract only /a -- /b should remain.
712        harness.retract_route("a").await;
713        assert_status_eventually(harness.addr, "/a", StatusCode::NOT_FOUND).await;
714
715        let body = assert_status_eventually(harness.addr, "/b", StatusCode::OK).await;
716        assert_eq!(body, "bravo");
717    }
718
719    #[tokio::test]
720    async fn ignores_routes_for_different_endpoint_type() {
721        let harness = setup_test_harness(EndpointType::Unprivileged).await;
722
723        // Assert a Privileged route on an Unprivileged server -- should be ignored.
724        let wrong_route = DynamicRoute::http(
725            EndpointType::Privileged,
726            SimpleHandler {
727                path: "/secret",
728                body: "secret",
729            },
730        );
731        harness.assert_route("secret", wrong_route).await;
732
733        let (status, _) = http_get(harness.addr, "/secret").await;
734        assert_eq!(status, StatusCode::NOT_FOUND);
735
736        // Now assert the same path with the correct endpoint type.
737        let right_route = DynamicRoute::http(
738            EndpointType::Unprivileged,
739            SimpleHandler {
740                path: "/secret",
741                body: "not secret",
742            },
743        );
744        harness.assert_route("secret-unpriv", right_route).await;
745
746        let body = assert_status_eventually(harness.addr, "/secret", StatusCode::OK).await;
747        assert_eq!(body, "not secret");
748    }
749
750    #[tokio::test]
751    async fn overlapping_routes_do_not_crash_server() {
752        let harness = setup_test_harness(EndpointType::Unprivileged).await;
753
754        // Assert a route at /health with identifier "health-1".
755        let route_1 = DynamicRoute::http(
756            EndpointType::Unprivileged,
757            SimpleHandler {
758                path: "/health",
759                body: "health-1",
760            },
761        );
762        harness.assert_route("health-1", route_1).await;
763        let body = assert_status_eventually(harness.addr, "/health", StatusCode::OK).await;
764        assert_eq!(body, "health-1");
765
766        // Assert a DIFFERENT identifier with the SAME path/method. Previously this caused a panic
767        // in rebuild_router. The server should remain alive with first-writer-wins semantics.
768        let route_2 = DynamicRoute::http(
769            EndpointType::Unprivileged,
770            SimpleHandler {
771                path: "/health",
772                body: "health-2",
773            },
774        );
775        harness.assert_route("health-2", route_2).await;
776
777        // Give the event loop time to process and rebuild.
778        sleep(Duration::from_millis(200)).await;
779
780        // Server is still alive; first handler wins.
781        let (status, body) = http_get(harness.addr, "/health").await;
782        assert_eq!(status, StatusCode::OK);
783        assert_eq!(body, "health-1");
784
785        // Non-overlapping routes are unaffected.
786        let route_info = DynamicRoute::http(
787            EndpointType::Unprivileged,
788            SimpleHandler {
789                path: "/info",
790                body: "info",
791            },
792        );
793        harness.assert_route("info", route_info).await;
794        let body = assert_status_eventually(harness.addr, "/info", StatusCode::OK).await;
795        assert_eq!(body, "info");
796
797        // Retract the first /health handler -- the previously skipped second handler should now
798        // become active since the conflict no longer exists.
799        harness.retract_route("health-1").await;
800        let body = assert_status_eventually(harness.addr, "/health", StatusCode::OK).await;
801        assert_eq!(body, "health-2");
802    }
803
804    #[tokio::test]
805    async fn overlapping_route_retraction_then_reassertion() {
806        let harness = setup_test_harness(EndpointType::Unprivileged).await;
807
808        // Assert two overlapping handlers.
809        let route_a = DynamicRoute::http(
810            EndpointType::Unprivileged,
811            SimpleHandler {
812                path: "/overlap",
813                body: "a",
814            },
815        );
816        let route_b = DynamicRoute::http(
817            EndpointType::Unprivileged,
818            SimpleHandler {
819                path: "/overlap",
820                body: "b",
821            },
822        );
823        harness.assert_route("ov-a", route_a).await;
824        harness.assert_route("ov-b", route_b).await;
825
826        // Server alive; first writer wins.
827        let body = assert_status_eventually(harness.addr, "/overlap", StatusCode::OK).await;
828        assert_eq!(body, "a");
829
830        // Retract both.
831        harness.retract_route("ov-a").await;
832        harness.retract_route("ov-b").await;
833        assert_status_eventually(harness.addr, "/overlap", StatusCode::NOT_FOUND).await;
834
835        // Re-assert a single handler -- should work cleanly.
836        let route_c = DynamicRoute::http(
837            EndpointType::Unprivileged,
838            SimpleHandler {
839                path: "/overlap",
840                body: "c",
841            },
842        );
843        harness.assert_route("ov-c", route_c).await;
844        let body = assert_status_eventually(harness.addr, "/overlap", StatusCode::OK).await;
845        assert_eq!(body, "c");
846    }
847
848    #[tokio::test]
849    async fn static_handler_served_without_dynamic_routes() {
850        let harness = setup_test_harness_with(EndpointType::Unprivileged, |b| {
851            b.with_handler(SimpleHandler {
852                path: "/static",
853                body: "static",
854            })
855        })
856        .await;
857
858        let body = assert_status_eventually(harness.addr, "/static", StatusCode::OK).await;
859        assert_eq!(body, "static");
860    }
861
862    #[tokio::test]
863    async fn static_and_dynamic_routes_coexist() {
864        let harness = setup_test_harness_with(EndpointType::Unprivileged, |b| {
865            b.with_handler(SimpleHandler {
866                path: "/static",
867                body: "static",
868            })
869        })
870        .await;
871
872        // Static route is served immediately.
873        let body = assert_status_eventually(harness.addr, "/static", StatusCode::OK).await;
874        assert_eq!(body, "static");
875
876        // Add a dynamic route on a different path -- both should serve.
877        let dynamic_route = DynamicRoute::http(
878            EndpointType::Unprivileged,
879            SimpleHandler {
880                path: "/dynamic",
881                body: "dynamic",
882            },
883        );
884        harness.assert_route("dyn", dynamic_route).await;
885
886        let body = assert_status_eventually(harness.addr, "/dynamic", StatusCode::OK).await;
887        assert_eq!(body, "dynamic");
888
889        let (status, body) = http_get(harness.addr, "/static").await;
890        assert_eq!(status, StatusCode::OK);
891        assert_eq!(body, "static");
892
893        // Retracting the dynamic route leaves the static route untouched.
894        harness.retract_route("dyn").await;
895        assert_status_eventually(harness.addr, "/dynamic", StatusCode::NOT_FOUND).await;
896
897        let (status, body) = http_get(harness.addr, "/static").await;
898        assert_eq!(status, StatusCode::OK);
899        assert_eq!(body, "static");
900    }
901
902    #[tokio::test]
903    async fn unknown_grpc_method_returns_unimplemented() {
904        let harness = setup_test_harness(EndpointType::Unprivileged).await;
905        let (status, headers) = grpc_post(harness.addr, "/some.Service/Method").await;
906
907        // gRPC errors are reported with HTTP 200 plus a `grpc-status` header. UNIMPLEMENTED is code 12.
908        assert_eq!(status, StatusCode::OK);
909        let grpc_status = headers.get("grpc-status").and_then(|v| v.to_str().ok());
910        assert_eq!(grpc_status, Some("12"));
911    }
912
913    #[tokio::test]
914    async fn static_route_wins_overlap_with_dynamic() {
915        let harness = setup_test_harness_with(EndpointType::Unprivileged, |b| {
916            b.with_handler(SimpleHandler {
917                path: "/overlap",
918                body: "static",
919            })
920        })
921        .await;
922
923        // Static route is served.
924        let body = assert_status_eventually(harness.addr, "/overlap", StatusCode::OK).await;
925        assert_eq!(body, "static");
926
927        // Asserting a dynamic route at the same path is skipped due to overlap -- static still wins.
928        let dynamic_route = DynamicRoute::http(
929            EndpointType::Unprivileged,
930            SimpleHandler {
931                path: "/overlap",
932                body: "dynamic",
933            },
934        );
935        harness.assert_route("dyn-overlap", dynamic_route).await;
936
937        sleep(Duration::from_millis(200)).await;
938
939        let (status, body) = http_get(harness.addr, "/overlap").await;
940        assert_eq!(status, StatusCode::OK);
941        assert_eq!(body, "static");
942    }
943}
saluki_app/dynamic_api.rs

saluki_app/
dynamic_api.rs
