base_http_server/
lib.rs

1// Released under the MIT License.
2// Copyright, 2026, by Samuel Williams.
3
4//! Application handlers and supervised HTTP/1 connections over Base scheduler
5//! capabilities. All connection futures are owned and joined by the server.
6#![doc = include_str!("../readme.md")]
7#![forbid(unsafe_code)]
8
9use base_endpoint::stream::{ReadableStream, WritableStream};
10use base_http::{
11    InterimHandler, InterimResponses, Request, Response, Scheme, Status, body::ReadableBody,
12};
13use base_http1::{RequestReader, ResponseWriter};
14use base_scheduler::{
15    Cancellation, SocketAccept, SocketClose, SocketSetOption, SocketStreamReading,
16    SocketStreamWriting, TcpNoDelay,
17};
18use futures_util::{
19    future::{Either, select},
20    stream::{FuturesUnordered, StreamExt},
21};
22use std::{
23    fmt,
24    future::{Future, poll_fn},
25    pin::pin,
26    task::Poll,
27};
28
29/// Generic bodies and hooks keep handlers independent of the HTTP wire version
30/// and scheduler. A returned body may retain the request body, including borrows.
31pub trait Handler: Sync {
32    type Error: std::error::Error + Send + Sync + 'static;
33    type Body<B: ReadableBody>: ReadableBody;
34
35    fn call<B: ReadableBody, I: InterimHandler>(
36        &self,
37        request: Request<B>,
38        interim: &mut InterimResponses<I>,
39        cancellation: &Cancellation,
40    ) -> impl Future<Output = Result<Response<Self::Body<B>>, Self::Error>> + Send;
41}
42
43#[derive(Clone, Debug)]
44pub struct Options {
45    pub http1: base_http1::Options,
46    /// HTTP engine selection, shared by plain and encrypted establishment.
47    pub protocols: base_http_protocol::HttpProtocols,
48    pub max_connections: usize,
49    /// Disable Nagle's algorithm on each accepted connection, so small writes
50    /// such as streamed frames are not held back waiting for acknowledgement.
51    /// This is a best-effort latency hint: a socket that rejects it is still
52    /// served.
53    pub tcp_nodelay: bool,
54}
55impl Default for Options {
56    fn default() -> Self {
57        Self {
58            http1: base_http1::Options::default(),
59            protocols: base_http_protocol::HttpProtocols::default(),
60            max_connections: 256,
61            tcp_nodelay: true,
62        }
63    }
64}
65
66/// A server failure. Protocol and negotiation errors are transparent; other wrapping variants describe this layer and report their cause as the source, so a cause chain never repeats a message.
67#[derive(Debug)]
68pub enum Error {
69    Protocol(base_http1::Error),
70    Handler(Box<dyn std::error::Error + Send + Sync>),
71    Io(std::io::Error),
72    Cancelled,
73    Negotiation(base_http_protocol::NegotiationError),
74}
75impl fmt::Display for Error {
76    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
77        match self {
78            Self::Protocol(e) => e.fmt(f),
79            Self::Handler(_) => f.write_str("request handler failed"),
80            Self::Io(_) => f.write_str("HTTP server I/O failed"),
81            Self::Cancelled => f.write_str("HTTP server cancelled"),
82            Self::Negotiation(error) => error.fmt(f),
83        }
84    }
85}
86impl std::error::Error for Error {
87    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
88        match self {
89            Self::Protocol(e) => e.source(),
90            Self::Handler(e) => Some(e.as_ref()),
91            Self::Io(e) => Some(e),
92            Self::Cancelled => None,
93            Self::Negotiation(error) => error.source(),
94        }
95    }
96}
97impl From<base_http1::Error> for Error {
98    fn from(e: base_http1::Error) -> Self {
99        match e {
100            base_http1::Error::Cancelled => Self::Cancelled,
101            e => Self::Protocol(e),
102        }
103    }
104}
105impl From<std::io::Error> for Error {
106    fn from(e: std::io::Error) -> Self {
107        Self::Io(e)
108    }
109}
110impl From<base_scheduler::OperationError> for Error {
111    fn from(error: base_scheduler::OperationError) -> Self {
112        base_http1::Error::from(error).into()
113    }
114}
115
116async fn cancellable<F: Future>(
117    future: F,
118    cancellation: &Cancellation,
119) -> Result<F::Output, Error> {
120    cancellation.check().map_err(|_| Error::Cancelled)?;
121    if cancellation.is_never() {
122        return Ok(future.await);
123    }
124    let mut future = pin!(future);
125    let mut cancelled = pin!(cancellation.cancelled());
126    poll_fn(|cx| {
127        if cancelled.as_mut().poll(cx).is_ready() {
128            return Poll::Ready(Err(Error::Cancelled));
129        }
130        future.as_mut().poll(cx).map(Ok)
131    })
132    .await
133}
134
135/// Own one accepted socket through request admission, handler execution, output,
136/// uncancellable lane retirement, and finally exclusive socket closure. Await
137/// completion after cooperative cancellation; dropping this future is not a join.
138pub async fn serve_connection<S, H>(
139    scheduler: &S,
140    socket: S::Socket,
141    handler: &H,
142    options: &base_http1::Options,
143    cancellation: &Cancellation,
144) -> Result<u64, Error>
145where
146    S: SocketStreamReading + SocketStreamWriting + SocketClose,
147    S::Socket: 'static,
148    H: Handler,
149{
150    let outcome = connection_outcome(scheduler, socket, handler, options, cancellation).await;
151    outcome.result.map(|()| outcome.requests)
152}
153
154struct Outcome {
155    requests: u64,
156    result: Result<(), Error>,
157}
158
159async fn connection_outcome<S, H>(
160    scheduler: &S,
161    socket: S::Socket,
162    handler: &H,
163    options: &base_http1::Options,
164    cancellation: &Cancellation,
165) -> Outcome
166where
167    S: SocketStreamReading + SocketStreamWriting + SocketClose,
168    S::Socket: 'static,
169    H: Handler,
170{
171    let mut requests = 0;
172    // The outer owner closes the socket even when capability construction fails.
173    let result = async {
174        options.validate()?;
175        let mut reader = scheduler.socket_stream_reader(&socket)?;
176        let writer = match scheduler.socket_stream_writer(&socket) {
177            Ok(writer) => writer,
178            Err(error) => {
179                // Report the construction failure in preference to a retirement failure.
180                let _ = ReadableStream::close(&mut reader).await;
181                return Err(Error::Io(error));
182            }
183        };
184        // The plaintext engine is HTTP/1, whose codecs fail to construct only when `options.validate` above fails.
185        let (input, output) = base_http_protocol::HttpProtocols::default()
186            .plaintext()
187            .server(reader, writer, options.clone())
188            .expect("validated HTTP/1 options construct server codecs");
189        serve_codecs(
190            input,
191            output,
192            handler,
193            cancellation,
194            &mut requests,
195            &Scheme::HTTP,
196        )
197        .await
198    }
199    .await;
200    let closed = scheduler.socket_close(socket).await;
201    Outcome {
202        requests,
203        result: result.and_then(|()| Ok(closed?)),
204    }
205}
206
207/// Serve exchanges over established codecs, then retire both directions.
208async fn serve_codecs<R, W, H>(
209    mut input: RequestReader<R>,
210    mut output: ResponseWriter<W>,
211    handler: &H,
212    cancellation: &Cancellation,
213    count: &mut u64,
214    scheme: &Scheme,
215) -> Result<(), Error>
216where
217    R: ReadableStream,
218    W: WritableStream,
219    H: Handler,
220{
221    let result = exchanges(
222        &mut input,
223        &mut output,
224        handler,
225        cancellation,
226        count,
227        scheme,
228    )
229    .await;
230    // Both cleanups run regardless of the exchange or the other cleanup result.
231    let input_closed = input.close().await;
232    let output_closed = output.close().await;
233    result?;
234    input_closed?;
235    output_closed?;
236    Ok(())
237}
238
239async fn exchanges<R, W, H>(
240    input: &mut RequestReader<R>,
241    output: &mut ResponseWriter<W>,
242    handler: &H,
243    cancellation: &Cancellation,
244    count: &mut u64,
245    scheme: &Scheme,
246) -> Result<(), Error>
247where
248    R: ReadableStream,
249    W: WritableStream,
250    H: Handler,
251{
252    loop {
253        let (mut request, exchange) = match input.read_request(cancellation).await {
254            Ok(Some(pair)) => pair,
255            Ok(None) => return Ok(()),
256            Err(error) => {
257                if !matches!(error, base_http1::Error::Cancelled) {
258                    let status = if matches!(error, base_http1::Error::Limit(_)) {
259                        Status::new(431).unwrap()
260                    } else {
261                        Status::BAD_REQUEST
262                    };
263                    // A bad request cannot be reused even if rejection succeeds.
264                    let _ = output.reject(status, cancellation).await;
265                }
266                return Err(error.into());
267            }
268        };
269        request.scheme.get_or_insert_with(|| scheme.clone());
270        let result = {
271            let mut interim = InterimResponses::new(output.interim(exchange.version()));
272            let result = {
273                let called = pin!(handler.call(request, &mut interim, cancellation));
274                cancellable(called, cancellation).await?
275            };
276            interim.finish();
277            result
278        };
279        let mut response = result.map_err(|error| {
280            // Cancellation can arrive between polling the outer signal and
281            // polling a handler which propagates a cancelled body operation.
282            if cancellation.is_cancelled() {
283                Error::Cancelled
284            } else {
285                Error::Handler(Box::new(error))
286            }
287        })?;
288        let persistent = output
289            .write_response(&exchange, &mut response, cancellation)
290            .await?;
291        drop(response);
292        *count += 1;
293        if !persistent || !input.is_reusable() {
294            return Ok(());
295        }
296    }
297}
298
299#[derive(Default, Debug)]
300pub struct Statistics {
301    pub connections: u64,
302    pub requests: u64,
303    pub failed_connections: u64,
304}
305
306/// Serve an already-bound listener with bounded concurrent connection futures.
307/// No hidden tasks or detached jobs are created. Cancellation stops acceptance,
308/// cancels active exchanges, joins retirement, and closes the owned listener.
309/// Individual peer/handler failures are counted without stopping other peers.
310pub async fn serve<S, H>(
311    scheduler: &S,
312    listener: S::Socket,
313    handler: &H,
314    options: &Options,
315    cancellation: &Cancellation,
316) -> Result<Statistics, Error>
317where
318    S: SocketAccept
319        + SocketSetOption<TcpNoDelay>
320        + SocketStreamReading
321        + SocketStreamWriting
322        + SocketClose,
323    S::Socket: 'static,
324    H: Handler,
325{
326    serve_with(
327        scheduler,
328        listener,
329        options,
330        cancellation,
331        |socket, cancellation| async move {
332            connection_outcome(scheduler, socket, handler, &options.http1, &cancellation).await
333        },
334    )
335    .await
336}
337
338async fn serve_with<S, F, Fut>(
339    scheduler: &S,
340    listener: S::Socket,
341    options: &Options,
342    cancellation: &Cancellation,
343    connection: F,
344) -> Result<Statistics, Error>
345where
346    S: SocketAccept + SocketSetOption<TcpNoDelay> + SocketClose,
347    F: Fn(S::Socket, Cancellation) -> Fut,
348    Fut: Future<Output = Outcome>,
349{
350    let connections_cancel = cancellation.child();
351    let mut connections = FuturesUnordered::new();
352    let mut statistics = Statistics::default();
353    let result = async {
354        options.http1.validate()?;
355        if options.max_connections == 0 {
356            return Err(Error::Protocol(base_http1::Error::Invalid(
357                "zero connection limit",
358            )));
359        }
360        loop {
361            if connections.len() == options.max_connections {
362                record(
363                    &mut statistics,
364                    cancellable(connections.next(), cancellation)
365                        .await?
366                        .unwrap(),
367                );
368                continue;
369            }
370            // Keep this accept alive while existing connections complete. A
371            // backend may already own an accepted socket between polls; dropping
372            // and recreating acceptance can silently discard that new peer.
373            let accepted = {
374                let mut accept = pin!(scheduler.socket_accept(&listener, cancellation));
375                loop {
376                    let completed = pin!(async {
377                        if connections.is_empty() {
378                            std::future::pending().await
379                        } else {
380                            connections.next().await.unwrap()
381                        }
382                    });
383                    match select(accept.as_mut(), completed).await {
384                        Either::Left((result, _)) => break result,
385                        Either::Right((result, _)) => record(&mut statistics, result),
386                    }
387                }
388            };
389            let (socket, _) = accepted.map_err(base_http1::Error::from)?;
390            statistics.connections += 1;
391            let cancellation = connections_cancel.clone();
392            connections.push(async {
393                if options.tcp_nodelay {
394                    // A latency hint only: the connection is served without it.
395                    let _ = scheduler
396                        .socket_set_option(&socket, TcpNoDelay, true, &cancellation)
397                        .await;
398                }
399                connection(socket, cancellation).await
400            });
401        }
402    }
403    .await;
404    connections_cancel.cancel();
405    while let Some(result) = connections.next().await {
406        record(&mut statistics, result);
407    }
408    let closed = scheduler.socket_close(listener).await;
409    closed?;
410    match result {
411        Ok(()) | Err(Error::Cancelled) => Ok(statistics),
412        Err(error) => Err(error),
413    }
414}
415
416fn record(statistics: &mut Statistics, outcome: Outcome) {
417    statistics.requests += outcome.requests;
418    match outcome.result {
419        Ok(()) => {}
420        Err(Error::Cancelled) => {}
421        Err(_) => statistics.failed_connections += 1,
422    }
423}
424
425/// Serve HTTP/1 over an already authenticated TLS connection. ALPN is
426/// interpreted by HTTP before any request is admitted.
427#[cfg(feature = "rustls")]
428pub async fn serve_tls_connection<S, H>(
429    scheduler: &S,
430    connection: base_endpoint::tls::rustls::RustlsConnection<S::Socket>,
431    handler: &H,
432    options: &Options,
433    cancellation: &Cancellation,
434) -> Result<u64, Error>
435where
436    S: SocketStreamReading + SocketStreamWriting + SocketClose,
437    S::Socket: 'static,
438    H: Handler,
439{
440    let outcome =
441        tls_connection_outcome(scheduler, connection, handler, options, cancellation).await;
442    outcome.result.map(|()| outcome.requests)
443}
444
445#[cfg(feature = "rustls")]
446async fn tls_connection_outcome<S, H>(
447    scheduler: &S,
448    connection: base_endpoint::tls::rustls::RustlsConnection<S::Socket>,
449    handler: &H,
450    options: &Options,
451    cancellation: &Cancellation,
452) -> Outcome
453where
454    S: SocketStreamReading + SocketStreamWriting + SocketClose,
455    S::Socket: 'static,
456    H: Handler,
457{
458    use base_endpoint::tls::TlsSession;
459    let selected = connection.negotiated_protocol().map(<[u8]>::to_vec);
460    let mut requests = 0;
461    let result = connection
462        .with_duplex(scheduler, cancellation, |reader, writer| async {
463            let protocol = options.protocols.negotiated(selected.as_deref())?;
464            let (input, output) = protocol.server(reader, writer, options.http1.clone())?;
465            serve_codecs(
466                input,
467                output,
468                handler,
469                cancellation,
470                &mut requests,
471                &Scheme::HTTPS,
472            )
473            .await
474        })
475        .await;
476    Outcome { requests, result }
477}
478
479/// Configure HTTP/1 ALPN and serve TLS on an owned TCP listener. Handshakes
480/// occupy bounded connection slots and progress concurrently with acceptance.
481#[cfg(feature = "rustls")]
482pub async fn serve_tls<S, H>(
483    scheduler: &S,
484    listener: S::Socket,
485    config: std::sync::Arc<rustls::ServerConfig>,
486    handler: &H,
487    options: &Options,
488    cancellation: &Cancellation,
489) -> Result<Statistics, Error>
490where
491    S: SocketAccept
492        + SocketSetOption<TcpNoDelay>
493        + SocketStreamReading
494        + SocketStreamWriting
495        + SocketClose,
496    S::Socket: 'static,
497    H: Handler,
498{
499    use base_endpoint::tls::{TlsAcceptor, rustls::RustlsAcceptor};
500    let mut config = (*config).clone();
501    config.alpn_protocols = options.protocols.alpn_protocols();
502    let acceptor = RustlsAcceptor::new(std::sync::Arc::new(config));
503    serve_with(
504        scheduler,
505        listener,
506        options,
507        cancellation,
508        |socket, cancellation| {
509            let acceptor = &acceptor;
510            async move {
511                match acceptor.accept(scheduler, socket, &cancellation).await {
512                    Ok(connection) => {
513                        tls_connection_outcome(
514                            scheduler,
515                            connection,
516                            handler,
517                            options,
518                            &cancellation,
519                        )
520                        .await
521                    }
522                    Err(error) => Outcome {
523                        requests: 0,
524                        result: Err(error.into()),
525                    },
526                }
527            }
528        },
529    )
530    .await
531}
532
533impl From<base_http_protocol::NegotiationError> for Error {
534    fn from(error: base_http_protocol::NegotiationError) -> Self {
535        Self::Negotiation(error)
536    }
537}
538
539#[cfg(test)]
540mod tests {
541    use super::*;
542    use std::{error::Error as _, io};
543
544    #[test]
545    fn protocol_errors_are_transparent_and_io_errors_have_context() {
546        let error = Error::from(base_http1::Error::Io(io::Error::from(
547            io::ErrorKind::ConnectionReset,
548        )));
549        assert_eq!(error.to_string(), "HTTP/1 connection I/O failed");
550        assert_eq!(error.source().unwrap().to_string(), "connection reset");
551
552        let error = Error::from(io::Error::from(io::ErrorKind::ConnectionReset));
553        assert_eq!(error.to_string(), "HTTP server I/O failed");
554        assert_eq!(error.source().unwrap().to_string(), "connection reset");
555    }
556
557    #[test]
558    fn handler_errors_are_reported_as_the_source() {
559        let error = Error::Handler(Box::new(io::Error::other("database unavailable")));
560        assert_eq!(error.to_string(), "request handler failed");
561        assert_eq!(error.source().unwrap().to_string(), "database unavailable");
562    }
563
564    #[test]
565    fn scheduler_operation_errors_keep_cancellation_distinct() {
566        let error = Error::from(base_scheduler::OperationError::from(
567            base_scheduler::Cancelled,
568        ));
569        assert_eq!(format!("{error:?}"), "Cancelled");
570        assert_eq!(error.to_string(), "HTTP server cancelled");
571        assert!(error.source().is_none());
572
573        let error = Error::from(base_scheduler::OperationError::from(io::Error::from(
574            io::ErrorKind::ConnectionReset,
575        )));
576        let expected = Error::Protocol(base_http1::Error::Io(io::Error::from(
577            io::ErrorKind::ConnectionReset,
578        )));
579        assert_eq!(format!("{error:?}"), format!("{expected:?}"));
580        assert_eq!(error.source().unwrap().to_string(), "connection reset");
581    }
582
583    #[test]
584    fn negotiation_errors_are_transparent() {
585        let error = Error::from(base_http_protocol::NegotiationError::Missing);
586        assert_eq!(format!("{error:?}"), "Negotiation(Missing)");
587        assert_eq!(
588            error.to_string(),
589            "HTTP application protocol was not negotiated"
590        );
591        assert!(error.source().is_none());
592    }
593}
base_http_server/lib.rs

base_http_server/
lib.rs
