saluki_io/net/server/
http.rs

1//! HTTP servers.
2//!
3//! [`HttpServer`] is the supervised server, and is what new code should use: it runs as a child of whatever supervisor
4//! it is added to, binds its listener during initialization, and drains in-flight connections before it reports being
5//! done. [`UnsupervisedHttpServer`] is the older, self-spawning form, kept only until its remaining callers move over.
6
7use std::{
8    future::Future,
9    pin::Pin,
10    sync::Arc,
11    task::{ready, Context, Poll},
12    time::Duration,
13};
14
15use async_trait::async_trait;
16use http::{Request, Response};
17use http_body::Body;
18use hyper::{
19    body::Incoming,
20    rt::{Read, Write},
21    service::Service,
22};
23use hyper_util::{
24    rt::{TokioExecutor, TokioIo, TokioTimer},
25    server::conn::auto::Builder,
26};
27use rustls::ServerConfig;
28use saluki_common::{
29    sync::shutdown::{ShutdownCoordinator, ShutdownHandle},
30    task::{spawn_traced_named, HandleExt as _},
31};
32use saluki_core::runtime::{InitializationError, ShutdownStrategy, Supervisable, SupervisorFuture};
33use saluki_error::{ErrorContext as _, GenericError};
34use saluki_tls::ensure_server_config_fips_compliant;
35use tokio::{pin, runtime::Handle, select, sync::oneshot, time::timeout};
36use tokio_rustls::TlsAcceptor;
37use tracing::{debug, error, info, warn};
38
39use crate::net::{listener::ConnectionOrientedListener, ListenAddress};
40
41fn build_conn_builder() -> Builder<TokioExecutor> {
42    let mut builder = Builder::new(TokioExecutor::new());
43    builder
44        .http1()
45        .timer(TokioTimer::new())
46        .header_read_timeout(Duration::from_secs(10));
47    builder
48}
49
50/// An HTTP server.
51///
52/// Serves a single [`Service`] over a connection-oriented listener, optionally with TLS. The server can't be run
53/// directly: it is only usable by adding it to a supervisor.
54///
55/// # Supervision
56///
57/// The listen address is bound during initialization, so a failure to bind is raised before the supervised worker
58/// starts running, and a restart rebinds.
59///
60/// The server will attempt to gracefully shutdown existing connections when the parent supervisor signals shutdown.
61/// This will cause the worker to utilize the maximum allowable grace period during shutdown: it will attempt to take as
62/// long as necessary to gracefully shutdown existing connections, bounded only by the parent supervisor.
63pub struct HttpServer<S> {
64    listen_address: ListenAddress,
65    tls_config: Option<ServerConfig>,
66    conn_builder: Builder<TokioExecutor>,
67    graceful_shutdown_timeout: Option<Duration>,
68    service: S,
69}
70
71impl<S> HttpServer<S> {
72    /// Creates a server that will listen on the given address.
73    pub fn from_listen_address(listen_address: ListenAddress, service: S) -> Self {
74        Self {
75            listen_address,
76            tls_config: None,
77            conn_builder: build_conn_builder(),
78            graceful_shutdown_timeout: None,
79            service,
80        }
81    }
82
83    /// Sets the graceful shutdown timeout for this server.
84    ///
85    /// During shutdown, the server will for all in-flight connections to complete before ultimately completing itself.
86    /// When no timeout is specified, this will lead to the worker taking the maximum allowable time to shutdown if
87    /// connections are blocked or otherwise "stuck." Setting an explicit graceful shutdown timeout will cause the
88    /// worker to bound how long it waits for in-flight connections to shutdown before forcefully completing and moving
89    /// on.
90    ///
91    /// Defaults to no timeout (wait as long as allowed).
92    pub fn with_graceful_shutdown_timeout(mut self, timeout: Duration) -> Self {
93        self.graceful_shutdown_timeout = Some(timeout);
94        self
95    }
96
97    /// Sets the TLS configuration for the server.
98    ///
99    /// This enables TLS, after which the server only accepts connections that are encrypted with TLS.
100    ///
101    /// Defaults to TLS being disabled.
102    pub fn with_tls_config(mut self, config: ServerConfig) -> Self {
103        self.tls_config = Some(config);
104        self
105    }
106}
107
108#[async_trait]
109impl<S, B> Supervisable for HttpServer<S>
110where
111    S: Service<Request<Incoming>, Response = Response<B>> + Send + Sync + Clone + 'static,
112    S::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
113    S::Future: Send + 'static,
114    B: Body + Send + 'static,
115    B::Data: Send,
116    B::Error: std::error::Error + Send + Sync,
117{
118    fn name(&self) -> &str {
119        "http_server"
120    }
121
122    fn shutdown_strategy(&self) -> ShutdownStrategy {
123        // Utilize the maximum allowable grace period to give connections a chance to gracefully shutdown.
124        ShutdownStrategy::Graceful(Duration::MAX)
125    }
126
127    async fn initialize(&self, process_shutdown: ShutdownHandle) -> Result<SupervisorFuture, InitializationError> {
128        // Try binding our listener during initialization to surface issues earlier.
129        let listener = ConnectionOrientedListener::from_listen_address(self.listen_address.clone())
130            .await
131            .with_error_context(|| format!("Failed to bind listener for HTTP server ({}).", self.listen_address))?;
132
133        let conn_builder = self.conn_builder.clone();
134        let service = self.service.clone();
135        let tls_config = self.tls_config.clone();
136        let maybe_shutdown_timeout = self.graceful_shutdown_timeout;
137
138        // Connection handlers land on whichever runtime the supervisor placed this worker on, so nothing needs to be
139        // threaded through: where the server runs is decided at the point it's spawned.
140        //
141        // TODO: Create our own custom `Executor` impl that can be used to bridge to a given supervisor such that we
142        // spawn dynamic/temporary child workers instead of directly on the underlying Tokio runtime.
143        let executor = Handle::current();
144
145        Ok(Box::pin(run_accept_loop(
146            listener,
147            conn_builder,
148            service,
149            tls_config,
150            executor,
151            process_shutdown,
152            maybe_shutdown_timeout,
153        )))
154    }
155}
156
157/// An HTTP server that spawns and manages itself.
158///
159/// # Deprecated
160///
161/// Callers should generally prefer to use [`HttpServer`], as it is designed to run under supervision and play nicely
162/// with supervision trees: graceful shutdown, spawning of connection handlers in the right place, etc.
163pub struct UnsupervisedHttpServer<S> {
164    listener: ConnectionOrientedListener,
165    tls_config: Option<ServerConfig>,
166    conn_builder: Builder<TokioExecutor>,
167    executor: Handle,
168    service: S,
169}
170
171impl<S> UnsupervisedHttpServer<S> {
172    /// Creates a new `UnsupervisedHttpServer` from the given listener and service.
173    ///
174    /// # Panics
175    ///
176    /// This will panic if called outside the context of a Tokio runtime.
177    pub fn from_listener(listener: ConnectionOrientedListener, service: S) -> Self {
178        Self {
179            listener,
180            tls_config: None,
181            conn_builder: build_conn_builder(),
182            executor: Handle::current(),
183            service,
184        }
185    }
186
187    /// Sets the TLS configuration for the server.
188    ///
189    /// This will enable TLS for the server, and the server will only accept connections that are encrypted with TLS.
190    ///
191    /// Defaults to TLS being disabled.
192    pub fn with_tls_config(mut self, config: ServerConfig) -> Self {
193        self.tls_config = Some(config);
194        self
195    }
196
197    /// Sets the executor for the server.
198    ///
199    /// This executor will be used for spawning tasks to handle incoming connections, but _not_ for the spawn that accepts
200    /// new connections.
201    ///
202    /// Defaults to the current Tokio runtime at the time [`from_listener`][Self::from_listener] is called.
203    pub fn with_executor(mut self, executor: Handle) -> Self {
204        self.executor = executor;
205        self
206    }
207}
208
209impl<S, B> UnsupervisedHttpServer<S>
210where
211    S: Service<Request<Incoming>, Response = Response<B>> + Send + Clone + 'static,
212    S::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
213    S::Future: Send + 'static,
214    B: Body + Send + 'static,
215    B::Data: Send,
216    B::Error: std::error::Error + Send + Sync,
217{
218    /// Starts the server and listens for incoming connections.
219    ///
220    /// Returns two handles: one for shutting down the server, and one for receiving any errors that occur while the
221    /// server is running.
222    pub fn listen(self) -> (ShutdownCoordinator, ErrorHandle) {
223        let (shutdown_coordinator, shutdown) = ShutdownHandle::paired();
224        let (error_tx, error_rx) = oneshot::channel();
225
226        let Self {
227            executor,
228            listener,
229            conn_builder,
230            service,
231            tls_config,
232        } = self;
233
234        spawn_traced_named("http-server-acceptor", async move {
235            if let Err(e) = run_accept_loop(listener, conn_builder, service, tls_config, executor, shutdown, None).await
236            {
237                let _ = error_tx.send(e);
238            }
239        });
240
241        (shutdown_coordinator, ErrorHandle(error_rx))
242    }
243}
244
245/// Accepts connections until shutdown is signalled or the listener fails.
246///
247/// Returns once every connection it accepted has finished, so a caller that awaits this can be sure no request is still
248/// being served.
249async fn run_accept_loop<S, B>(
250    mut listener: ConnectionOrientedListener, conn_builder: Builder<TokioExecutor>, service: S,
251    tls_config: Option<ServerConfig>, executor: Handle, shutdown: ShutdownHandle,
252    maybe_shutdown_timeout: Option<Duration>,
253) -> Result<(), GenericError>
254where
255    S: Service<Request<Incoming>, Response = Response<B>> + Send + Clone + 'static,
256    S::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
257    S::Future: Send + 'static,
258    B: Body + Send + 'static,
259    B::Data: Send,
260    B::Error: std::error::Error + Send + Sync,
261{
262    let maybe_tls_acceptor = match tls_config {
263        Some(mut config) => {
264            // Allow for HTTP/1.1 and HTTP/2.
265            config.alpn_protocols.push(b"h2".to_vec());
266            config.alpn_protocols.push(b"http/1.1".to_vec());
267
268            ensure_server_config_fips_compliant(&mut config)?;
269
270            Some(TlsAcceptor::from(Arc::new(config)))
271        }
272        None => None,
273    };
274    let tls_enabled = maybe_tls_acceptor.is_some();
275    let listen_addr = listener.listen_address().clone();
276
277    info!(%listen_addr, tls_enabled, "HTTP server started.");
278
279    // Every connection handler holds a handle from this coordinator, which is what lets us wait for in-flight requests
280    // below instead of abandoning them.
281    let mut conn_shutdown_coordinator = ShutdownCoordinator::default();
282
283    pin!(shutdown);
284
285    let result = loop {
286        select! {
287            result = listener.accept() => match result {
288                Ok(stream) => {
289                    let conn_builder = conn_builder.clone();
290                    let service = service.clone();
291                    let listen_addr = listen_addr.clone();
292                    let conn_shutdown = conn_shutdown_coordinator.register();
293
294                    match &maybe_tls_acceptor {
295                        Some(acceptor) => {
296                            let tls_stream = match acceptor.accept(stream).await {
297                                Ok(stream) => stream,
298                                Err(e) => {
299                                    error!(%listen_addr, error = %e, "Failed to complete TLS handshake.");
300                                    continue
301                                },
302                            };
303
304                            executor.spawn_traced_named("http_server_tls_conn", drive_connection(
305                                conn_builder, TokioIo::new(tls_stream), service, listen_addr, conn_shutdown, maybe_shutdown_timeout
306                            ));
307                        },
308                        None => {
309                            executor.spawn_traced_named("http_server_conn", drive_connection(
310                                conn_builder, TokioIo::new(stream), service, listen_addr, conn_shutdown, maybe_shutdown_timeout
311                            ));
312                        },
313                    }
314                },
315                Err(e) => break Err(GenericError::from(e)),
316            },
317
318            _ = &mut shutdown => {
319                debug!(%listen_addr, "Received shutdown signal.");
320                break Ok(());
321            }
322        }
323    };
324
325    // We've stopped accepting; now let anything still being served finish before we report being done.
326    debug!(%listen_addr, "Waiting for in-flight HTTP connections to finish...");
327    conn_shutdown_coordinator.shutdown_and_wait().await;
328
329    info!(%listen_addr, "HTTP server stopped.");
330
331    result
332}
333
334/// Serves a single connection, finishing what it has started if asked to shut down.
335///
336/// When shutdown is triggered, the connection is gracefully shutdown: new requests aren't allowed, but any pending
337/// or in-flight reads/writes will be completed prior to closing the connection.
338async fn drive_connection<I, S, B>(
339    conn_builder: Builder<TokioExecutor>, io: I, service: S, listen_addr: ListenAddress, shutdown: ShutdownHandle,
340    maybe_shutdown_timeout: Option<Duration>,
341) where
342    I: Read + Write + Unpin + Send + 'static,
343    S: Service<Request<Incoming>, Response = Response<B>> + 'static,
344    S::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
345    S::Future: Send + 'static,
346    B: Body + Send + 'static,
347    B::Data: Send,
348    B::Error: std::error::Error + Send + Sync,
349{
350    let conn = conn_builder.serve_connection(io, service);
351    pin!(conn, shutdown);
352
353    select! {
354        result = conn.as_mut() => if let Err(e) = result {
355            error!(%listen_addr, error = %e, "Failed to serve HTTP connection.");
356        },
357
358        _ = &mut shutdown => {
359            debug!(%listen_addr, "Draining HTTP connection.");
360
361            conn.as_mut().graceful_shutdown();
362
363            let shutdown_timeout = maybe_shutdown_timeout.unwrap_or(Duration::MAX);
364            match timeout(shutdown_timeout, conn.as_mut()).await {
365                Ok(Ok(())) => {},
366                Ok(Err(e)) => warn!(%listen_addr, error = %e, "Failed to drain HTTP connection."),
367                Err(_) => warn!(%listen_addr, "Failed to gracefully drain HTTP connection after {:?}.", shutdown_timeout)
368            }
369        },
370    }
371}
372
373/// A future that resolves when [`UnsupervisedHttpServer`] encounters an unrecoverable error.
374pub struct ErrorHandle(oneshot::Receiver<GenericError>);
375
376impl Future for ErrorHandle {
377    type Output = Option<GenericError>;
378
379    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
380        match ready!(Pin::new(&mut self.0).poll(cx)) {
381            Ok(err) => Poll::Ready(Some(err)),
382            Err(_) => Poll::Ready(None),
383        }
384    }
385}
386
387#[cfg(test)]
388mod tests {
389    use std::net::{SocketAddr, TcpListener as StdTcpListener};
390    use std::sync::atomic::{AtomicBool, Ordering};
391
392    use http_body_util::Full;
393    use hyper::service::service_fn;
394    use saluki_common::sync::shutdown::ShutdownCoordinator;
395    use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
396    use tokio::net::TcpStream;
397    use tokio::time::timeout;
398
399    use super::*;
400
401    /// Bound on any server await in these tests, so a hang fails rather than stalling the suite.
402    const TEST_TIMEOUT: Duration = Duration::from_secs(5);
403
404    /// Reserves a loopback port and releases it, yielding an address a server can bind.
405    ///
406    /// Inherently racy against anything else on the host, but the window is small and there is no way to hand an
407    /// already-bound listener to the supervised server.
408    fn free_local_addr() -> SocketAddr {
409        let listener = StdTcpListener::bind("127.0.0.1:0").expect("should bind an ephemeral port");
410        let addr = listener.local_addr().expect("should have a local address");
411        drop(listener);
412        addr
413    }
414
415    /// Builds a server whose handler runs `f` for every request.
416    fn server_with<F, Fut>(
417        addr: SocketAddr, f: F,
418    ) -> HttpServer<
419        impl Service<
420                Request<Incoming>,
421                Response = Response<Full<bytes::Bytes>>,
422                Error = std::convert::Infallible,
423                Future = Fut,
424            > + Send
425            + Sync
426            + Clone
427            + 'static,
428    >
429    where
430        F: Fn() -> Fut + Send + Sync + Clone + 'static,
431        Fut: Future<Output = Result<Response<Full<bytes::Bytes>>, std::convert::Infallible>> + Send + 'static,
432    {
433        let service = service_fn(move |_req: Request<Incoming>| f());
434        HttpServer::from_listen_address(ListenAddress::Tcp(addr), service)
435    }
436
437    /// A handler that responds immediately.
438    async fn ok_response() -> Result<Response<Full<bytes::Bytes>>, std::convert::Infallible> {
439        Ok(Response::new(Full::new(bytes::Bytes::from_static(b"ok"))))
440    }
441
442    #[tokio::test]
443    async fn binds_during_initialization() {
444        // The listener is bound by `initialize`, not by the worker future, so the port is already taken before anything
445        // starts serving. That is what makes a bind failure a non-restartable initialization error.
446        let addr = free_local_addr();
447        let server = server_with(addr, ok_response);
448
449        let run = server
450            .initialize(ShutdownHandle::noop())
451            .await
452            .expect("should initialize");
453
454        assert!(
455            StdTcpListener::bind(addr).is_err(),
456            "initialization should have bound {addr} before the worker future ran"
457        );
458
459        drop(run);
460    }
461
462    #[tokio::test]
463    async fn bind_failure_is_an_initialization_error() {
464        // Hold the address so the server can't have it. An initialization error is non-restartable, which is the point:
465        // an unusable listen address should fail the child rather than being retried forever.
466        let addr = free_local_addr();
467        let _held = StdTcpListener::bind(addr).expect("should hold the address");
468
469        let server = server_with(addr, ok_response);
470        match server.initialize(ShutdownHandle::noop()).await {
471            Ok(_) => panic!("initialization should have failed to bind {addr}"),
472            Err(e) => {
473                let error = e.to_string();
474                assert!(error.contains("Failed to bind listener"), "unexpected error: {error}");
475            }
476        }
477    }
478
479    #[tokio::test]
480    async fn releases_its_port_once_the_worker_finishes() {
481        // The whole reason for supervising the server: when its worker stops, the socket is gone. Previously the
482        // acceptor was a detached task that outlived whatever spawned it.
483        let addr = free_local_addr();
484        let server = server_with(addr, ok_response);
485
486        let mut coordinator = ShutdownCoordinator::default();
487        let run = server
488            .initialize(coordinator.register())
489            .await
490            .expect("should initialize");
491
492        coordinator.shutdown();
493        timeout(TEST_TIMEOUT, run)
494            .await
495            .expect("server should stop on shutdown")
496            .expect("server should stop cleanly");
497
498        assert!(
499            StdTcpListener::bind(addr).is_ok(),
500            "the server should have released {addr} when its worker finished"
501        );
502    }
503
504    #[tokio::test]
505    async fn a_half_sent_request_does_not_wedge_the_drain() {
506        // A peer that writes a partial request head and stalls keeps its connection permanently non-idle, so
507        // `graceful_shutdown` alone never closes it. Before the connection builder had a timer and the drain had a
508        // deadline, one such socket stalled shutdown indefinitely -- for the OTLP receivers that meant every ADP
509        // shutdown hanging until the component budget forced an abort.
510        let addr = free_local_addr();
511        let server = server_with(addr, ok_response).with_graceful_shutdown_timeout(Duration::from_secs(1));
512
513        let mut coordinator = ShutdownCoordinator::default();
514        let run = server
515            .initialize(coordinator.register())
516            .await
517            .expect("should initialize");
518        let run = tokio::spawn(run);
519
520        let mut stream = TcpStream::connect(addr).await.expect("should connect");
521        stream
522            .write_all(b"GET / HTTP/1.1\r\nHost: localhost")
523            .await
524            .expect("should write a partial request head");
525        stream.flush().await.expect("should flush");
526
527        // Let the server read what there is before signalling, so the connection is genuinely mid-parse.
528        tokio::time::sleep(Duration::from_millis(100)).await;
529        coordinator.shutdown();
530
531        timeout(TEST_TIMEOUT, run)
532            .await
533            .expect("server should finish draining rather than waiting on a half-sent request")
534            .expect("server task should not panic")
535            .expect("server should stop cleanly");
536    }
537
538    #[tokio::test]
539    async fn does_not_finish_until_in_flight_requests_do() {
540        // Shutdown stops the server accepting, but a request already being served has to complete first. Without the
541        // connection drain, the worker future would return immediately and the response would be lost.
542        let addr = free_local_addr();
543
544        let handler_started = Arc::new(AtomicBool::new(false));
545        let started = Arc::clone(&handler_started);
546        let server = server_with(addr, move || {
547            let started = Arc::clone(&started);
548            async move {
549                started.store(true, Ordering::SeqCst);
550                tokio::time::sleep(Duration::from_millis(300)).await;
551                ok_response().await
552            }
553        });
554
555        let mut coordinator = ShutdownCoordinator::default();
556        let run = server
557            .initialize(coordinator.register())
558            .await
559            .expect("should initialize");
560        let mut run = tokio::spawn(run);
561
562        // Issue a request by hand rather than pulling in a client: all we need is for the handler to be running.
563        let mut stream = TcpStream::connect(addr).await.expect("should connect");
564        stream
565            .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")
566            .await
567            .expect("should write request");
568
569        while !handler_started.load(Ordering::SeqCst) {
570            tokio::time::sleep(Duration::from_millis(5)).await;
571        }
572
573        // Signal shutdown mid-request.
574        coordinator.shutdown();
575
576        // The handler is still working, so the worker must not be finished yet. This ordering is the actual assertion:
577        // without the drain the worker returns here and the response is abandoned to a detached task.
578        assert!(
579            timeout(Duration::from_millis(50), &mut run).await.is_err(),
580            "server should not finish while a request is still being served"
581        );
582
583        let mut response = Vec::new();
584        timeout(TEST_TIMEOUT, stream.read_to_end(&mut response))
585            .await
586            .expect("response should arrive")
587            .expect("should read response");
588        assert!(
589            response.ends_with(b"ok"),
590            "expected the in-flight response to complete, got {:?}",
591            String::from_utf8_lossy(&response)
592        );
593
594        timeout(TEST_TIMEOUT, &mut run)
595            .await
596            .expect("server should finish after draining")
597            .expect("server task should not panic")
598            .expect("server should stop cleanly");
599    }
600}
saluki_io/net/server/http.rs

saluki_io/net/server/
http.rs
