sozu_lib/protocol/mux/
h1.rs

1//! H1 mux connection wrapper.
2//!
3//! Hosts the single active H1 stream (`stream: GlobalStreamId`) and wires
4//! Kawa-owned H1 parsing + serialization into the shared mux `Context` so the
5//! same routing / shutdown / readiness machinery applies across H1 and H2
6//! connections. Long-form lifecycle: `lib/src/protocol/mux/LIFECYCLE.md`.
7
8use std::io::IoSlice;
9
10use rusty_ulid::Ulid;
11use sozu_command::{logging::ansi_palette, ready::Ready};
12
13use crate::metrics::names;
14use crate::{
15    L7ListenerHandler, ListenerHandler, Readiness,
16    protocol::mux::{
17        BackendStatus, Context, DebugEvent, Endpoint, GlobalStreamId, MuxResult, Position,
18        StreamState, forcefully_terminate_answer,
19        parser::H2Error,
20        remove_backend_stream, set_default_answer,
21        shared::{EndStreamAction, drain_tls_close_notify, end_stream_decision},
22        update_readiness_after_read, update_readiness_after_write,
23    },
24    socket::{SocketHandler, SocketResult, stats::socket_rtt},
25    timer::TimeoutContainer,
26};
27
28/// Prefix applied to every [`ConnectionH1`] log line. Matches the RUSTLS
29/// log-context convention (`MUX-H1\tSession(...)\t >>>`). When the logger is
30/// in colored mode the label is bold bright-white (uniform across every
31/// protocol) and the session detail is rendered in light grey.
32///
33/// Fields included in the session block (chosen to surface the most common
34/// H1 troubleshooting axes — keep-alive churn, stream pinning, buffer-pressure
35/// stall and graceful TLS shutdown):
36/// - `peer` — peer address (or `None` if the socket is gone)
37/// - `position` — `Server` / `Client(...)` orientation
38/// - `stream` — currently active [`GlobalStreamId`] (or `none`)
39/// - `requests` — request count served on this connection (keep-alive)
40/// - `parked` — set when the kawa buffer is full and `READABLE` is suspended
41/// - `close_notify` — TLS `close_notify` send state
42/// - `readiness` — connection-level mio readiness snapshot
43macro_rules! log_context {
44    ($self:expr) => {{
45        let (open, reset, grey, gray, white) = ansi_palette();
46        format!(
47            "[{ulid} - - -]\t{open}MUX-H1{reset}\t{grey}Session{reset}({gray}peer{reset}={white}{peer:?}{reset}, {gray}position{reset}={white}{position:?}{reset}, {gray}stream{reset}={white}{stream:?}{reset}, {gray}requests{reset}={white}{requests}{reset}, {gray}parked{reset}={white}{parked}{reset}, {gray}close_notify{reset}={white}{close_notify}{reset}, {gray}readiness{reset}={white}{readiness}{reset})\t >>>",
48            open = open,
49            reset = reset,
50            grey = grey,
51            gray = gray,
52            white = white,
53            ulid = $self.session_ulid,
54            peer = $self.socket.socket_ref().peer_addr().ok(),
55            position = $self.position,
56            stream = $self.stream,
57            requests = $self.requests,
58            parked = $self.parked_on_buffer_pressure,
59            close_notify = $self.close_notify_sent,
60            readiness = $self.readiness,
61        )
62    }};
63}
64
65/// Per-stream variant of [`log_context!`] used when a `HttpContext` is in
66/// scope. Fills the `request_id` slot of the bracket so the log line can be
67/// grepped by the specific request that triggered it.
68#[allow(unused_macros)]
69macro_rules! log_context_stream {
70    ($self:expr, $http_context:expr) => {{
71        let (open, reset, grey, gray, white) = ansi_palette();
72        format!(
73            "[{ulid} {req} {cluster} {backend}]\t{open}MUX-H1{reset}\t{grey}Session{reset}({gray}peer{reset}={white}{peer:?}{reset}, {gray}position{reset}={white}{position:?}{reset}, {gray}stream{reset}={white}{stream:?}{reset}, {gray}requests{reset}={white}{requests}{reset}, {gray}parked{reset}={white}{parked}{reset}, {gray}close_notify{reset}={white}{close_notify}{reset}, {gray}readiness{reset}={white}{readiness}{reset})\t >>>",
74            open = open,
75            reset = reset,
76            grey = grey,
77            gray = gray,
78            white = white,
79            ulid = $self.session_ulid,
80            req = $http_context.id,
81            cluster = $http_context.cluster_id.as_deref().unwrap_or("-"),
82            backend = $http_context.backend_id.as_deref().unwrap_or("-"),
83            peer = $self.socket.socket_ref().peer_addr().ok(),
84            position = $self.position,
85            stream = $self.stream,
86            requests = $self.requests,
87            parked = $self.parked_on_buffer_pressure,
88            close_notify = $self.close_notify_sent,
89            readiness = $self.readiness,
90        )
91    }};
92}
93
94/// Module-level prefix for logs without a [`ConnectionH1`] in scope. Honours
95/// the colored flag.
96macro_rules! log_module_context {
97    () => {{
98        let (open, reset, _, _, _) = ansi_palette();
99        format!("{open}MUX-H1{reset}\t >>>", open = open, reset = reset)
100    }};
101}
102
103/// HTTP/1.1 connection handler within the mux layer.
104///
105/// Manages a single HTTP/1.1 connection (either frontend or backend),
106/// handling request/response forwarding through kawa buffers. Supports
107/// keep-alive, chunked transfer encoding, close-delimited responses,
108/// and upgrade (e.g., WebSocket).
109pub struct ConnectionH1<Front: SocketHandler> {
110    pub position: Position,
111    pub readiness: Readiness,
112    pub requests: usize,
113    pub socket: Front,
114    /// Active stream index, or `None` when the connection has no assigned stream
115    /// (initial client state before `start_stream`, or after `end_stream` detaches).
116    pub stream: Option<GlobalStreamId>,
117    pub timeout_container: TimeoutContainer,
118    /// Set when `readable` exits early because the kawa buffer was full.
119    /// Edge-triggered epoll will not re-fire READABLE for data already in the
120    /// kernel socket buffer, so the cross-readiness mechanism must re-arm it
121    /// via `try_resume_reading` once the peer drains the buffer.
122    pub parked_on_buffer_pressure: bool,
123    /// True once we've asked rustls to emit TLS close_notify for this frontend.
124    pub close_notify_sent: bool,
125    /// Connection/session ULID propagated from the parent [`Mux`]. Used to
126    /// stamp the session slot of the `[session req cluster backend]` log
127    /// prefix emitted by the local `log_context!` macro.
128    pub session_ulid: Ulid,
129}
130
131impl<Front: SocketHandler> std::fmt::Debug for ConnectionH1<Front> {
132    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
133        f.debug_struct("ConnectionH1")
134            .field("position", &self.position)
135            .field("readiness", &self.readiness)
136            .field("socket", &self.socket.socket_ref())
137            .field("stream", &self.stream)
138            .finish()
139    }
140}
141
142impl<Front: SocketHandler> ConnectionH1<Front> {
143    fn defer_close_for_tls_flush(&mut self, reason: &'static str) -> MuxResult {
144        if self.initiate_close_notify() {
145            trace!(
146                "{} H1 writable delaying close after {}: stream={:?}, close_notify_sent={}, wants_write={}, readiness={:?}",
147                log_context!(self),
148                reason,
149                self.stream,
150                self.close_notify_sent,
151                self.socket.socket_wants_write(),
152                self.readiness
153            );
154            MuxResult::Continue
155        } else {
156            MuxResult::CloseSession
157        }
158    }
159
160    /// Terminate a close-delimited kawa body by pushing END_STREAM flags.
161    /// Called when the backend closes the connection to signal end-of-body
162    /// (no Content-Length, no chunked encoding).
163    ///
164    /// Chunked responses that TCP-close before the terminating `0\r\n\r\n`
165    /// are demoted to `ParsingPhase::Error` so the H2 converter emits
166    /// RST_STREAM(InternalError) rather than a silent END_STREAM with a
167    /// truncated body — RFC 9112 §7.1 requires the zero-chunk terminator.
168    fn terminate_close_delimited(kawa: &mut super::GenericHttpStream, stream_id: GlobalStreamId) {
169        // Pre: we only synthesize an end-of-body for a response still in its
170        // body phase. A kawa already Terminated/Error must not be re-terminated
171        // (it would double-push an END_STREAM flag onto the converter).
172        debug_assert!(
173            !kawa.is_terminated(),
174            "terminate_close_delimited must not run on an already-terminated kawa"
175        );
176        if kawa.body_size == kawa::BodySize::Chunked {
177            warn!(
178                "{} H1 backend EOF mid-chunked response on stream {}: emitting RST_STREAM",
179                log_module_context!(),
180                stream_id
181            );
182            incr!(names::h1::BACKEND_EOF_BEFORE_MESSAGE_COMPLETE);
183            kawa.parsing_phase
184                .error(kawa::ParsingErrorKind::Processing {
185                    message: "INTERNAL_ERROR",
186                });
187            // Post: a truncated chunked body is demoted to Error so the
188            // converter emits RST_STREAM, never a silent END_STREAM.
189            debug_assert!(
190                kawa.is_error(),
191                "truncated chunked response must end in the Error phase"
192            );
193            return;
194        }
195        debug!(
196            "{} H1 close-delimited EOF on stream {}: terminating body",
197            log_module_context!(),
198            stream_id
199        );
200        kawa.push_block(kawa::Block::Flags(kawa::Flags {
201            end_body: true,
202            end_chunk: false,
203            end_header: false,
204            end_stream: true,
205        }));
206        kawa.parsing_phase = kawa::ParsingPhase::Terminated;
207        // Post: a close-delimited body is now Terminated so the converter
208        // emits DATA with END_STREAM on the last frame.
209        debug_assert!(
210            kawa.is_terminated(),
211            "close-delimited body must end in the Terminated phase"
212        );
213    }
214
215    pub fn readable<E, L>(&mut self, context: &mut Context<L>, mut endpoint: E) -> MuxResult
216    where
217        E: Endpoint,
218        L: ListenerHandler + L7ListenerHandler,
219    {
220        trace!(
221            "{} ======= MUX H1 READABLE {:?}",
222            log_context!(self),
223            self.position
224        );
225        let Some(stream_id) = self.stream else {
226            error!(
227                "{} readable() called on H1 connection with no active stream",
228                log_context!(self)
229            );
230            return MuxResult::Continue;
231        };
232        self.timeout_container.reset();
233        let answers_rc = context.listener.borrow().get_answers().clone();
234        let stream = &mut context.streams[stream_id];
235        if stream.metrics.start.is_none() {
236            stream.metrics.mark_request_start();
237        }
238        let parts = stream.split(&self.position);
239        let kawa = parts.rbuffer;
240
241        // If the buffer has no space, don't attempt a read — socket_read with
242        // an empty buffer returns (0, Continue) which is indistinguishable from
243        // a real EOF. Remove READABLE from the event so the inner loop doesn't
244        // spin; `try_resume_reading` will re-arm it once the peer drains the
245        // buffer (edge-triggered epoll won't re-fire for data already in the
246        // kernel socket buffer).
247        if kawa.storage.available_space() == 0 {
248            self.readiness.event.remove(Ready::READABLE);
249            self.parked_on_buffer_pressure = true;
250            // Pair the park flag with the cleared READABLE event: only
251            // `try_resume_reading` may re-arm it once the peer drains space.
252            debug_assert!(
253                self.parked_on_buffer_pressure && !self.readiness.event.is_readable(),
254                "parking on buffer pressure must clear the READABLE event"
255            );
256            return MuxResult::Continue;
257        }
258
259        self.parked_on_buffer_pressure = false;
260        // Pre: we never ask the socket to read into a full buffer (that path
261        // returned above) — `socket_read` requires a non-empty target slice.
262        let space_before = kawa.storage.available_space();
263        debug_assert!(
264            space_before > 0,
265            "socket_read must target a buffer with free space"
266        );
267        let (size, status) = self.socket.socket_read(kawa.storage.space());
268        // A read cannot deliver more bytes than the buffer had room for.
269        // `debug_assert!` only — `size` is socket-derived, never trusted to
270        // panic, but a violation here is a SocketHandler contract bug.
271        debug_assert!(
272            size <= space_before,
273            "socket_read returned more bytes than the buffer could hold"
274        );
275        context.debug.push(DebugEvent::StreamEvent(0, size));
276        kawa.storage.fill(size);
277        debug_assert_eq!(
278            kawa.storage.available_space(),
279            space_before - size,
280            "fill must consume exactly `size` bytes of free space"
281        );
282        self.position.count_bytes_in_counter(size);
283        self.position.count_bytes_in(parts.metrics, size);
284        if update_readiness_after_read(size, status, &mut self.readiness) {
285            // size=0: the socket returned EOF (Closed) or WouldBlock.
286            // For a close-delimited backend response (no Content-Length, no
287            // chunked), a graceful EOF IS the end-of-body signal. Terminate
288            // the kawa so the H2 converter emits DATA with END_STREAM.
289            // SocketResult::Error (ECONNRESET etc.) is NOT treated as a valid
290            // close-delimiter — transport errors should produce 502, not a
291            // truncated response.
292            if status == SocketResult::Closed
293                && self.position.is_client()
294                && kawa.is_main_phase()
295                && !kawa.is_terminated()
296                && !parts.context.keep_alive_backend
297            {
298                Self::terminate_close_delimited(kawa, stream_id);
299                self.timeout_container.cancel();
300                self.readiness.interest.remove(Ready::READABLE);
301                if let StreamState::Linked(token) = stream.state {
302                    // Signal pending write alongside the WRITABLE interest flip:
303                    // edge-triggered epoll won't re-fire for bytes we just queued
304                    // onto the peer — the synthetic event is the only wake path.
305                    let peer = endpoint.readiness_mut(token);
306                    peer.arm_writable();
307                }
308            }
309            return MuxResult::Continue;
310        }
311
312        let was_main_phase = kawa.is_main_phase();
313        kawa::h1::parse(kawa, parts.context);
314        if kawa.is_error() {
315            match self.position {
316                Position::Client(..) => {
317                    incr!(names::http::BACKEND_PARSE_ERRORS);
318                    let StreamState::Linked(token) = stream.state else {
319                        error!(
320                            "{} client stream in error is not in Linked state",
321                            log_context!(self)
322                        );
323                        return MuxResult::CloseSession;
324                    };
325                    let global_stream_id = stream_id;
326                    self.end_stream(global_stream_id, context);
327                    endpoint.end_stream(token, global_stream_id, context);
328                }
329                Position::Server => {
330                    incr!(names::http::FRONTEND_PARSE_ERRORS);
331                    let answers = answers_rc.borrow();
332                    set_default_answer(stream, &mut self.readiness, 400, &answers);
333                }
334            }
335            return MuxResult::Continue;
336        }
337        // Capture borrow-sensitive values after parsing but before the 1xx block
338        // accesses stream.state (which ends the split borrow from `parts`).
339        let is_keep_alive_backend = parts.context.keep_alive_backend;
340        let is_body_phase_after_parse = kawa.is_main_phase();
341
342        // 1xx informational responses (100 Continue, 103 Early Hints): the H1
343        // parser treats them as complete (Terminated + end_stream=true), but for
344        // H2 frontends they must be forwarded WITHOUT END_STREAM so the real
345        // response can follow on the same stream. Also keep READABLE interest
346        // so the backend can send the final response.
347        let is_1xx_backend = if self.position.is_client() {
348            if let kawa::StatusLine::Response { code, .. } = &kawa.detached.status_line {
349                if (100..200).contains(code) {
350                    debug!(
351                        "{} H1 backend: received {} informational response",
352                        log_context!(self),
353                        code
354                    );
355                    for block in &mut kawa.blocks {
356                        if let kawa::Block::Flags(flags) = block {
357                            flags.end_stream = false;
358                            flags.end_body = false;
359                        }
360                    }
361                    true
362                } else {
363                    false
364                }
365            } else {
366                false
367            }
368        } else {
369            false
370        };
371        if kawa.is_terminated() && !is_1xx_backend {
372            self.timeout_container.cancel();
373            self.readiness.interest.remove(Ready::READABLE);
374        }
375        if kawa.is_main_phase() {
376            if !was_main_phase && self.position.is_server() {
377                if parts.context.method.is_none()
378                    || parts.context.authority.is_none()
379                    || parts.context.path.is_none()
380                {
381                    if let kawa::StatusLine::Request {
382                        version: kawa::Version::V10,
383                        ..
384                    } = kawa.detached.status_line
385                    {
386                        error!(
387                            "{} Unexpected malformed request: HTTP/1.0 from {:?} with {:?} {:?} {:?}",
388                            log_context!(self),
389                            parts.context.session_address,
390                            parts.context.method,
391                            parts.context.authority,
392                            parts.context.path
393                        );
394                    } else {
395                        error!("{} Unexpected malformed request", log_context!(self));
396                        kawa::debug_kawa(kawa);
397                    }
398                    let answers = answers_rc.borrow();
399                    set_default_answer(stream, &mut self.readiness, 400, &answers);
400                    return MuxResult::Continue;
401                }
402                // First-seen request on this (server) connection: the keep-alive
403                // request counter advances by exactly one and the matching
404                // `http.active_requests` gauge `+1` is paired with flipping
405                // `request_counted` true (the `generate_access_log` `-1` is
406                // gated on that flag, so they must stay balanced).
407                let requests_before = self.requests;
408                let links_before = context.pending_links.len();
409                self.requests += 1;
410                debug_assert_eq!(
411                    self.requests,
412                    requests_before + 1,
413                    "server keep-alive request counter must advance by exactly one"
414                );
415                trace!("{} REQUESTS: {}", log_context!(self), self.requests);
416                incr!(names::http::REQUESTS);
417                gauge_add!(names::http::ACTIVE_REQUESTS, 1);
418                parts.metrics.service_start();
419                // Set request_counted after the last use of `parts` to satisfy the borrow checker
420                stream.request_counted = true;
421                stream.state = StreamState::Link;
422                context.pending_links.push_back(stream_id);
423                // Post: the stream is queued for backend linking exactly once
424                // and is now in the Link state the ready loop expects.
425                debug_assert!(
426                    stream.request_counted,
427                    "request_counted must be set when the active-requests gauge is incremented"
428                );
429                debug_assert_eq!(
430                    stream.state,
431                    StreamState::Link,
432                    "a first-seen request must transition the stream to Link"
433                );
434                debug_assert_eq!(
435                    context.pending_links.len(),
436                    links_before + 1,
437                    "a first-seen request must enqueue exactly one pending link"
438                );
439            }
440            if let StreamState::Linked(token) = stream.state {
441                // Signal pending write alongside the WRITABLE interest flip: the
442                // bytes we just parsed live in sozu's buffers, not the kernel,
443                // so edge-triggered epoll won't re-fire on its own.
444                let peer = endpoint.readiness_mut(token);
445                peer.arm_writable();
446            }
447        };
448        // 1xx informational: the 100 response skips main_phase (goes straight to
449        // Terminated), so the normal "set endpoint writable" above never fires.
450        // Trigger the frontend to write the 1xx response after all borrows end.
451        if is_1xx_backend && let StreamState::Linked(token) = stream.state {
452            let peer = endpoint.readiness_mut(token);
453            peer.arm_writable();
454        }
455
456        // Close-delimited response: socket_read returned (size > 0, Closed) —
457        // the last data chunk arrived together with the EOF in a single read.
458        // After parsing the data above, terminate the kawa now so the H2
459        // converter emits END_STREAM on the last DATA frame.
460        if status == SocketResult::Closed
461            && self.position.is_client()
462            && is_body_phase_after_parse
463            && !is_keep_alive_backend
464            && !context.streams[stream_id].back.is_terminated()
465        {
466            let kawa = &mut context.streams[stream_id].back;
467            Self::terminate_close_delimited(kawa, stream_id);
468            self.timeout_container.cancel();
469            self.readiness.interest.remove(Ready::READABLE);
470        }
471
472        MuxResult::Continue
473    }
474
475    pub fn writable<E, L>(&mut self, context: &mut Context<L>, mut endpoint: E) -> MuxResult
476    where
477        E: Endpoint,
478        L: ListenerHandler + L7ListenerHandler,
479    {
480        trace!(
481            "{} ======= MUX H1 WRITABLE {:?}",
482            log_context!(self),
483            self.position
484        );
485        let Some(stream_id) = self.stream else {
486            if self.socket.socket_wants_write() {
487                let (size, status) = self.socket.socket_write_vectored(&[]);
488                let _ = update_readiness_after_write(size, status, &mut self.readiness);
489                if self.socket.socket_wants_write() {
490                    self.readiness.signal_pending_write();
491                }
492            }
493            return MuxResult::Continue;
494        };
495        self.timeout_container.reset();
496        let stream = &mut context.streams[stream_id];
497        let parts = stream.split(&self.position);
498        let kawa = parts.wbuffer;
499        // Apply per-frontend response-side header edits stashed by the
500        // routing layer at request time. Only the Server-position pass
501        // touches the response back-kawa; the Client-position pass
502        // (writing the request to the backend) has already had its
503        // edits applied in `Router::route_from_request`.
504        //
505        // Drained via `mem::take` so the injection runs exactly once
506        // per response. H1 keep-alive can re-enter this writable path
507        // for the same stream when the backend response spans more
508        // than one TCP read; without the take, the second pass would
509        // re-insert the same headers (typically as duplicate STS lines
510        // on the wire — RFC 6797 §6.1 expects a single header). On H2
511        // the same multi-prepare-cycle pattern surfaces as a
512        // `H2BlockConverter::finalize` "out buffer not empty" leak.
513        if matches!(self.position, Position::Server) && !parts.context.headers_response.is_empty() {
514            let edits = std::mem::take(&mut parts.context.headers_response);
515            super::shared::apply_response_header_edits(kawa, &edits);
516        }
517        kawa.prepare(&mut kawa::h1::BlockConverter);
518        let mut io_slices = Vec::new();
519        for block in kawa.out.iter() {
520            match block {
521                kawa::OutBlock::Delimiter => break,
522                kawa::OutBlock::Store(store) => {
523                    io_slices.push(IoSlice::new(store.data(kawa.storage.buffer())));
524                }
525            }
526        }
527        let can_finalize_server_close = matches!(self.position, Position::Server)
528            && kawa.is_terminated()
529            && kawa.is_completed();
530        if io_slices.is_empty() && !self.socket.socket_wants_write() && !can_finalize_server_close {
531            self.readiness.interest.remove(Ready::WRITABLE);
532            return MuxResult::Continue;
533        }
534        let tls_only_flush = io_slices.is_empty();
535        // Total bytes we offered the socket across the gathered slices; a
536        // vectored write can never report more consumed than we handed it.
537        let queued: usize = io_slices.iter().map(|s| s.len()).sum();
538        let (size, status) = self.socket.socket_write_vectored(&io_slices);
539        debug_assert!(
540            size <= queued,
541            "socket_write_vectored reported more bytes written than were queued"
542        );
543        context.debug.push(DebugEvent::StreamEvent(1, size));
544        kawa.consume(size);
545        self.position.count_bytes_out_counter(size);
546        self.position.count_bytes_out(parts.metrics, size);
547        let should_yield = update_readiness_after_write(size, status, &mut self.readiness);
548        if self.socket.socket_wants_write() {
549            self.readiness.signal_pending_write();
550            // Pair the queued-write signal with the socket's own report: we
551            // only synthesize a WRITABLE event when the socket still has bytes
552            // buffered (edge-triggered epoll won't re-fire on its own).
553            debug_assert!(
554                self.readiness.event.is_writable(),
555                "signal_pending_write must leave a WRITABLE event queued"
556            );
557            return MuxResult::Continue;
558        }
559        if !tls_only_flush && should_yield {
560            return MuxResult::Continue;
561        }
562
563        if kawa.is_terminated() && kawa.is_completed() {
564            match self.position {
565                Position::Client(..) => self.readiness.interest.insert(Ready::READABLE),
566                Position::Server => {
567                    if stream.context.closing {
568                        return self.defer_close_for_tls_flush("closing-context");
569                    }
570                    let kawa = &mut stream.back;
571                    match kawa.detached.status_line {
572                        kawa::StatusLine::Response { code: 101, .. } => {
573                            debug!("{} ============== HANDLE UPGRADE!", log_context!(self));
574                            stream.metrics.backend_stop();
575                            let client_rtt = socket_rtt(self.socket.socket_ref());
576                            let server_rtt = stream
577                                .linked_token()
578                                .and_then(|t| endpoint.socket(t))
579                                .and_then(socket_rtt);
580                            stream.generate_access_log(
581                                false,
582                                Some("H1::Upgrade"),
583                                context.listener.clone(),
584                                client_rtt,
585                                server_rtt,
586                            );
587                            return MuxResult::Upgrade;
588                        }
589                        kawa::StatusLine::Response { code: 100, .. } => {
590                            debug!("{} ============== HANDLE CONTINUE!", log_context!(self));
591                            // After a 100 Continue, we expect the client to continue
592                            // with its request body. Do NOT call generate_access_log
593                            // here — the final response will emit the access log.
594                            // Calling it here would double-decrement http.active_requests.
595                            self.timeout_container.reset();
596                            self.readiness.interest.insert(Ready::READABLE);
597                            kawa.clear();
598                            stream.metrics.backend_stop();
599                            if let StreamState::Linked(token) = stream.state {
600                                endpoint
601                                    .readiness_mut(token)
602                                    .interest
603                                    .insert(Ready::READABLE);
604                            }
605                            return MuxResult::Continue;
606                        }
607                        kawa::StatusLine::Response { code: 103, .. } => {
608                            debug!("{} ============== HANDLE EARLY HINT!", log_context!(self));
609                            // Do NOT call generate_access_log for 103 Early Hints.
610                            // The final response will emit the access log.
611                            // Calling it here would double-decrement http.active_requests.
612                            if let StreamState::Linked(token) = stream.state {
613                                // after a 103 early hints, we expect the backend to send its response
614                                endpoint
615                                    .readiness_mut(token)
616                                    .interest
617                                    .insert(Ready::READABLE);
618                                kawa.clear();
619                                stream.metrics.backend_stop();
620                                return MuxResult::Continue;
621                            } else {
622                                stream.metrics.backend_stop();
623                                let client_rtt = socket_rtt(self.socket.socket_ref());
624                                let server_rtt = stream
625                                    .linked_token()
626                                    .and_then(|t| endpoint.socket(t))
627                                    .and_then(socket_rtt);
628                                stream.generate_access_log(
629                                    false,
630                                    Some("H1::EarlyHint"),
631                                    context.listener.clone(),
632                                    client_rtt,
633                                    server_rtt,
634                                );
635                                return self.defer_close_for_tls_flush("early-hint");
636                            }
637                        }
638                        _ => {}
639                    }
640                    incr!(names::http::E2E_HTTP11);
641                    stream.metrics.backend_stop();
642                    let client_rtt = socket_rtt(self.socket.socket_ref());
643                    let server_rtt = stream
644                        .linked_token()
645                        .and_then(|t| endpoint.socket(t))
646                        .and_then(socket_rtt);
647                    stream.generate_access_log(
648                        false,
649                        Some("H1::Complete"),
650                        context.listener.clone(),
651                        client_rtt,
652                        server_rtt,
653                    );
654                    stream.metrics.reset();
655                    let old_state = std::mem::replace(&mut stream.state, StreamState::Unlinked);
656                    if let StreamState::Linked(token) = old_state {
657                        remove_backend_stream(&mut context.backend_streams, token, stream_id);
658                    }
659                    if stream.context.keep_alive_frontend {
660                        self.timeout_container.reset();
661                        if let StreamState::Linked(token) = old_state {
662                            endpoint.end_stream(token, stream_id, context);
663                        }
664                        self.readiness.interest.insert(Ready::READABLE);
665                        let stream = &mut context.streams[stream_id];
666                        stream.context.reset();
667                        stream.back.clear();
668                        stream.back.storage.clear();
669                        stream.front.clear();
670                        // do not stream.front.storage.clear() because of H1 pipelining
671                        stream.attempts = 0;
672                        // Transition back to Idle so buffered pipelined requests
673                        // trigger a phase transition on the next readable() call.
674                        stream.state = StreamState::Idle;
675                        // Post: the keep-alive reset leaves a clean, closed slot
676                        // ready for the next pipelined request. `request_counted`
677                        // was already cleared by `generate_access_log` above, so
678                        // the slot carries no pending active-requests charge.
679                        debug_assert_eq!(
680                            stream.state,
681                            StreamState::Idle,
682                            "keep-alive reset must return the stream to Idle"
683                        );
684                        debug_assert_eq!(stream.attempts, 0, "keep-alive reset must zero attempts");
685                        debug_assert!(
686                            !stream.request_counted,
687                            "keep-alive reset must leave no counted request (active-requests leak)"
688                        );
689                        debug_assert!(
690                            stream.back.storage.is_empty(),
691                            "keep-alive reset must drain the response storage"
692                        );
693                        // HTTP/1.1 pipelining: if there's still data in the frontend
694                        // storage (pipelined requests already read from the socket),
695                        // parse it now. We can't rely on a new READABLE event because
696                        // the socket buffer may be empty — all requests were already
697                        // read into kawa storage in the first socket_read.
698                        if !stream.front.storage.is_empty() {
699                            kawa::h1::parse(&mut stream.front, &mut stream.context);
700                            let is_error = stream.front.is_error();
701                            let is_main = stream.front.is_main_phase();
702                            let malformed = is_main
703                                && (stream.context.method.is_none()
704                                    || stream.context.authority.is_none()
705                                    || stream.context.path.is_none());
706                            if is_error || malformed {
707                                let answers_rc = context.listener.borrow().get_answers().clone();
708                                let answers = answers_rc.borrow();
709                                set_default_answer(stream, &mut self.readiness, 400, &answers);
710                            } else if is_main {
711                                self.requests += 1;
712                                incr!(names::http::REQUESTS);
713                                gauge_add!(names::http::ACTIVE_REQUESTS, 1);
714                                stream.metrics.service_start();
715                                stream.request_counted = true;
716                                stream.state = StreamState::Link;
717                                context.pending_links.push_back(stream_id);
718                            }
719                            // else: incomplete parse, wait for more data via READABLE
720                        }
721                    } else {
722                        return self.defer_close_for_tls_flush("response-complete");
723                    }
724                }
725            }
726        }
727        MuxResult::Continue
728    }
729
730    pub fn force_disconnect(&mut self) -> MuxResult {
731        match &mut self.position {
732            Position::Client(_, _, status) => {
733                *status = BackendStatus::Disconnecting;
734                self.readiness.event = Ready::HUP;
735                debug!(
736                    "{} H1 force_disconnect client: stream={:?}, wants_write={}, readiness={:?}",
737                    log_context!(self),
738                    self.stream,
739                    self.socket.socket_wants_write(),
740                    self.readiness
741                );
742                MuxResult::Continue
743            }
744            Position::Server => {
745                if self.socket.socket_wants_write() {
746                    debug!(
747                        "{} H1 force_disconnect delaying close: stream={:?}, wants_write=true, readiness={:?}",
748                        log_context!(self),
749                        self.stream,
750                        self.readiness
751                    );
752                    self.readiness.interest = Ready::WRITABLE | Ready::HUP | Ready::ERROR;
753                    self.readiness.signal_pending_write();
754                    MuxResult::Continue
755                } else {
756                    debug!(
757                        "{} H1 force_disconnect closing session: stream={:?}, wants_write=false, readiness={:?}",
758                        log_context!(self),
759                        self.stream,
760                        self.readiness
761                    );
762                    MuxResult::CloseSession
763                }
764            }
765        }
766    }
767
768    pub fn has_pending_write(&self) -> bool {
769        self.socket.socket_wants_write()
770    }
771
772    pub fn initiate_close_notify(&mut self) -> bool {
773        if !self.position.is_server() {
774            return false;
775        }
776        // Past the guard we are always server-side; close_notify is a
777        // frontend-only TLS concern.
778        debug_assert!(
779            self.position.is_server(),
780            "initiate_close_notify past the guard must be server-side"
781        );
782        if !self.close_notify_sent {
783            trace!("{} H1 initiating CLOSE_NOTIFY", log_context!(self));
784            self.socket.socket_close();
785            self.close_notify_sent = true;
786        }
787        // `close_notify` is monotone: once requested it stays sent for the
788        // connection's lifetime (a second send would corrupt the TLS stream).
789        debug_assert!(
790            self.close_notify_sent,
791            "close_notify_sent must be set once initiate_close_notify has run"
792        );
793        if self.socket.socket_wants_write() {
794            self.readiness.arm_writable();
795            // arm_writable pairs interest + event so the deferred TLS flush is
796            // actually scheduled under edge-triggered epoll.
797            debug_assert!(
798                self.readiness.interest.is_writable() && self.readiness.event.is_writable(),
799                "arm_writable must set both WRITABLE interest and event"
800            );
801            true
802        } else {
803            false
804        }
805    }
806
807    pub fn close<E, L>(&mut self, context: &mut Context<L>, mut endpoint: E)
808    where
809        E: Endpoint,
810        L: ListenerHandler + L7ListenerHandler,
811    {
812        match self.position {
813            Position::Client(_, _, BackendStatus::KeepAlive)
814            | Position::Client(_, _, BackendStatus::Disconnecting) => {
815                trace!("{} close detached client ConnectionH1", log_context!(self));
816                return;
817            }
818            Position::Client(_, _, BackendStatus::Connecting(_))
819            | Position::Client(_, _, BackendStatus::Connected) => {
820                debug!(
821                    "{} BACKEND CLOSING FOR: {:?} {:?}",
822                    log_context!(self),
823                    self.position,
824                    self.stream
825                );
826            }
827            Position::Server => {
828                let tls_pending_before = self.socket.socket_wants_write();
829                let (tls_pending_after, drain_rounds) =
830                    drain_tls_close_notify(&mut self.socket, &mut self.close_notify_sent);
831                if tls_pending_after {
832                    error!(
833                        "{} H1 TLS buffer NOT fully drained on close: pending_before={}, pending_after={}, drain_rounds={}, stream={:?}, close_notify_sent={}, readiness={:?}",
834                        log_context!(self),
835                        tls_pending_before,
836                        tls_pending_after,
837                        drain_rounds,
838                        self.stream,
839                        self.close_notify_sent,
840                        self.readiness
841                    );
842                }
843                return;
844            }
845        }
846        let Some(stream_id) = self.stream else {
847            trace!(
848                "{} closing detached H1 client with no active stream",
849                log_context!(self)
850            );
851            return;
852        };
853        // reconnection is handled by the server
854        let StreamState::Linked(token) = context.streams[stream_id].state else {
855            trace!(
856                "{} closing detached H1 client in state {:?} on stream {}",
857                log_context!(self),
858                context.streams[stream_id].state,
859                stream_id
860            );
861            return;
862        };
863        endpoint.end_stream(token, stream_id, context)
864    }
865
866    pub fn end_stream<L>(&mut self, stream: GlobalStreamId, context: &mut Context<L>)
867    where
868        L: ListenerHandler + L7ListenerHandler,
869    {
870        if self.stream != Some(stream) {
871            error!(
872                "{} end_stream called with stream {} but expected {:?}",
873                log_context!(self),
874                stream,
875                self.stream
876            );
877            return;
878        }
879        // Reached only on the matched-stream path: the connection's active
880        // stream is exactly the one being ended.
881        debug_assert_eq!(
882            self.stream,
883            Some(stream),
884            "end_stream past the guard must target the active stream"
885        );
886        context.unlink_stream(stream);
887        // Post: whatever backend token this stream was Linked to no longer
888        // lists it in the reverse index — `unlink_stream` is the single
889        // eviction point, so a subsequent end/close cannot double-remove it.
890        // (The `state` field is still `Linked` here; the arms below retire it.)
891        #[cfg(debug_assertions)]
892        if let StreamState::Linked(token) = context.streams[stream].state {
893            debug_assert!(
894                context
895                    .backend_streams
896                    .get(&token)
897                    .is_none_or(|ids| !ids.contains(&stream)),
898                "unlink_stream must evict the stream from the backend reverse index"
899            );
900        }
901        let answers_rc = context.listener.borrow().get_answers().clone();
902        let stream_id = stream;
903        let stream = &mut context.streams[stream_id];
904        let stream_context = &mut stream.context;
905        trace!(
906            "{} end H1 stream {:?}: {:#?}",
907            log_context!(self),
908            self.stream,
909            stream_context
910        );
911        match &mut self.position {
912            Position::Client(_, _, BackendStatus::Connecting(_)) => {
913                self.stream = None;
914                if stream.state != StreamState::Recycle {
915                    stream.state = StreamState::Unlinked;
916                }
917                // Post: the client connection detaches from its stream and the
918                // slot is retired (Unlinked) unless already marked for reuse —
919                // never left dangling in an open/Linked state.
920                debug_assert!(
921                    self.stream.is_none(),
922                    "client end_stream must detach the stream"
923                );
924                debug_assert!(
925                    !matches!(stream.state, StreamState::Linked(_)),
926                    "detached stream must not remain Linked"
927                );
928                self.readiness.interest.remove(Ready::ALL);
929                self.force_disconnect();
930            }
931            Position::Client(_, _, status @ BackendStatus::Connected) => {
932                self.stream = None;
933                if stream.state != StreamState::Recycle {
934                    stream.state = StreamState::Unlinked;
935                }
936                debug_assert!(
937                    self.stream.is_none(),
938                    "client end_stream must detach the stream"
939                );
940                debug_assert!(
941                    !matches!(stream.state, StreamState::Linked(_)),
942                    "detached stream must not remain Linked"
943                );
944                self.readiness.interest.remove(Ready::ALL);
945                // keep alive should probably be used only if the http context is fully reset
946                // in case end_stream occurs due to an error the connection state is probably
947                // unrecoverable and should be terminated
948                if stream_context.keep_alive_backend && stream.back.is_terminated() {
949                    *status = BackendStatus::KeepAlive;
950                } else {
951                    self.force_disconnect();
952                }
953            }
954            Position::Client(_, _, BackendStatus::KeepAlive)
955            | Position::Client(_, _, BackendStatus::Disconnecting) => {
956                error!(
957                    "{} end_stream called on KeepAlive or Disconnecting H1 client",
958                    log_context!(self)
959                );
960            }
961            Position::Server => match end_stream_decision(stream) {
962                EndStreamAction::ForwardTerminated => {
963                    debug!("{} CLOSING H1 TERMINATED STREAM", log_context!(self));
964                    stream.state = StreamState::Unlinked;
965                    self.readiness.interest.insert(Ready::WRITABLE);
966                    // End-of-stream was already queued into kawa by the parser;
967                    // no fresh WRITABLE event will arrive from the kernel.
968                    self.readiness.signal_pending_write();
969                }
970                EndStreamAction::CloseDelimited => {
971                    debug!("{} CLOSE DELIMITED", log_context!(self));
972                    stream.state = StreamState::Unlinked;
973                    self.readiness.arm_writable();
974                }
975                EndStreamAction::ForwardUnterminated => {
976                    debug!("{} CLOSING H1 UNTERMINATED STREAM", log_context!(self));
977                    forcefully_terminate_answer(
978                        stream,
979                        &mut self.readiness,
980                        H2Error::InternalError,
981                    );
982                }
983                EndStreamAction::SendDefault(status) => {
984                    let answers = answers_rc.borrow();
985                    set_default_answer(stream, &mut self.readiness, status, &answers);
986                }
987                EndStreamAction::Reconnect => {
988                    debug!("{} H1 RECONNECT", log_context!(self));
989                    stream.state = StreamState::Link;
990                    context.pending_links.push_back(stream_id);
991                }
992            },
993        }
994    }
995
996    pub fn start_stream<L>(&mut self, stream: GlobalStreamId, _context: &mut Context<L>) -> bool
997    where
998        L: ListenerHandler + L7ListenerHandler,
999    {
1000        trace!(
1001            "{} start H1 stream {} {:?}",
1002            log_context!(self),
1003            stream,
1004            self.readiness
1005        );
1006        self.readiness.interest.insert(Ready::ALL);
1007        self.stream = Some(stream);
1008        debug_assert_eq!(
1009            self.stream,
1010            Some(stream),
1011            "start_stream must pin the connection's active stream"
1012        );
1013        match &mut self.position {
1014            Position::Client(_, _, status @ BackendStatus::KeepAlive) => {
1015                *status = BackendStatus::Connected;
1016                // A keep-alive client transitions to Connected when it picks up
1017                // a new stream; it must not stay parked in KeepAlive.
1018                debug_assert!(
1019                    matches!(
1020                        self.position,
1021                        Position::Client(_, _, BackendStatus::Connected)
1022                    ),
1023                    "a reused keep-alive client must become Connected on start_stream"
1024                );
1025            }
1026            Position::Client(_, _, BackendStatus::Disconnecting) => {
1027                error!(
1028                    "{} start_stream called on Disconnecting H1 client",
1029                    log_context!(self)
1030                );
1031                return false;
1032            }
1033            Position::Client(_, _, _) => {}
1034            Position::Server => {
1035                error!(
1036                    "{} start_stream must not be called on H1 server connection",
1037                    log_context!(self)
1038                );
1039                return false;
1040            }
1041        }
1042        true
1043    }
1044}
sozu_lib/protocol/mux/h1.rs

sozu_lib/protocol/mux/
h1.rs
