goose_http/conn/
mod.rs

1//! Connection state machine for handling HTTP/1.1 keep-alive and pipelining.
2//!
3//! Each accepted TCP stream is wrapped by [`Connection`] and progressed through
4//! the parsing, routing, and response lifecycle while ensuring the wire remains
5//! synchronised per RFC 9112 requirements.
6
7use std::{
8    sync::Arc,
9    time::{Duration, SystemTime},
10};
11
12use bytes::{Buf, Bytes, BytesMut};
13use thiserror::Error;
14use tokio::{io::AsyncReadExt, net::TcpStream, time};
15
16use crate::{
17    body::Body,
18    common::{HttpVersion, Method, StatusCode},
19    date,
20    encode::{ConnectionDirective, EncodeError, ResponseWriter},
21    headers::header_keys,
22    log,
23    parse::{self, BodyError, BodyMode, ParseError},
24    request::Request,
25    response::{Response, ResponseBody},
26    routing::Handler,
27};
28
29/// Represents an individual client connection.
30pub struct Connection {
31    id: u64,
32    stream: TcpStream,
33    handler: Arc<dyn Handler>,
34    buffer: BytesMut,
35    state: ConnectionState,
36    config: ConnectionConfig,
37}
38
39/// Tunable connection behaviour.
40#[derive(Debug, Clone)]
41pub struct ConnectionConfig {
42    pub header_read_timeout: Duration,
43    pub body_read_timeout: Duration,
44    pub idle_timeout: Duration,
45}
46
47impl Default for ConnectionConfig {
48    fn default() -> Self {
49        Self {
50            header_read_timeout: Duration::from_secs(5),
51            body_read_timeout: Duration::from_secs(30),
52            idle_timeout: Duration::from_secs(60),
53        }
54    }
55}
56
57/// Categorises timeout failures for diagnostics.
58#[derive(Debug, Clone, Copy, PartialEq, Eq)]
59pub enum TimeoutKind {
60    Header,
61    Body,
62    Idle,
63}
64
65/// High-level phases a connection can occupy.
66#[derive(Debug, Clone, Copy, PartialEq, Eq)]
67pub enum ConnectionState {
68    /// Awaiting a new request head.
69    Idle,
70    /// Streaming a request/response pair.
71    Streaming,
72    /// Preparing to close the underlying transport.
73    Closing,
74}
75
76impl Connection {
77    /// Create a new connection wrapper with a monotonically increasing id.
78    pub fn new(
79        id: u64,
80        stream: TcpStream,
81        handler: Arc<dyn Handler>,
82        config: ConnectionConfig,
83    ) -> Self {
84        Self {
85            id,
86            stream,
87            handler,
88            buffer: BytesMut::with_capacity(16 * 1024),
89            state: ConnectionState::Idle,
90            config,
91        }
92    }
93
94    /// Returns the identifier associated with the connection.
95    pub fn id(&self) -> u64 {
96        self.id
97    }
98
99    /// Drive the connection until it terminates, handling pipelined requests in order.
100    pub async fn run(mut self) -> Result<(), ConnectionError> {
101        let mut processed_requests = false;
102
103        loop {
104            self.state = ConnectionState::Idle;
105
106            if parse::needs_more_head(&self.buffer[..]) {
107                let buffer_empty = self.buffer.is_empty();
108                let (timeout_duration, timeout_kind) = if buffer_empty && processed_requests {
109                    (self.config.idle_timeout, TimeoutKind::Idle)
110                } else {
111                    (self.config.header_read_timeout, TimeoutKind::Header)
112                };
113
114                match time::timeout(timeout_duration, self.stream.read_buf(&mut self.buffer)).await
115                {
116                    Ok(Ok(0)) => return Ok(()),
117                    Ok(Ok(_)) => {}
118                    Ok(Err(error)) => return Err(ConnectionError::Io(error)),
119                    Err(_) => {
120                        log::warn(&format!(
121                            "connection {} timed out during {:?} read",
122                            self.id, timeout_kind
123                        ));
124                        self.respond_with_error(StatusCode::REQUEST_TIMEOUT, "Request timeout")
125                            .await?;
126                        return Err(ConnectionError::Timeout(timeout_kind));
127                    }
128                }
129
130                continue;
131            }
132
133            self.state = ConnectionState::Streaming;
134
135            let (mut request, body_mode, consumed) =
136                match parse::parse_request_head(&self.buffer[..]) {
137                    Ok(result) => result,
138                    Err(err) => {
139                        self.respond_with_error(StatusCode::BAD_REQUEST, "Bad Request")
140                            .await?;
141                        self.state = ConnectionState::Closing;
142                        return Err(ConnectionError::Parse(err));
143                    }
144                };
145            self.buffer.advance(consumed);
146
147            let request_method = request.method().clone();
148            let request_version = request.version();
149            let request_close = should_close_after_request(request_version, request.headers());
150
151            if request.expect_100_continue() {
152                let mut writer = ResponseWriter::new(&mut self.stream);
153                writer.write_continue().await?;
154            }
155
156            if body_mode != BodyMode::None {
157                let mut reader = parse::body_reader(body_mode, &mut self.stream, &mut self.buffer);
158                let read_result = time::timeout(self.config.body_read_timeout, async move {
159                    let mut buf = BytesMut::new();
160                    while let Some(chunk) = reader.read_next().await? {
161                        buf.extend_from_slice(&chunk);
162                    }
163                    Ok::<BytesMut, BodyError>(buf)
164                })
165                .await;
166
167                match read_result {
168                    Ok(Ok(bytes)) => {
169                        request.set_body_bytes(bytes.freeze());
170                    }
171                    Ok(Err(err)) => {
172                        self.respond_with_error(StatusCode::BAD_REQUEST, "Malformed request body")
173                            .await?;
174                        self.state = ConnectionState::Closing;
175                        return Err(ConnectionError::Body(err));
176                    }
177                    Err(_) => {
178                        log::warn(&format!(
179                            "connection {} timed out while reading request body",
180                            self.id
181                        ));
182                        self.respond_with_error(
183                            StatusCode::REQUEST_TIMEOUT,
184                            "Request body timeout",
185                        )
186                        .await?;
187                        self.state = ConnectionState::Closing;
188                        return Err(ConnectionError::Timeout(TimeoutKind::Body));
189                    }
190                }
191            }
192            let request_snapshot = request.clone();
193            request.set_body(Body::Empty);
194
195            let mut response = self.handler.handle(request);
196            if response.status().as_u16() < 400 {
197                apply_request_preconditions(&request_snapshot, &mut response);
198            }
199            let response_close = response_requests_close(&response);
200
201            let mut directive = if request_close || response_close {
202                ConnectionDirective::Close
203            } else {
204                ConnectionDirective::KeepAlive
205            };
206
207            if request_version != HttpVersion::Http11 {
208                directive = ConnectionDirective::Close;
209            }
210
211            let mut writer = ResponseWriter::new(&mut self.stream);
212            writer
213                .write_response(&mut response, &request_method, directive)
214                .await?;
215            writer.flush().await?;
216
217            processed_requests = true;
218
219            if matches!(directive, ConnectionDirective::Close) {
220                self.state = ConnectionState::Closing;
221                return Ok(());
222            }
223        }
224    }
225
226    async fn respond_with_error(
227        &mut self,
228        status: StatusCode,
229        message: &str,
230    ) -> Result<(), EncodeError> {
231        log::warn(&format!(
232            "connection {} sending {}: {}",
233            self.id, status, message
234        ));
235        let mut response = Response::new(status);
236        response
237            .headers_mut()
238            .insert(header_keys::CONTENT_TYPE, "text/plain; charset=utf-8");
239        response.set_body_bytes(Bytes::copy_from_slice(message.as_bytes()));
240
241        let mut writer = ResponseWriter::new(&mut self.stream);
242        let method = Method::Get;
243        writer
244            .write_response(&mut response, &method, ConnectionDirective::Close)
245            .await?;
246        writer.flush().await
247    }
248}
249
250/// Errors that can arise while servicing a connection.
251#[derive(Debug, Error)]
252pub enum ConnectionError {
253    #[error(transparent)]
254    Io(#[from] std::io::Error),
255    #[error(transparent)]
256    Encode(#[from] EncodeError),
257    #[error(transparent)]
258    Body(#[from] BodyError),
259    #[error(transparent)]
260    Parse(#[from] ParseError),
261    #[error("connection timeout while handling {0:?}")]
262    Timeout(TimeoutKind),
263}
264
265fn should_close_after_request(version: HttpVersion, headers: &crate::headers::Headers) -> bool {
266    match version {
267        HttpVersion::Http11 => headers
268            .get(header_keys::CONNECTION)
269            .map_or(false, |value| contains_token(value, "close")),
270        HttpVersion::Http10 => !headers
271            .get(header_keys::CONNECTION)
272            .map_or(false, |value| contains_token(value, "keep-alive")),
273        _ => true,
274    }
275}
276
277fn response_requests_close(response: &Response) -> bool {
278    response
279        .headers()
280        .get(header_keys::CONNECTION)
281        .map_or(false, |value| contains_token(value, "close"))
282}
283
284fn contains_token(value: &str, token: &str) -> bool {
285    value
286        .split(',')
287        .any(|part| part.trim().eq_ignore_ascii_case(token))
288}
289
290fn apply_request_preconditions(request: &Request, response: &mut Response) {
291    let mut decision: Option<StatusCode> = None;
292    let response_etag = response.headers().get(header_keys::ETAG);
293
294    if let Some(if_match) = request.if_match() {
295        if !etag_list_matches(if_match, response_etag, true) {
296            decision = Some(StatusCode::PRECONDITION_FAILED);
297        }
298    }
299
300    if decision.is_none() {
301        if let Some(unmodified_since) = request.if_unmodified_since() {
302            if let Some(last_modified) = parse_last_modified(response) {
303                if last_modified > unmodified_since {
304                    decision = Some(StatusCode::PRECONDITION_FAILED);
305                }
306            }
307        }
308    }
309
310    if decision.is_none() {
311        if let Some(if_none_match) = request.if_none_match() {
312            if etag_list_matches(if_none_match, response_etag, false) {
313                if matches!(request.method(), Method::Get | Method::Head) {
314                    decision = Some(StatusCode::NOT_MODIFIED);
315                } else {
316                    decision = Some(StatusCode::PRECONDITION_FAILED);
317                }
318            }
319        } else if let Some(if_modified_since) = request.if_modified_since() {
320            if matches!(request.method(), Method::Get | Method::Head) {
321                if let Some(last_modified) = parse_last_modified(response) {
322                    if last_modified <= if_modified_since {
323                        decision = Some(StatusCode::NOT_MODIFIED);
324                    }
325                }
326            }
327        }
328    }
329
330    if let Some(status) = decision {
331        log::info(&format!(
332            "precondition evaluation changed response to {} for {}",
333            status,
334            request.method().as_str()
335        ));
336        response.set_status(status);
337        response.set_body(ResponseBody::Empty);
338        response.take_trailers();
339        let headers = response.headers_mut();
340        headers.remove(header_keys::CONTENT_LENGTH);
341        headers.remove(header_keys::TRANSFER_ENCODING);
342        if status == StatusCode::NOT_MODIFIED {
343            headers.remove(header_keys::CONTENT_TYPE);
344        }
345    }
346}
347
348fn parse_last_modified(response: &Response) -> Option<SystemTime> {
349    response
350        .headers()
351        .get(header_keys::LAST_MODIFIED)
352        .and_then(|value| date::parse_http_date(value.trim()))
353}
354
355fn etag_list_matches(value: &str, entity_tag: Option<&str>, strong: bool) -> bool {
356    let trimmed = value.trim();
357    if trimmed == "*" {
358        return entity_tag.is_some();
359    }
360
361    let entity = entity_tag.map(|v| v.trim()).filter(|v| !v.is_empty());
362    let Some(entity) = entity else {
363        return false;
364    };
365
366    trimmed
367        .split(',')
368        .map(|candidate| candidate.trim())
369        .filter(|candidate| !candidate.is_empty())
370        .any(|candidate| {
371            if strong {
372                candidate == entity
373            } else {
374                weak_etag_equal(candidate, entity)
375            }
376        })
377}
378
379fn weak_etag_equal(a: &str, b: &str) -> bool {
380    if a == b {
381        return true;
382    }
383
384    let a = a.trim_start_matches("W/");
385    let b = b.trim_start_matches("W/");
386    a == b
387}
goose_http/conn/mod.rs

goose_http/conn/
mod.rs
