sozu_lib/protocol/kawa_h1/

mod.rs

//! HTTP/1.1 session state on top of Kawa.
//!
//! Owns the H1 session state machine: Kawa-driven request/response parsing,
//! header rewriting via `editor.rs`, default-answer rendering via
//! `answers.rs`, and route lookup via the listener
//! (`cluster_id_from_request`). Long-form lifecycle:
//! `lib/src/protocol/kawa_h1/LIFECYCLE.md` (created in this changeset).

pub mod answers;
pub mod diagnostics;
pub mod editor;
pub mod parser;

use std::{
    cell::RefCell,
    io::ErrorKind,
    net::{Shutdown, SocketAddr},
    rc::{Rc, Weak},
    time::{Duration, Instant},
};

use mio::{Interest, Token, net::TcpStream};
use rusty_ulid::Ulid;
use sozu_command::{
    config::MAX_LOOP_ITERATIONS,
    logging::EndpointRecord,
    proto::command::{Event, EventKind, ListenerType},
};

// use time::{Duration, Instant};
use crate::metrics::names;
use crate::{
    AcceptError, BackendConnectAction, BackendConnectionError, BackendConnectionStatus,
    FrontendFromRequestError, L7ListenerHandler, L7Proxy, ListenerHandler, Protocol, ProxySession,
    Readiness, RetrieveClusterError, SessionIsToBeClosed, SessionMetrics, SessionResult,
    StateResult,
    backends::{Backend, BackendError},
    pool::{Checkout, Pool},
    protocol::{
        SessionState,
        http::{
            answers::DefaultAnswerStream,
            diagnostics::{diagnostic_400_502, diagnostic_413_507},
            editor::HttpContext,
            parser::Method,
        },
        pipe::WebSocketContext,
    },
    retry::RetryPolicy,
    server::{CONN_RETRIES, push_event},
    socket::{SocketHandler, SocketResult, TransportProtocol, stats::socket_rtt},
    sozu_command::{
        logging::{LogContext, ansi_palette},
        ready::Ready,
    },
    timer::TimeoutContainer,
};

/// This macro is defined uniquely in this module to help the tracking of
/// kawa h1 issues inside Sōzu. Colored output: bold bright-white protocol
/// label (uniform across every protocol), light grey `Session` keyword, gray
/// keys, bright white values. The `[ulid - - -]` context comes first to stay
/// aligned with `MUX-*` and `SOCKET` log lines.
///
/// Two invocation forms:
/// - `log_context!(self)` — reads the response parsing phase via
///   [`Http::response_parsing_phase`]. Use when no `&mut self.response_stream`
///   borrow is currently live.
/// - `log_context!(self, response_phase)` — caller supplies an already-resolved
///   `Option<kawa::ParsingPhase>`. Use when a mutable borrow of
///   `self.response_stream` (or its `BackendAnswer` inner) is live — pass
///   `Some(inner.parsing_phase)` through the existing mutable reference.
macro_rules! log_context {
    ($self:expr) => {
        log_context!($self, $self.response_parsing_phase())
    };
    ($self:expr, $response_phase:expr) => {{
        let (open, reset, grey, gray, white) = ansi_palette();
        format!(
            "{gray}{ctx}{reset}\t{open}KAWA-H1{reset}\t{grey}Session{reset}({gray}public{reset}={white}{public}{reset}, {gray}session{reset}={white}{session}{reset}, {gray}frontend{reset}={white}{frontend}{reset}, {gray}request_parsing_phase{reset}={white}{request_parsing_phase:?}{reset}, {gray}response_parsing_phase{reset}={white}{response_parsing_phase:?}{reset}, {gray}frontend_readiness{reset}={white}{frontend_readiness}{reset}, {gray}backend{reset}={white}{backend}{reset}, {gray}backend_readiness{reset}={white}{backend_readiness}{reset})\t >>>",
            open = open,
            reset = reset,
            grey = grey,
            gray = gray,
            white = white,
            ctx = $self.context.log_context(),
            public = $self.context.public_address.to_string(),
            session = $self.context.session_address.map(|addr| addr.to_string()).unwrap_or_else(|| "<none>".to_string()),
            frontend = $self.frontend_token.0,
            request_parsing_phase = $self.request_stream.parsing_phase,
            response_parsing_phase = $response_phase,
            frontend_readiness = $self.frontend_readiness,
            backend = $self.backend_token.map(|token| token.0.to_string()).unwrap_or_else(|| "<none>".to_string()),
            backend_readiness = $self.backend_readiness,
        )
    }};
}

/// Generic Http representation using the Kawa crate using the Checkout of Sozu as buffer
type GenericHttpStream = kawa::Kawa<Checkout>;

impl kawa::AsBuffer for Checkout {
    fn as_buffer(&self) -> &[u8] {
        self.inner.extra()
    }
    fn as_mut_buffer(&mut self) -> &mut [u8] {
        self.inner.extra_mut()
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum DefaultAnswer {
    Answer301 {
        location: String,
    },
    /// RFC 9110 §15.4.3 — temporary redirect; user agents may rewrite POST
    /// to GET on follow. Surfaced by `RedirectPolicy::Found` (#1009).
    Answer302 {
        location: String,
    },
    /// RFC 9110 §15.4.9 — permanent redirect that PRESERVES the request
    /// method on follow (no GET-rewrite on POST). Surfaced by
    /// `RedirectPolicy::PermanentRedirect` (#1009).
    Answer308 {
        location: String,
    },
    Answer400 {
        message: String,
        phase: kawa::ParsingPhaseMarker,
        successfully_parsed: String,
        partially_parsed: String,
        invalid: String,
    },
    Answer401 {
        /// Optional `WWW-Authenticate` value (e.g. `Basic realm="…"`).
        /// When `None` or empty the realm header line is elided by the
        /// template engine so the response stays a bare 401.
        www_authenticate: Option<String>,
    },
    Answer404 {},
    Answer408 {
        duration: String,
    },
    Answer413 {
        message: String,
        phase: kawa::ParsingPhaseMarker,
        capacity: usize,
    },
    /// RFC 9110 §15.5.20 — returned when the request's `:authority` / `Host`
    /// host does not match the TLS SNI negotiated for this connection.
    /// The peer may retry on a fresh TLS connection that negotiates an SNI
    /// matching the authority.
    Answer421 {},
    /// RFC 6585 §4 — emitted when the per-(cluster, source-IP) connection
    /// limit is reached. `retry_after` is the suggested wait, in seconds;
    /// `None` (or `Some(0)`) tells the template engine to omit the header
    /// — `Retry-After: 0` invites an immediate retry that defeats the
    /// limit.
    Answer429 {
        retry_after: Option<u32>,
    },
    Answer502 {
        message: String,
        phase: kawa::ParsingPhaseMarker,
        successfully_parsed: String,
        partially_parsed: String,
        invalid: String,
    },
    Answer503 {
        message: String,
    },
    Answer504 {
        duration: String,
    },
    Answer507 {
        phase: kawa::ParsingPhaseMarker,
        message: String,
        capacity: usize,
    },
}

impl From<&DefaultAnswer> for u16 {
    fn from(answer: &DefaultAnswer) -> u16 {
        match answer {
            DefaultAnswer::Answer301 { .. } => 301,
            DefaultAnswer::Answer302 { .. } => 302,
            DefaultAnswer::Answer308 { .. } => 308,
            DefaultAnswer::Answer400 { .. } => 400,
            DefaultAnswer::Answer401 { .. } => 401,
            DefaultAnswer::Answer404 { .. } => 404,
            DefaultAnswer::Answer408 { .. } => 408,
            DefaultAnswer::Answer413 { .. } => 413,
            DefaultAnswer::Answer421 { .. } => 421,
            DefaultAnswer::Answer429 { .. } => 429,
            DefaultAnswer::Answer502 { .. } => 502,
            DefaultAnswer::Answer503 { .. } => 503,
            DefaultAnswer::Answer504 { .. } => 504,
            DefaultAnswer::Answer507 { .. } => 507,
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TimeoutStatus {
    Request,
    Response,
    WaitingForNewRequest,
    WaitingForResponse,
}

pub enum ResponseStream {
    BackendAnswer(GenericHttpStream),
    DefaultAnswer(u16, DefaultAnswerStream),
}

/// Http will be contained in State which itself is contained by Session
pub struct Http<Front: SocketHandler, L: ListenerHandler + L7ListenerHandler> {
    answers: Rc<RefCell<answers::HttpAnswers>>,
    pub backend: Option<Rc<RefCell<Backend>>>,
    backend_connection_status: BackendConnectionStatus,
    pub backend_readiness: Readiness,
    pub backend_socket: Option<TcpStream>,
    backend_stop: Option<Instant>,
    pub backend_token: Option<Token>,
    pub container_backend_timeout: TimeoutContainer,
    pub container_frontend_timeout: TimeoutContainer,
    configured_backend_timeout: Duration,
    configured_connect_timeout: Duration,
    configured_frontend_timeout: Duration,
    /// attempts to connect to the backends during the session
    connection_attempts: u8,
    pub frontend_readiness: Readiness,
    pub frontend_socket: Front,
    frontend_token: Token,
    keepalive_count: usize,
    listener: Rc<RefCell<L>>,
    pub request_stream: GenericHttpStream,
    pub response_stream: ResponseStream,
    /// The HTTP context was separated from the State for borrowing reasons.
    /// Calling a kawa parser mutably borrows the State through request_stream or response_stream,
    /// so Http can't be borrowed again to be used in callbacks. HttContext is an independant
    /// subsection of Http that can be mutably borrowed for parser callbacks.
    pub context: HttpContext,
}

impl<Front: SocketHandler, L: ListenerHandler + L7ListenerHandler> Http<Front, L> {
    /// Instantiate a new HTTP SessionState with:
    ///
    /// - frontend_interest: READABLE | HUP | ERROR
    /// - frontend_event: EMPTY
    /// - backend_interest: EMPTY
    /// - backend_event: EMPTY
    ///
    /// Remember to set the events from the previous State!
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        answers: Rc<RefCell<answers::HttpAnswers>>,
        configured_backend_timeout: Duration,
        configured_connect_timeout: Duration,
        configured_frontend_timeout: Duration,
        container_frontend_timeout: TimeoutContainer,
        frontend_socket: Front,
        frontend_token: Token,
        listener: Rc<RefCell<L>>,
        pool: Weak<RefCell<Pool>>,
        protocol: Protocol,
        public_address: SocketAddr,
        session_id: Ulid,
        request_id: Ulid,
        session_address: Option<SocketAddr>,
        sticky_name: String,
    ) -> Result<Http<Front, L>, AcceptError> {
        let (front_buffer, back_buffer) = match pool.upgrade() {
            Some(pool) => {
                let mut pool = pool.borrow_mut();
                match (pool.checkout(), pool.checkout()) {
                    (Some(front_buffer), Some(back_buffer)) => (front_buffer, back_buffer),
                    _ => return Err(AcceptError::BufferCapacityReached),
                }
            }
            None => return Err(AcceptError::BufferCapacityReached),
        };
        let sozu_id_header = listener.borrow().get_sozu_id_header().to_string();
        let elide_x_real_ip = listener.borrow().get_elide_x_real_ip();
        let send_x_real_ip = listener.borrow().get_send_x_real_ip();
        Ok(Http {
            answers,
            backend_connection_status: BackendConnectionStatus::NotConnected,
            backend_readiness: Readiness::new(),
            backend_socket: None,
            backend_stop: None,
            backend_token: None,
            backend: None,
            configured_backend_timeout,
            configured_connect_timeout,
            configured_frontend_timeout,
            connection_attempts: 0,
            container_backend_timeout: TimeoutContainer::new_empty(configured_connect_timeout),
            container_frontend_timeout,
            frontend_readiness: Readiness {
                interest: Ready::READABLE | Ready::HUP | Ready::ERROR,
                event: Ready::EMPTY,
            },
            frontend_socket,
            frontend_token,
            keepalive_count: 0,
            listener,
            request_stream: GenericHttpStream::new(
                kawa::Kind::Request,
                kawa::Buffer::new(front_buffer),
            ),
            response_stream: ResponseStream::BackendAnswer(GenericHttpStream::new(
                kawa::Kind::Response,
                kawa::Buffer::new(back_buffer),
            )),
            context: HttpContext::new(
                session_id,
                request_id,
                protocol,
                public_address,
                session_address,
                sticky_name,
                sozu_id_header,
                elide_x_real_ip,
                send_x_real_ip,
            ),
        })
    }

    /// Reset the connection in case of keep-alive to be ready for the next request
    pub fn reset(&mut self) {
        trace!("{} ============== reset", log_context!(self));
        let response_stream = match &mut self.response_stream {
            ResponseStream::BackendAnswer(response_stream) => response_stream,
            _ => return,
        };

        self.context.id = Ulid::generate();
        self.context.reset();

        self.request_stream.clear();
        response_stream.clear();
        // Pair-assert the keep-alive counter advances by exactly one per
        // completed request: it gates the `WaitingForNewRequest` timeout-status
        // branch and the keepalive-vs-error accounting in `readable`. A
        // double-increment or stall would mis-attribute the next idle close.
        let keepalive_before = self.keepalive_count;
        self.keepalive_count += 1;
        debug_assert_eq!(
            self.keepalive_count,
            keepalive_before + 1,
            "reset() must advance keepalive_count by exactly one"
        );
        // The response side has just been cleared, so it is back to its
        // initial parsing phase — `writable` must not see a half-consumed
        // response when the next request starts.
        debug_assert!(
            response_stream.is_initial(),
            "response stream must be reset to initial on keep-alive"
        );
        gauge_add!(names::http::ACTIVE_REQUESTS, -1);

        if let Some(backend) = &mut self.backend {
            let mut backend = backend.borrow_mut();
            backend.active_requests = backend.active_requests.saturating_sub(1);
        }

        // reset the front timeout and cancel the back timeout while we are
        // waiting for a new request
        self.container_backend_timeout.cancel();
        self.container_frontend_timeout
            .set_duration(self.configured_frontend_timeout);
        self.frontend_readiness.interest = Ready::READABLE | Ready::HUP | Ready::ERROR;
        self.backend_readiness.interest = Ready::HUP | Ready::ERROR;

        // We are resetting the offset of request and response stream buffers
        // We do have to keep cursor position on the request, if there is data
        // in the request stream to preserve http pipelining.

        // Print the left-over response buffer output to track in which case it
        // may happens
        let response_phase = response_stream.parsing_phase;
        let response_storage = &mut response_stream.storage;
        if !response_storage.is_empty() {
            warn!(
                "{} Leftover fragment from response: {}",
                log_context!(self, Some(response_phase)),
                parser::view(
                    response_storage.used(),
                    16,
                    &[response_storage.start, response_storage.end,],
                )
            );
        }

        response_storage.clear();
        if !self.request_stream.storage.is_empty() {
            self.frontend_readiness.event.insert(Ready::READABLE);
        } else {
            self.request_stream.storage.clear();
        }
    }

    pub fn readable(&mut self, metrics: &mut SessionMetrics) -> StateResult {
        trace!("{} ============== readable", log_context!(self));
        if !self.container_frontend_timeout.reset() {
            error!(
                "could not reset front timeout {:?}",
                self.configured_frontend_timeout
            );
            self.print_state(self.protocol_string());
        }

        let response_stream = match &mut self.response_stream {
            ResponseStream::BackendAnswer(response_stream) => response_stream,
            ResponseStream::DefaultAnswer(..) => {
                error!(
                    "{} Sending default answer, should not read from frontend socket",
                    log_context!(self)
                );

                self.frontend_readiness.interest.remove(Ready::READABLE);
                self.frontend_readiness.interest.insert(Ready::WRITABLE);
                return StateResult::Continue;
            }
        };

        if self.request_stream.storage.is_full() {
            self.frontend_readiness.interest.remove(Ready::READABLE);
            if self.request_stream.is_main_phase() {
                self.backend_readiness.interest.insert(Ready::WRITABLE);
            } else {
                // client has filled its buffer and we can't empty it
                self.set_answer(DefaultAnswer::Answer413 {
                    capacity: self.request_stream.storage.capacity(),
                    phase: self.request_stream.parsing_phase.marker(),
                    message: diagnostic_413_507(self.request_stream.parsing_phase),
                });
            }
            return StateResult::Continue;
        }

        let available_space = self.request_stream.storage.available_space();
        let (size, socket_state) = self
            .frontend_socket
            .socket_read(self.request_stream.storage.space());

        debug!(
            "{} Read {} bytes",
            log_context!(self, Some(response_stream.parsing_phase)),
            size
        );

        // A read can never deliver more bytes than the buffer slice we handed
        // it; `storage.fill(size)` below trusts this to advance the write
        // cursor, and an over-long `size` would push the cursor past the
        // buffer end (OOB on the next parse).
        debug_assert!(
            size <= available_space,
            "socket_read returned {size} bytes for a {available_space}-byte buffer slice"
        );

        if size > 0 {
            let used_before = self.request_stream.storage.used().len();
            self.request_stream.storage.fill(size);
            debug_assert_eq!(
                self.request_stream.storage.used().len(),
                used_before + size,
                "storage.fill(size) must grow the used region by exactly the bytes read"
            );
            count!(names::backend::BYTES_IN, size as i64);
            metrics.bin += size;
            // if self.kawa_request.storage.is_full() {
            //     self.frontend_readiness.interest.remove(Ready::READABLE);
            // }
        } else {
            self.frontend_readiness.event.remove(Ready::READABLE);
        }

        match socket_state {
            SocketResult::Error | SocketResult::Closed => {
                if self.request_stream.is_initial() {
                    // count an error if we were waiting for the first request
                    // otherwise, if we already had one completed request and response,
                    // and are waiting for the next one, we do not count a socket
                    // closing abruptly as an error
                    if self.keepalive_count == 0 {
                        self.frontend_socket.read_error();
                    }
                } else {
                    self.frontend_socket.read_error();
                    self.log_request_error(
                        metrics,
                        &format!(
                            "front socket {socket_state:?}, closing the session. Readiness: {:?} -> {:?}, read {size} bytes",
                            self.frontend_readiness,
                            self.backend_readiness,
                        )
                    );
                }
                return StateResult::CloseSession;
            }
            SocketResult::WouldBlock => {
                self.frontend_readiness.event.remove(Ready::READABLE);
            }
            SocketResult::Continue => {}
        };

        trace!(
            "{} ============== readable_parse",
            log_context!(self, Some(response_stream.parsing_phase))
        );
        let was_initial = self.request_stream.is_initial();
        let was_not_proxying = !self.request_stream.is_main_phase();

        kawa::h1::parse(&mut self.request_stream, &mut self.context);
        // kawa::debug_kawa(&self.request_stream);

        if was_initial && !self.request_stream.is_initial() {
            // if it was the first request, the front timeout duration
            // was set to request_timeout, which is much lower. For future
            // requests on this connection, we can wait a bit more
            self.container_frontend_timeout
                .set_duration(self.configured_frontend_timeout);
            gauge_add!(names::http::ACTIVE_REQUESTS, 1);
            incr!(names::http::REQUESTS);
        }

        if let kawa::ParsingPhase::Error { marker, kind } = self.request_stream.parsing_phase {
            incr!(names::http::FRONTEND_PARSE_ERRORS);
            warn!(
                "{} Parsing request error in {:?}: {}",
                log_context!(self, Some(response_stream.parsing_phase)),
                marker,
                match kind {
                    kawa::ParsingErrorKind::Consuming { index } => {
                        let kawa = &self.request_stream;
                        parser::view(
                            kawa.storage.used(),
                            16,
                            &[
                                kawa.storage.start,
                                kawa.storage.head,
                                index as usize,
                                kawa.storage.end,
                            ],
                        )
                    }
                    kawa::ParsingErrorKind::Processing { message } => message.to_owned(),
                }
            );
            if response_stream.consumed {
                self.log_request_error(metrics, "Parsing error on the request");
                return StateResult::CloseSession;
            } else {
                let (message, successfully_parsed, partially_parsed, invalid) =
                    diagnostic_400_502(marker, kind, &self.request_stream);
                self.set_answer(DefaultAnswer::Answer400 {
                    message,
                    phase: marker,
                    successfully_parsed,
                    partially_parsed,
                    invalid,
                });
                return StateResult::Continue;
            }
        }

        if self.request_stream.is_main_phase() {
            self.backend_readiness.interest.insert(Ready::WRITABLE);
            if was_not_proxying {
                // Sozu tries to connect only once all the headers were gathered and edited
                // this could be improved
                trace!("{} ============== HANDLE CONNECTION!", log_context!(self));
                return StateResult::ConnectBackend;
            }
        }
        if self.request_stream.is_terminated() {
            self.frontend_readiness.interest.remove(Ready::READABLE);
        }

        StateResult::Continue
    }

    pub fn writable(&mut self, metrics: &mut SessionMetrics) -> StateResult {
        trace!("{} ============== writable", log_context!(self));
        let response_stream = match &mut self.response_stream {
            ResponseStream::BackendAnswer(response_stream) => response_stream,
            _ => return self.writable_default_answer(metrics),
        };

        response_stream.prepare(&mut kawa::h1::BlockConverter);

        let bufs = response_stream.as_io_slice();
        if bufs.is_empty() && !self.frontend_socket.socket_wants_write() {
            self.frontend_readiness.interest.remove(Ready::WRITABLE);
            // do not shortcut, response might have been terminated without anything more to send
        }

        let (size, socket_state) = self.frontend_socket.socket_write_vectored(&bufs);

        debug!(
            "{} Wrote {} bytes",
            log_context!(self, Some(response_stream.parsing_phase)),
            size
        );

        if size > 0 {
            // A vectored write can only have flushed bytes that were actually
            // queued in `bufs`; consuming more than we sent would advance the
            // kawa output cursor past the data the peer received and truncate
            // the next chunk. (`as_io_slice` is built from the same stream we
            // now consume from.)
            let queued = bufs.iter().map(|b| b.len()).sum::<usize>();
            debug_assert!(
                size <= queued,
                "socket_write_vectored reported {size} bytes for {queued} queued"
            );
            response_stream.consume(size);
            count!(names::backend::BYTES_OUT, size as i64);
            metrics.bout += size;
            self.backend_readiness.interest.insert(Ready::READABLE);
        }

        match socket_state {
            SocketResult::Error | SocketResult::Closed => {
                self.frontend_socket.write_error();
                self.log_request_error(
                    metrics,
                    &format!(
                        "front socket {socket_state:?}, closing session.  Readiness: {:?} -> {:?}, read {size} bytes",
                        self.frontend_readiness,
                        self.backend_readiness,
                    ),
                );
                return StateResult::CloseSession;
            }
            SocketResult::WouldBlock => {
                self.frontend_readiness.event.remove(Ready::WRITABLE);
            }
            SocketResult::Continue => {}
        }

        if self.frontend_socket.socket_wants_write() {
            return StateResult::Continue;
        }

        if response_stream.is_terminated() && response_stream.is_completed() {
            if self.context.closing {
                debug!("{} closing proxy, no keep alive", log_context!(self));
                self.log_request_success(metrics);
                return StateResult::CloseSession;
            }

            match response_stream.detached.status_line {
                kawa::StatusLine::Response { code: 101, .. } => {
                    trace!("{} ============== HANDLE UPGRADE!", log_context!(self));
                    self.log_request_success(metrics);
                    return StateResult::Upgrade;
                }
                kawa::StatusLine::Response { code: 100, .. } => {
                    trace!(
                        "{} ============== HANDLE CONTINUE!",
                        log_context!(self, Some(response_stream.parsing_phase))
                    );
                    response_stream.clear();
                    self.log_request_success(metrics);
                    return StateResult::Continue;
                }
                kawa::StatusLine::Response { code: 103, .. } => {
                    self.backend_readiness.event.insert(Ready::READABLE);
                    trace!(
                        "{} ============== HANDLE EARLY HINT!",
                        log_context!(self, Some(response_stream.parsing_phase))
                    );
                    response_stream.clear();
                    self.log_request_success(metrics);
                    return StateResult::Continue;
                }
                _ => (),
            }

            let response_length_known = response_stream.body_size != kawa::BodySize::Empty;
            let request_length_known = self.request_stream.body_size != kawa::BodySize::Empty;
            if !(self.request_stream.is_terminated() && self.request_stream.is_completed())
                && request_length_known
            {
                error!(
                    "{} Response terminated before request, this case is not handled properly yet",
                    log_context!(self)
                );
                incr!(names::http::EARLY_RESPONSE_CLOSE);
                // FIXME: this will cause problems with pipelining
                // return StateResult::CloseSession;
            }

            // FIXME: we could get smarter about this
            // with no keepalive on backend, we could open a new backend ConnectionError
            // with no keepalive on front but keepalive on backend, we could have
            // a pool of connections
            trace!(
                "{} ============== HANDLE KEEP-ALIVE: {} {} {}",
                log_context!(self),
                self.context.keep_alive_frontend,
                self.context.keep_alive_backend,
                response_length_known
            );

            self.log_request_success(metrics);
            return match (
                self.context.keep_alive_frontend,
                self.context.keep_alive_backend,
                response_length_known,
            ) {
                (true, true, true) => {
                    debug!("{} Keep alive frontend/backend", log_context!(self));
                    metrics.reset();
                    self.reset();
                    StateResult::Continue
                }
                (true, false, true) => {
                    debug!("{} Keep alive frontend", log_context!(self));
                    metrics.reset();
                    self.reset();
                    StateResult::CloseBackend
                }
                _ => {
                    debug!("{} No keep alive", log_context!(self));
                    StateResult::CloseSession
                }
            };
        }
        StateResult::Continue
    }

    fn writable_default_answer(&mut self, metrics: &mut SessionMetrics) -> StateResult {
        trace!(
            "{} ============== writable_default_answer",
            log_context!(self)
        );
        let response_stream = match &mut self.response_stream {
            ResponseStream::DefaultAnswer(_, response_stream) => response_stream,
            _ => return StateResult::CloseSession,
        };
        let bufs = response_stream.as_io_slice();
        let (size, socket_state) = self.frontend_socket.socket_write_vectored(&bufs);

        // Never consume more of the rendered default answer than the socket
        // actually accepted, or the next flush would skip bytes and ship a
        // truncated error page.
        let queued = bufs.iter().map(|b| b.len()).sum::<usize>();
        debug_assert!(
            size <= queued,
            "default-answer write reported {size} bytes for {queued} queued"
        );
        count!(names::backend::BYTES_OUT, size as i64);
        metrics.bout += size;
        response_stream.consume(size);

        if size == 0 || socket_state != SocketResult::Continue {
            self.frontend_readiness.event.remove(Ready::WRITABLE);
        }

        if response_stream.is_completed() {
            save_http_status_metric(self.context.status, self.context.log_context());
            self.log_default_answer_success(metrics);
            self.frontend_readiness.reset();
            self.backend_readiness.reset();
            return StateResult::CloseSession;
        }

        if socket_state == SocketResult::Error {
            self.frontend_socket.write_error();
            self.log_request_error(
                metrics,
                "error writing default answer to front socket, closing",
            );
            StateResult::CloseSession
        } else {
            StateResult::Continue
        }
    }

    pub fn backend_writable(&mut self, metrics: &mut SessionMetrics) -> SessionResult {
        trace!("{} ============== backend_writable", log_context!(self));
        if let ResponseStream::DefaultAnswer(..) = self.response_stream {
            error!(
                "{}\tsending default answer, should not write to back",
                log_context!(self)
            );
            self.backend_readiness.interest.remove(Ready::WRITABLE);
            self.frontend_readiness.interest.insert(Ready::WRITABLE);
            return SessionResult::Continue;
        }

        let backend_socket = if let Some(backend_socket) = &mut self.backend_socket {
            backend_socket
        } else {
            self.log_request_error(metrics, "back socket not found, closing session");
            return SessionResult::Close;
        };

        self.request_stream.prepare(&mut kawa::h1::BlockConverter);

        let bufs = self.request_stream.as_io_slice();
        if bufs.is_empty() {
            self.backend_readiness.interest.remove(Ready::WRITABLE);
            return SessionResult::Continue;
        }

        let (size, socket_state) = backend_socket.socket_write_vectored(&bufs);
        debug!("{} Wrote {} bytes", log_context!(self), size);

        if size > 0 {
            // Consume only what the backend socket actually accepted from the
            // queued request slices — over-consuming would drop request bytes
            // the backend never saw (silent request truncation / smuggling).
            let queued = bufs.iter().map(|b| b.len()).sum::<usize>();
            debug_assert!(
                size <= queued,
                "backend socket_write_vectored reported {size} bytes for {queued} queued"
            );
            self.request_stream.consume(size);
            count!(names::backend::BACK_BYTES_OUT, size as i64);
            metrics.backend_bout += size;
            self.frontend_readiness.interest.insert(Ready::READABLE);
            self.backend_readiness.interest.insert(Ready::READABLE);
        } else {
            self.backend_readiness.event.remove(Ready::WRITABLE);
        }

        match socket_state {
            // the back socket is not writable anymore, so we can drop
            // the front buffer, no more data can be transmitted.
            // But the socket might still be readable, or if it is
            // closed, we might still have some data in the buffer.
            // As an example, we can get an early response for a large
            // POST request to refuse it and prevent all of the data
            // from being transmitted, with the backend server closing
            // the socket right after sending the response
            // FIXME: shouldn't we test for response_state then?
            SocketResult::Error | SocketResult::Closed => {
                self.frontend_readiness.interest.remove(Ready::READABLE);
                self.backend_readiness.interest.remove(Ready::WRITABLE);
                return SessionResult::Continue;
            }
            SocketResult::WouldBlock => {
                self.backend_readiness.event.remove(Ready::WRITABLE);
            }
            SocketResult::Continue => {}
        }

        if self.request_stream.is_terminated() && self.request_stream.is_completed() {
            self.backend_readiness.interest.remove(Ready::WRITABLE);

            // cancel the front timeout while we are waiting for the server to answer
            self.container_frontend_timeout.cancel();
            self.container_backend_timeout.reset();
        }
        SessionResult::Continue
    }

    // Read content from cluster
    pub fn backend_readable(&mut self, metrics: &mut SessionMetrics) -> SessionResult {
        trace!("{} ============== backend_readable", log_context!(self));
        if !self.container_backend_timeout.reset() {
            error!(
                "{} Could not reset back timeout {:?}",
                log_context!(self),
                self.configured_backend_timeout
            );
            self.print_state(self.protocol_string());
        }

        let response_stream = match &mut self.response_stream {
            ResponseStream::BackendAnswer(response_stream) => response_stream,
            _ => {
                error!(
                    "{} Sending default answer, should not read from backend socket",
                    log_context!(self),
                );

                self.backend_readiness.interest.remove(Ready::READABLE);
                self.frontend_readiness.interest.insert(Ready::WRITABLE);
                return SessionResult::Continue;
            }
        };

        let backend_socket = if let Some(backend_socket) = &mut self.backend_socket {
            backend_socket
        } else {
            self.log_request_error(metrics, "back socket not found, closing session");
            return SessionResult::Close;
        };

        if response_stream.storage.is_full() {
            self.backend_readiness.interest.remove(Ready::READABLE);
            if response_stream.is_main_phase() {
                self.frontend_readiness.interest.insert(Ready::WRITABLE);
            } else {
                // server has filled its buffer and we can't empty it
                let capacity = response_stream.storage.capacity();
                let phase = response_stream.parsing_phase.marker();
                let message = diagnostic_413_507(response_stream.parsing_phase);
                self.set_answer(DefaultAnswer::Answer507 {
                    capacity,
                    phase,
                    message,
                });
            }
            return SessionResult::Continue;
        }

        let available_space = response_stream.storage.available_space();
        let (size, socket_state) = backend_socket.socket_read(response_stream.storage.space());
        debug!(
            "{} Read {} bytes",
            log_context!(self, Some(response_stream.parsing_phase)),
            size
        );

        // Same buffer-slice bound as the frontend read path: a backend read
        // can never exceed the space slice we handed it, or `fill(size)` would
        // push the response write cursor past the buffer end.
        debug_assert!(
            size <= available_space,
            "backend socket_read returned {size} bytes for a {available_space}-byte slice"
        );

        if size > 0 {
            let used_before = response_stream.storage.used().len();
            response_stream.storage.fill(size);
            debug_assert_eq!(
                response_stream.storage.used().len(),
                used_before + size,
                "response storage.fill(size) must grow the used region by exactly the bytes read"
            );
            count!(names::backend::BACK_BYTES_IN, size as i64);
            metrics.backend_bin += size;
            // if self.kawa_response.storage.is_full() {
            //     self.backend_readiness.interest.remove(Ready::READABLE);
            // }

            // In case the response starts while the request is not tagged as "terminated",
            // we place the timeout responsibility on the backend.
            // This can happen when:
            // - the request is malformed and doesn't have length information
            // - kawa fails to detect a properly terminated request (e.g. a GET request with no body and no length)
            // - the response can start before the end of the request (e.g. stream processing like compression)
            self.container_frontend_timeout.cancel();
        } else {
            self.backend_readiness.event.remove(Ready::READABLE);
        }

        // TODO: close delimited and backend_hup should be handled better
        match socket_state {
            SocketResult::Error => {
                backend_socket.read_error();
                self.log_request_error(
                    metrics,
                    &format!(
                        "back socket {socket_state:?}, closing session.  Readiness: {:?} -> {:?}, read {size} bytes",
                        self.frontend_readiness,
                        self.backend_readiness,
                    ),
                );
                return SessionResult::Close;
            }
            SocketResult::WouldBlock | SocketResult::Closed => {
                self.backend_readiness.event.remove(Ready::READABLE);
            }
            SocketResult::Continue => {}
        }

        trace!(
            "{} ============== backend_readable_parse",
            log_context!(self, Some(response_stream.parsing_phase))
        );
        kawa::h1::parse(response_stream, &mut self.context);
        // kawa::debug_kawa(&self.response_stream);

        if let kawa::ParsingPhase::Error { marker, kind } = response_stream.parsing_phase {
            incr!(names::http::BACKEND_PARSE_ERRORS);
            warn!(
                "{} Parsing response error in {:?}: {}",
                log_context!(self, Some(response_stream.parsing_phase)),
                marker,
                match kind {
                    kawa::ParsingErrorKind::Consuming { index } => {
                        parser::view(
                            response_stream.storage.used(),
                            16,
                            &[
                                response_stream.storage.start,
                                response_stream.storage.head,
                                index as usize,
                                response_stream.storage.end,
                            ],
                        )
                    }
                    kawa::ParsingErrorKind::Processing { message } => message.to_owned(),
                }
            );
            if response_stream.consumed {
                return SessionResult::Close;
            } else {
                let (message, successfully_parsed, partially_parsed, invalid) =
                    diagnostic_400_502(marker, kind, response_stream);
                self.set_answer(DefaultAnswer::Answer502 {
                    message,
                    phase: marker,
                    successfully_parsed,
                    partially_parsed,
                    invalid,
                });
                return SessionResult::Continue;
            }
        }

        if response_stream.is_main_phase() {
            self.frontend_readiness.interest.insert(Ready::WRITABLE);
        }
        if response_stream.is_terminated() {
            metrics.backend_stop();
            self.backend_stop = Some(Instant::now());
            self.backend_readiness.interest.remove(Ready::READABLE);
        }
        SessionResult::Continue
    }
}

impl<Front: SocketHandler, L: ListenerHandler + L7ListenerHandler> Http<Front, L> {
    fn log_endpoint(&self) -> EndpointRecord<'_> {
        EndpointRecord::Http {
            method: self.context.method.as_deref(),
            authority: self.context.authority.as_deref(),
            path: self.context.path.as_deref(),
            reason: self.context.reason.as_deref(),
            status: self.context.status,
        }
    }

    /// Returns the kawa `ParsingPhase` of the response side, if a backend answer
    /// has been accepted. Used by `log_context!` to surface H1 wedged-response
    /// states without poking inside the `ResponseStream` enum at every log site.
    fn response_parsing_phase(&self) -> Option<kawa::ParsingPhase> {
        match &self.response_stream {
            ResponseStream::BackendAnswer(inner) => Some(inner.parsing_phase),
            _ => None,
        }
    }

    pub fn get_session_address(&self) -> Option<SocketAddr> {
        self.context
            .session_address
            .or_else(|| self.frontend_socket.socket_ref().peer_addr().ok())
    }

    pub fn get_backend_address(&self) -> Option<SocketAddr> {
        self.backend
            .as_ref()
            .map(|backend| backend.borrow().address)
            .or_else(|| {
                self.backend_socket
                    .as_ref()
                    .and_then(|backend| backend.peer_addr().ok())
            })
    }

    // The protocol name used in the access logs
    fn protocol_string(&self) -> &'static str {
        match self.context.protocol {
            Protocol::HTTP => "HTTP",
            Protocol::HTTPS => match self.frontend_socket.protocol() {
                TransportProtocol::Ssl2 => "HTTPS-SSL2",
                TransportProtocol::Ssl3 => "HTTPS-SSL3",
                TransportProtocol::Tls1_0 => "HTTPS-TLS1.0",
                TransportProtocol::Tls1_1 => "HTTPS-TLS1.1",
                TransportProtocol::Tls1_2 => "HTTPS-TLS1.2",
                TransportProtocol::Tls1_3 => "HTTPS-TLS1.3",
                _ => unreachable!(),
            },
            _ => unreachable!(),
        }
    }

    /// Format the context of the websocket into a loggable String
    pub fn websocket_context(&self) -> WebSocketContext {
        WebSocketContext::Http {
            method: self.context.method.clone(),
            authority: self.context.authority.clone(),
            path: self.context.path.clone(),
            reason: self.context.reason.clone(),
            status: self.context.status,
        }
    }

    pub fn log_request(&self, metrics: &SessionMetrics, error: bool, message: Option<&str>) {
        let listener = self.listener.borrow();
        let tags = self.context.authority.as_ref().and_then(|host| {
            let hostname = match host.split_once(':') {
                None => host,
                Some((hostname, _)) => hostname,
            };
            listener.get_tags(hostname)
        });

        let context = self.context.log_context();
        metrics.register_end_of_session(&context);

        log_access! {
            error,
            on_failure: { incr!(names::access_logs::UNSENT) },
            message,
            context,
            session_address: self.get_session_address(),
            backend_address: self.get_backend_address(),
            protocol: self.protocol_string(),
            endpoint: self.log_endpoint(),
            tags,
            client_rtt: socket_rtt(self.front_socket()),
            server_rtt: self.backend_socket.as_ref().and_then(socket_rtt),
            service_time: metrics.service_time(),
            response_time: metrics.backend_response_time(),
            request_time: metrics.request_time(),
            start_time_ns: metrics.start_wall_ns(),
            bytes_in: metrics.bin,
            bytes_out: metrics.bout,
            user_agent: self.context.user_agent.as_deref(),
            x_request_id: self.context.x_request_id.as_deref(),
            tls_version: self.context.tls_version,
            tls_cipher: self.context.tls_cipher,
            tls_sni: self.context.tls_server_name.as_deref(),
            tls_alpn: self.context.tls_alpn,
            xff_chain: self.context.xff_chain.as_deref(),
            #[cfg(feature = "opentelemetry")]
            otel: self.context.otel.as_ref(),
            #[cfg(not(feature = "opentelemetry"))]
            otel: None,
        };
    }

    pub fn log_request_success(&self, metrics: &SessionMetrics) {
        save_http_status_metric(self.context.status, self.context.log_context());
        self.log_request(metrics, false, None);
    }

    pub fn log_default_answer_success(&self, metrics: &SessionMetrics) {
        // Surface the timeout discriminator (set by `timeout()` before
        // `set_answer`) on the access-log `message` field. Non-timeout
        // default-answer paths leave `access_log_message` as `None` and
        // emit `message: None` exactly as before.
        self.log_request(metrics, false, self.context.access_log_message);
    }
    pub fn log_request_error(&self, metrics: &mut SessionMetrics, message: &str) {
        // Labelled with `(cluster_id, backend_id)` so per-cluster dashboards
        // can attribute error rates. The labels are dropped centrally by
        // `filter_labels_for_detail` (`metrics/mod.rs`) when
        // `metrics.detail` is `process` / `frontend` / `cluster`, so default
        // setups still see the same proxy-wide counter — no double-counting,
        // no surprise cardinality.
        incr!(
            "http.errors",
            self.context.cluster_id.as_deref(),
            self.context.backend_id.as_deref()
        );
        error!(
            "{} Could not process request properly got: {}",
            log_context!(self),
            message
        );
        self.print_state(self.protocol_string());
        self.log_request(metrics, true, Some(message));
    }

    pub fn set_answer(&mut self, answer: DefaultAnswer) {
        let status = u16::from(&answer);
        if let ResponseStream::DefaultAnswer(old_status, ..) = self.response_stream {
            error!(
                "already set the default answer to {}, trying to set to {}",
                old_status, status
            );
        } else {
            match answer {
                DefaultAnswer::Answer301 { .. } => incr!(
                    "http.301.redirection",
                    self.context.cluster_id.as_deref(),
                    self.context.backend_id.as_deref()
                ),
                DefaultAnswer::Answer302 { .. } => incr!(
                    "http.302.redirection",
                    self.context.cluster_id.as_deref(),
                    self.context.backend_id.as_deref()
                ),
                DefaultAnswer::Answer308 { .. } => incr!(
                    "http.308.redirection",
                    self.context.cluster_id.as_deref(),
                    self.context.backend_id.as_deref()
                ),
                DefaultAnswer::Answer400 { .. } => incr!(names::http::ERR_400),
                DefaultAnswer::Answer401 { .. } => incr!(
                    "http.401.errors",
                    self.context.cluster_id.as_deref(),
                    self.context.backend_id.as_deref()
                ),
                DefaultAnswer::Answer404 { .. } => incr!(names::http::ERR_404),
                DefaultAnswer::Answer408 { .. } => incr!(
                    "http.408.errors",
                    self.context.cluster_id.as_deref(),
                    self.context.backend_id.as_deref()
                ),
                DefaultAnswer::Answer413 { .. } => incr!(
                    "http.413.errors",
                    self.context.cluster_id.as_deref(),
                    self.context.backend_id.as_deref()
                ),
                DefaultAnswer::Answer421 { .. } => incr!(
                    "http.421.errors",
                    self.context.cluster_id.as_deref(),
                    self.context.backend_id.as_deref()
                ),
                DefaultAnswer::Answer429 { .. } => incr!(
                    "connections.rejected_per_cluster_ip",
                    self.context.cluster_id.as_deref(),
                    self.context.backend_id.as_deref()
                ),
                DefaultAnswer::Answer502 { .. } => incr!(
                    "http.502.errors",
                    self.context.cluster_id.as_deref(),
                    self.context.backend_id.as_deref()
                ),
                DefaultAnswer::Answer503 { .. } => incr!(
                    "http.503.errors",
                    self.context.cluster_id.as_deref(),
                    self.context.backend_id.as_deref()
                ),
                DefaultAnswer::Answer504 { .. } => incr!(
                    "http.504.errors",
                    self.context.cluster_id.as_deref(),
                    self.context.backend_id.as_deref()
                ),
                DefaultAnswer::Answer507 { .. } => incr!(
                    "http.507.errors",
                    self.context.cluster_id.as_deref(),
                    self.context.backend_id.as_deref()
                ),
            };
        }

        let (resolved_status, keep_alive, mut kawa) = self.answers.borrow().get(
            answer,
            self.context.id.to_string(),
            self.context.cluster_id.as_deref(),
            self.context.backend_id.as_deref(),
            self.get_route(),
        );
        // The answer engine only ever resolves to a template whose status line
        // parsed as a 3-digit HTTP code (`Template::new` enforces it); a value
        // outside `100..=999` here would mean a malformed template slipped
        // through and the bytes on the wire would carry a bogus status.
        debug_assert!(
            (100..=999).contains(&resolved_status),
            "default answer must resolve to a 3-digit HTTP status, got {resolved_status}"
        );
        kawa.prepare(&mut kawa::h1::BlockConverter);
        // The template's resolved status may differ from the requested
        // code when a fallback template is selected (e.g. unknown status).
        let status = resolved_status;
        self.context.status = Some(status);
        self.context.reason = None;
        // Honour the rendered template's `Connection` header. Built-in
        // error templates carry `Connection: close`, but operators can
        // ship a custom template that keeps the frontend connection alive.
        if !keep_alive {
            self.context.keep_alive_frontend = false;
        }
        self.response_stream = ResponseStream::DefaultAnswer(status, kawa);
        self.frontend_readiness.interest = Ready::WRITABLE | Ready::HUP | Ready::ERROR;
        self.backend_readiness.interest = Ready::HUP | Ready::ERROR;
        // Post: the session is now in the default-answer regime —
        // `writable_default_answer` flushes the queued bytes and the frontend
        // is steered to WRITABLE (never READABLE) while the backend is parked.
        // This mirrors the `DefaultAnswer` arm of `check_invariants`.
        debug_assert!(
            matches!(self.response_stream, ResponseStream::DefaultAnswer(s, _) if s == status),
            "set_answer must leave the response stream as the queued DefaultAnswer"
        );
        debug_assert!(
            self.frontend_readiness.interest.is_writable()
                && !self.frontend_readiness.interest.is_readable(),
            "default answer must steer the frontend to WRITABLE only"
        );
    }

    pub fn test_backend_socket(&self) -> bool {
        match self.backend_socket {
            Some(ref s) => {
                let mut tmp = [0u8; 1];
                let res = s.peek(&mut tmp[..]);

                match res {
                    // if the socket is half open, it will report 0 bytes read (EOF)
                    Ok(0) => false,
                    Ok(_) => true,
                    Err(e) => matches!(e.kind(), std::io::ErrorKind::WouldBlock),
                }
            }
            None => false,
        }
    }

    pub fn is_valid_backend_socket(&self) -> bool {
        // if socket was last used in the last second, test it
        match self.backend_stop.as_ref() {
            Some(stop_instant) => {
                let now = Instant::now();
                let dur = now - *stop_instant;
                if dur > Duration::from_secs(1) {
                    return self.test_backend_socket();
                }
            }
            None => return self.test_backend_socket(),
        }

        true
    }

    pub fn set_backend_socket(&mut self, socket: TcpStream, backend: Option<Rc<RefCell<Backend>>>) {
        self.backend_socket = Some(socket);
        self.backend = backend;
    }

    pub fn set_cluster_id(&mut self, cluster_id: String) {
        self.context.cluster_id = Some(cluster_id);
    }

    pub fn set_backend_id(&mut self, backend_id: String) {
        self.context.backend_id = Some(backend_id);
    }

    pub fn set_backend_token(&mut self, token: Token) {
        self.backend_token = Some(token);
    }

    pub fn clear_backend_token(&mut self) {
        self.backend_token = None;
    }

    pub fn set_backend_timeout(&mut self, dur: Duration) {
        if let Some(token) = self.backend_token.as_ref() {
            self.container_backend_timeout.set_duration(dur);
            self.container_backend_timeout.set(*token);
        }
    }

    pub fn front_socket(&self) -> &TcpStream {
        self.frontend_socket.socket_ref()
    }

    /// WARNING: this function removes the backend entry in the session manager
    /// IF the backend_token is set, so that entry can be reused for new backend.
    /// I don't think this is a good idea, but it is a quick fix
    fn close_backend(&mut self, proxy: Rc<RefCell<dyn L7Proxy>>, metrics: &mut SessionMetrics) {
        self.container_backend_timeout.cancel();
        debug!(
            "{}\tPROXY [{}->{}] CLOSED BACKEND",
            log_context!(self),
            self.frontend_token.0,
            self.backend_token
                .map(|t| format!("{}", t.0))
                .unwrap_or_else(|| "-".to_string())
        );

        let proxy = proxy.borrow();
        if let Some(socket) = &mut self.backend_socket.take() {
            if let Err(e) = proxy.deregister_socket(socket) {
                error!(
                    "{} Error deregistering back socket({:?}): {:?}",
                    log_context!(self),
                    socket,
                    e
                );
            }
            // SAFETY (TLS-truncation invariant): `Shutdown::Both` is permitted
            // here because this is the *backend* socket of an H1 session — Sōzu
            // currently speaks plaintext H1 to backends, so there is no
            // outbound TLS write buffer that the kernel could discard with a
            // RST. If backend-TLS lands later (#1218), this MUST move to
            // `Shutdown::Write` (or `socket.send_close_notify()`) to avoid the
            // canonical truncation anti-pattern documented in CLAUDE.md.
            if let Err(e) = socket.shutdown(Shutdown::Both) {
                if e.kind() != ErrorKind::NotConnected {
                    error!(
                        "{} Error shutting down back socket({:?}): {:?}",
                        log_context!(self),
                        socket,
                        e
                    );
                }
            }
        }

        if let Some(token) = self.backend_token.take() {
            proxy.remove_session(token);

            if self.backend_connection_status != BackendConnectionStatus::NotConnected {
                self.backend_readiness.event = Ready::EMPTY;
            }

            if self.backend_connection_status == BackendConnectionStatus::Connected {
                gauge_add!(names::backend::CONNECTIONS, -1);
                gauge_add!(
                    names::backend::CONNECTIONS_PER_BACKEND,
                    -1,
                    self.context.cluster_id.as_deref(),
                    metrics.backend_id.as_deref()
                );
            }

            self.set_backend_connected(BackendConnectionStatus::NotConnected, metrics);

            if let Some(backend) = self.backend.take() {
                backend.borrow_mut().dec_connections();
            }
        }
        // Post: the backend token is fully released — `close_backend` is the
        // sole site that decrements the `CONNECTIONS` gauge, paired with the
        // `+1` in `set_backend_connected`. Leaving a token behind would orphan
        // the slab entry and skip the next session's gauge decrement.
        debug_assert!(
            self.backend_token.is_none(),
            "close_backend must release the backend token"
        );
        debug_assert_ne!(
            self.backend_connection_status,
            BackendConnectionStatus::Connected,
            "close_backend must leave the backend out of the Connected state"
        );
    }

    /// Check the number of connection attempts against authorized connection retries
    fn check_circuit_breaker(&mut self) -> Result<(), BackendConnectionError> {
        if self.connection_attempts >= CONN_RETRIES {
            incr!(
                "backend.connect.retries_exhausted",
                self.context.cluster_id.as_deref(),
                self.context.backend_id.as_deref()
            );
            warn!(
                "{} Max connection attempt reached ({})",
                log_context!(self),
                self.connection_attempts,
            );

            self.set_answer(DefaultAnswer::Answer503 {
                message: format!(
                    "Max connection attempt reached: {}",
                    self.connection_attempts
                ),
            });
            return Err(BackendConnectionError::MaxConnectionRetries(None));
        }
        // Post (Ok branch): we only fall through when there is still budget for
        // another connect attempt. `connect_to_backend` relies on this — it
        // proceeds straight to the dial after a clean breaker check.
        debug_assert!(
            self.connection_attempts < CONN_RETRIES,
            "check_circuit_breaker must only return Ok with retry budget remaining"
        );
        Ok(())
    }

    fn check_backend_connection(&mut self, metrics: &mut SessionMetrics) -> bool {
        let is_valid_backend_socket = self.is_valid_backend_socket();

        if !is_valid_backend_socket {
            return false;
        }

        //matched on keepalive
        metrics.backend_id = self.backend.as_ref().map(|i| i.borrow().backend_id.clone());

        metrics.backend_start();
        metrics.backend_connected();
        if let Some(b) = self.backend.as_mut() {
            b.borrow_mut().active_requests += 1;
        }
        true
    }

    // -> host, path, method
    pub fn extract_route(&self) -> Result<(&str, &str, &Method), RetrieveClusterError> {
        let given_method = self
            .context
            .method
            .as_ref()
            .ok_or(RetrieveClusterError::NoMethod)?;
        let given_authority = self
            .context
            .authority
            .as_deref()
            .ok_or(RetrieveClusterError::NoHost)?;
        let given_path = self
            .context
            .path
            .as_deref()
            .ok_or(RetrieveClusterError::NoPath)?;

        Ok((given_authority, given_path, given_method))
    }

    pub fn get_route(&self) -> String {
        if let Some(method) = &self.context.method {
            if let Some(authority) = &self.context.authority {
                if let Some(path) = &self.context.path {
                    return format!("{method} {authority}{path}");
                }
                return format!("{method} {authority}");
            }
            return format!("{method}");
        }
        String::new()
    }

    fn cluster_id_from_request(
        &mut self,
        proxy: Rc<RefCell<dyn L7Proxy>>,
    ) -> Result<String, RetrieveClusterError> {
        let (host, uri, method) = match self.extract_route() {
            Ok(tuple) => tuple,
            Err(cluster_error) => {
                self.set_answer(DefaultAnswer::Answer400 {
                    message: "Could not extract the route after connection started, this should not happen.".into(),
                    phase: self.request_stream.parsing_phase.marker(),
                    successfully_parsed: "null".into(),
                    partially_parsed: "null".into(),
                    invalid: "null".into(),
                });
                return Err(cluster_error);
            }
        };

        let route_result = self
            .listener
            .borrow()
            .frontend_from_request(host, uri, method);

        let route = match route_result {
            Ok(route) => route,
            Err(frontend_error) => {
                // RFC 9110 §15.5.1: a malformed authority is a 400. A
                // syntactically valid authority that simply has no matching
                // frontend stays on the historical 404 path.
                match &frontend_error {
                    FrontendFromRequestError::HostParse { .. }
                    | FrontendFromRequestError::InvalidCharsAfterHost(_) => {
                        self.set_answer(DefaultAnswer::Answer400 {
                            message: frontend_error.to_string(),
                            phase: self.request_stream.parsing_phase.marker(),
                            successfully_parsed: "null".into(),
                            partially_parsed: "null".into(),
                            invalid: "null".into(),
                        });
                    }
                    FrontendFromRequestError::NoClusterFound(_) => {
                        self.set_answer(DefaultAnswer::Answer404 {});
                    }
                }
                return Err(RetrieveClusterError::RetrieveFrontend(frontend_error));
            }
        };

        // `RouteResult` already carries the full routing decision, but the
        // mux integration that consumes every field is not in place yet.
        // Until it lands, this legacy kawa_h1 path collapses `RouteResult`
        // down to its `cluster_id` and treats an absent cluster as the
        // historical `Route::Deny`.
        let cluster_id = match route.cluster_id {
            Some(cluster_id) => cluster_id,
            None => {
                self.set_answer(DefaultAnswer::Answer401 {
                    www_authenticate: None,
                });
                return Err(RetrieveClusterError::UnauthorizedRoute);
            }
        };

        let frontend_should_redirect_https = matches!(proxy.borrow().kind(), ListenerType::Http)
            && proxy
                .borrow()
                .clusters()
                .get(&cluster_id)
                .map(|cluster| cluster.https_redirect)
                .unwrap_or(false);

        if frontend_should_redirect_https {
            self.set_answer(DefaultAnswer::Answer301 {
                location: format!("https://{host}{uri}"),
            });
            return Err(RetrieveClusterError::UnauthorizedRoute);
        }

        Ok(cluster_id)
    }

    pub fn backend_from_request(
        &mut self,
        cluster_id: &str,
        frontend_should_stick: bool,
        proxy: Rc<RefCell<dyn L7Proxy>>,
        metrics: &mut SessionMetrics,
    ) -> Result<TcpStream, BackendConnectionError> {
        let (backend, conn) = self
            .get_backend_for_sticky_session(
                frontend_should_stick,
                self.context.sticky_session_found.as_deref(),
                cluster_id,
                proxy,
            )
            .map_err(|backend_error| {
                // some backend errors are actually retryable
                // TODO: maybe retry or return a different default answer
                self.set_answer(DefaultAnswer::Answer503 {
                    message: backend_error.to_string(),
                });
                BackendConnectionError::Backend(backend_error)
            })?;

        if frontend_should_stick {
            // update sticky name in case it changed I guess?
            self.context.sticky_name = self.listener.borrow().get_sticky_name().to_string();

            self.context.sticky_session = Some(
                backend
                    .borrow()
                    .sticky_id
                    .clone()
                    .unwrap_or_else(|| backend.borrow().backend_id.clone()),
            );
        }

        metrics.backend_id = Some(backend.borrow().backend_id.clone());
        metrics.backend_start();
        self.set_backend_id(backend.borrow().backend_id.clone());

        self.backend = Some(backend);
        Ok(conn)
    }

    fn get_backend_for_sticky_session(
        &self,
        frontend_should_stick: bool,
        sticky_session: Option<&str>,
        cluster_id: &str,
        proxy: Rc<RefCell<dyn L7Proxy>>,
    ) -> Result<(Rc<RefCell<Backend>>, TcpStream), BackendError> {
        match (frontend_should_stick, sticky_session) {
            (true, Some(sticky_session)) => proxy
                .borrow()
                .backends()
                .borrow_mut()
                .backend_from_sticky_session(cluster_id, sticky_session),
            _ => proxy
                .borrow()
                .backends()
                .borrow_mut()
                .backend_from_cluster_id(cluster_id),
        }
    }

    fn connect_to_backend(
        &mut self,
        session_rc: Rc<RefCell<dyn ProxySession>>,
        proxy: Rc<RefCell<dyn L7Proxy>>,
        metrics: &mut SessionMetrics,
    ) -> Result<BackendConnectAction, BackendConnectionError> {
        let old_cluster_id = self.context.cluster_id.clone();
        let old_backend_token = self.backend_token;

        self.check_circuit_breaker()?;

        let cluster_id = self
            .cluster_id_from_request(proxy.clone())
            .map_err(BackendConnectionError::RetrieveClusterError)?;

        trace!(
            "{} Connect_to_backend: {:?} {:?} {:?}",
            log_context!(self),
            self.context.cluster_id,
            cluster_id,
            self.backend_connection_status
        );
        // check if we can reuse the backend connection
        if (self.context.cluster_id.as_ref()) == Some(&cluster_id)
            && self.backend_connection_status == BackendConnectionStatus::Connected
        {
            let has_backend = self
                .backend
                .as_ref()
                .map(|backend| {
                    let backend = backend.borrow();
                    proxy
                        .borrow()
                        .backends()
                        .borrow()
                        .has_backend(&cluster_id, &backend)
                })
                .unwrap_or(false);

            if has_backend && self.check_backend_connection(metrics) {
                return Ok(BackendConnectAction::Reuse);
            } else if self.backend_token.take().is_some() {
                self.close_backend(proxy.clone(), metrics);
            }
        }

        //replacing with a connection to another cluster
        if old_cluster_id.is_some()
            && old_cluster_id.as_ref() != Some(&cluster_id)
            && self.backend_token.take().is_some()
        {
            self.close_backend(proxy.clone(), metrics);
        }

        self.context.cluster_id = Some(cluster_id.clone());

        let frontend_should_stick = proxy
            .borrow()
            .clusters()
            .get(&cluster_id)
            .map(|cluster| cluster.sticky_session)
            .unwrap_or(false);

        let mut socket =
            self.backend_from_request(&cluster_id, frontend_should_stick, proxy.clone(), metrics)?;
        if let Err(e) = socket.set_nodelay(true) {
            error!(
                "{} Error setting nodelay on backend socket({:?}): {:?}",
                log_context!(self),
                socket,
                e
            );
        }

        self.backend_readiness.interest = Ready::WRITABLE | Ready::HUP | Ready::ERROR;
        self.backend_connection_status = BackendConnectionStatus::Connecting(Instant::now());

        match old_backend_token {
            Some(backend_token) => {
                self.set_backend_token(backend_token);
                if let Err(e) = proxy.borrow().register_socket(
                    &mut socket,
                    backend_token,
                    Interest::READABLE | Interest::WRITABLE,
                ) {
                    error!(
                        "{} Error registering back socket({:?}): {:?}",
                        log_context!(self),
                        socket,
                        e
                    );
                }

                self.set_backend_socket(socket, self.backend.clone());
                self.set_backend_timeout(self.configured_connect_timeout);

                Ok(BackendConnectAction::Replace)
            }
            None => {
                let backend_token = proxy.borrow().add_session(session_rc);

                if let Err(e) = proxy.borrow().register_socket(
                    &mut socket,
                    backend_token,
                    Interest::READABLE | Interest::WRITABLE,
                ) {
                    error!(
                        "{} Error registering back socket({:?}): {:?}",
                        log_context!(self),
                        socket,
                        e
                    );
                }

                self.set_backend_socket(socket, self.backend.clone());
                self.set_backend_token(backend_token);
                self.set_backend_timeout(self.configured_connect_timeout);

                Ok(BackendConnectAction::New)
            }
        }
    }

    fn set_backend_connected(
        &mut self,
        connected: BackendConnectionStatus,
        metrics: &mut SessionMetrics,
    ) {
        let last = self.backend_connection_status;
        self.backend_connection_status = connected;

        if connected == BackendConnectionStatus::Connected {
            // Gauge ±1 pairing: the `CONNECTIONS` gauge is incremented here and
            // decremented exactly once in `close_backend`. Transitioning into
            // `Connected` from an already-`Connected` state would double-count
            // and leave the gauge permanently inflated (it never underflows,
            // but it would never return to zero either). The only legitimate
            // predecessors are `NotConnected` and `Connecting(_)`.
            debug_assert_ne!(
                last,
                BackendConnectionStatus::Connected,
                "set_backend_connected(Connected) must not run twice without a close in between"
            );
            gauge_add!(names::backend::CONNECTIONS, 1);
            gauge_add!(
                names::backend::CONNECTIONS_PER_BACKEND,
                1,
                self.context.cluster_id.as_deref(),
                metrics.backend_id.as_deref()
            );

            // the back timeout was of connect_timeout duration before,
            // now that we're connected, move to backend_timeout duration
            self.set_backend_timeout(self.configured_backend_timeout);
            // if we are not waiting for the backend response, its timeout is concelled
            // it should be set when the request has been entirely transmitted
            if !self.backend_readiness.interest.is_readable() {
                self.container_backend_timeout.cancel();
            }

            if let Some(backend) = &self.backend {
                let mut backend = backend.borrow_mut();

                if backend.retry_policy.is_down() {
                    incr!(
                        "backend.up",
                        self.context.cluster_id.as_deref(),
                        metrics.backend_id.as_deref()
                    );
                    gauge!(
                        names::backend::AVAILABLE,
                        1,
                        self.context.cluster_id.as_deref(),
                        metrics.backend_id.as_deref()
                    );

                    info!(
                        "{} backend server {} at {} is up",
                        log_context!(self),
                        backend.backend_id,
                        backend.address
                    );

                    push_event(Event {
                        kind: EventKind::BackendUp as i32,
                        backend_id: Some(backend.backend_id.to_owned()),
                        address: Some(backend.address.into()),
                        cluster_id: None,
                        metric_detail: None,
                    });
                }

                if let BackendConnectionStatus::Connecting(start) = last {
                    backend.set_connection_time(Instant::now() - start);
                }

                //successful connection, reset failure counter
                backend.failures = 0;
                backend.active_requests += 1;
                backend.retry_policy.succeed();
            }
        }
    }

    fn fail_backend_connection(&mut self, metrics: &SessionMetrics) {
        if let Some(backend) = &self.backend {
            let mut backend = backend.borrow_mut();
            backend.failures += 1;

            let already_unavailable = backend.retry_policy.is_down();
            backend.retry_policy.fail();
            incr!(
                "backend.connections.error",
                self.context.cluster_id.as_deref(),
                metrics.backend_id.as_deref()
            );

            if !already_unavailable && backend.retry_policy.is_down() {
                error!(
                    "{} backend server {} at {} is down",
                    log_context!(self),
                    backend.backend_id,
                    backend.address
                );

                incr!(
                    "backend.down",
                    self.context.cluster_id.as_deref(),
                    metrics.backend_id.as_deref()
                );
                gauge!(
                    names::backend::AVAILABLE,
                    0,
                    self.context.cluster_id.as_deref(),
                    metrics.backend_id.as_deref()
                );

                push_event(Event {
                    kind: EventKind::BackendDown as i32,
                    backend_id: Some(backend.backend_id.to_owned()),
                    address: Some(backend.address.into()),
                    cluster_id: None,
                    metric_detail: None,
                });
            }
        }
    }

    pub fn backend_hup(&mut self, _metrics: &mut SessionMetrics) -> StateResult {
        let response_stream = match &mut self.response_stream {
            ResponseStream::BackendAnswer(response_stream) => response_stream,
            _ => return StateResult::CloseBackend,
        };

        // there might still data we can read on the socket
        if self.backend_readiness.event.is_readable()
            && self.backend_readiness.interest.is_readable()
        {
            return StateResult::Continue;
        }

        // the backend finished to answer we can close
        if response_stream.is_terminated() {
            return StateResult::CloseBackend;
        }
        // Reaching the match means the response is neither fully readable nor
        // already terminated — the early returns above peeled those cases off.
        // The `(_, false)` arm relies on this when it force-terminates a
        // length-less response: terminating an already-terminated stream would
        // be a no-op masking a real double-close.
        debug_assert!(
            !response_stream.is_terminated(),
            "backend_hup reached its decision match with an already-terminated response"
        );
        match (
            self.request_stream.is_initial(),
            response_stream.is_initial(),
        ) {
            // backend stopped before response is finished,
            // or maybe it was malformed in the first place (no Content-Length)
            (_, false) => {
                error!(
                    "{} Backend closed before session is over",
                    log_context!(self, Some(response_stream.parsing_phase)),
                );

                trace!(
                    "{} Backend hang-up, setting the parsing phase of the response stream to terminated, this also takes care of responses that lack length information.",
                    log_context!(self, Some(response_stream.parsing_phase))
                );

                response_stream.parsing_phase = kawa::ParsingPhase::Terminated;

                // Post: the forced termination took — required for the
                // length-less / truncated-response close path to make progress.
                debug_assert!(
                    response_stream.is_terminated(),
                    "forced backend-hup termination must mark the response terminated"
                );

                // writable() will be called again and finish the session properly
                // for this reason, writable must not short cut
                self.frontend_readiness.interest.insert(Ready::WRITABLE);
                // Post: the frontend is now armed to write so `writable` can
                // flush whatever the backend already sent and then close.
                debug_assert!(
                    self.frontend_readiness.interest.is_writable(),
                    "backend-hup termination must arm the frontend for the final flush"
                );
                StateResult::Continue
            }
            // probably backend hup between keep alive request, change backend
            (true, true) => {
                trace!(
                    "{} Backend hanged up in between requests",
                    log_context!(self)
                );
                StateResult::CloseBackend
            }
            // the frontend already transmitted data so we can't redirect
            (false, true) => {
                error!(
                    "{}  Frontend transmitted data but the back closed",
                    log_context!(self)
                );

                self.set_answer(DefaultAnswer::Answer503 {
                    message: "Backend closed after consuming part of the request".into(),
                });

                self.backend_readiness.interest = Ready::EMPTY;
                StateResult::Continue
            }
        }
    }

    /// The main session loop, processing all events triggered by mio since last time
    /// and proxying http traffic. The main flow can be summed up by:
    ///
    /// - if connecting an back has event:
    ///   - if backend hanged up, try again
    ///   - else, set as connected
    /// - while front or back has event:
    ///   - read request on front
    ///   - write request to back
    ///   - read response on back
    ///   - write response to front
    fn ready_inner(
        &mut self,
        session: Rc<RefCell<dyn crate::ProxySession>>,
        proxy: Rc<RefCell<dyn L7Proxy>>,
        metrics: &mut SessionMetrics,
    ) -> SessionResult {
        let mut counter = 0;

        if self.backend_connection_status.is_connecting()
            && !self.backend_readiness.event.is_empty()
        {
            if self.backend_readiness.event.is_hup() && !self.test_backend_socket() {
                //retry connecting the backend
                warn!(
                    "{} Error connecting to backend, trying again, attempt {}",
                    log_context!(self),
                    self.connection_attempts
                );

                // Each backend HUP-retry bumps the attempt counter by exactly
                // one; the very next `connect_to_backend` runs the circuit
                // breaker, which converts the `== CONN_RETRIES` case into a 503
                // instead of dialling again — so the counter is never observed
                // above `CONN_RETRIES` once the loop settles (see
                // `check_invariants`).
                let attempts_before = self.connection_attempts;
                self.connection_attempts += 1;
                debug_assert_eq!(
                    self.connection_attempts,
                    attempts_before + 1,
                    "a backend retry must bump connection_attempts by exactly one"
                );
                debug_assert!(
                    self.connection_attempts <= CONN_RETRIES,
                    "connection_attempts ({}) must not exceed the breaker budget on retry",
                    self.connection_attempts
                );
                self.fail_backend_connection(metrics);

                self.backend_connection_status =
                    BackendConnectionStatus::Connecting(Instant::now());

                // trigger a backend reconnection
                self.close_backend(proxy.clone(), metrics);

                let connection_result =
                    self.connect_to_backend(session.clone(), proxy.clone(), metrics);
                if let Err(err) = &connection_result {
                    match err {
                        // Already logged at warn! + metered at check_circuit_breaker;
                        // avoid double-emission.
                        BackendConnectionError::MaxConnectionRetries(_) => trace!(
                            "{} Error connecting to backend: {}",
                            log_context!(self),
                            err
                        ),
                        _ => warn!(
                            "{} Error connecting to backend: {}",
                            log_context!(self),
                            err
                        ),
                    }
                }

                if let Some(session_result) = handle_connection_result(connection_result) {
                    return session_result;
                }
            } else {
                metrics.backend_connected();
                self.connection_attempts = 0;
                self.set_backend_connected(BackendConnectionStatus::Connected, metrics);
                // we might get an early response from the backend, so we want to look
                // at readable events
                self.backend_readiness.interest.insert(Ready::READABLE);
            }
        }

        if self.frontend_readiness.event.is_hup() {
            if !self.request_stream.is_initial() {
                self.log_request_error(metrics, "Client disconnected abruptly");
            }
            return SessionResult::Close;
        }

        while counter < MAX_LOOP_ITERATIONS {
            let frontend_interest = self.frontend_readiness.filter_interest();
            let backend_interest = self.backend_readiness.filter_interest();

            trace!(
                "{} Frontend interest({:?}) and backend interest({:?})",
                log_context!(self),
                frontend_interest,
                backend_interest,
            );

            if frontend_interest.is_empty() && backend_interest.is_empty() {
                break;
            }

            if self.backend_readiness.event.is_hup()
                && self.frontend_readiness.interest.is_writable()
                && !self.frontend_readiness.event.is_writable()
            {
                break;
            }

            if frontend_interest.is_readable() {
                let state_result = self.readable(metrics);
                trace!(
                    "{} frontend_readable: {:?}",
                    log_context!(self),
                    state_result
                );

                match state_result {
                    StateResult::Continue => {}
                    StateResult::ConnectBackend => {
                        let connection_result =
                            self.connect_to_backend(session.clone(), proxy.clone(), metrics);
                        if let Err(err) = &connection_result {
                            match err {
                                // Already logged at warn! + metered at check_circuit_breaker;
                                // avoid double-emission.
                                BackendConnectionError::MaxConnectionRetries(_) => trace!(
                                    "{} Error connecting to backend: {}",
                                    log_context!(self),
                                    err
                                ),
                                _ => warn!(
                                    "{} Error connecting to backend: {}",
                                    log_context!(self),
                                    err
                                ),
                            }
                        }

                        if let Some(session_result) = handle_connection_result(connection_result) {
                            return session_result;
                        }
                    }
                    StateResult::CloseBackend => unreachable!(),
                    StateResult::CloseSession => return SessionResult::Close,
                    StateResult::Upgrade => return SessionResult::Upgrade,
                }
            }

            if backend_interest.is_writable() {
                let session_result = self.backend_writable(metrics);
                trace!(
                    "{} backend_writable: {:?}",
                    log_context!(self),
                    session_result
                );
                if session_result != SessionResult::Continue {
                    return session_result;
                }
            }

            if backend_interest.is_readable() {
                let session_result = self.backend_readable(metrics);
                trace!(
                    "{} backend_readable: {:?}",
                    log_context!(self),
                    session_result
                );
                if session_result != SessionResult::Continue {
                    return session_result;
                }
            }

            if frontend_interest.is_writable() {
                let state_result = self.writable(metrics);
                trace!(
                    "{} frontend_writable: {:?}",
                    log_context!(self),
                    state_result
                );
                match state_result {
                    StateResult::CloseBackend => self.close_backend(proxy.clone(), metrics),
                    StateResult::CloseSession => return SessionResult::Close,
                    StateResult::Upgrade => return SessionResult::Upgrade,
                    StateResult::Continue => {}
                    StateResult::ConnectBackend => unreachable!(),
                }
            }

            if frontend_interest.is_error() {
                error!(
                    "{} frontend socket error, disconnecting",
                    log_context!(self)
                );

                return SessionResult::Close;
            }

            if backend_interest.is_hup() || backend_interest.is_error() {
                let state_result = self.backend_hup(metrics);

                trace!("{} backend_hup: {:?}", log_context!(self), state_result);
                match state_result {
                    StateResult::Continue => {}
                    StateResult::CloseBackend => self.close_backend(proxy.clone(), metrics),
                    StateResult::CloseSession => return SessionResult::Close,
                    StateResult::ConnectBackend | StateResult::Upgrade => unreachable!(),
                }
            }

            counter += 1;
        }

        if counter >= MAX_LOOP_ITERATIONS {
            error!(
                "{}\tHandling session went through {} iterations, there's a probable infinite loop bug, closing the connection",
                log_context!(self),
                MAX_LOOP_ITERATIONS
            );

            incr!(names::http::INFINITE_LOOP_ERROR);
            self.print_state(self.protocol_string());

            return SessionResult::Close;
        }

        // Post: a `Continue` exit means the loop drained all pending interest
        // within the iteration budget (the `>= MAX_LOOP_ITERATIONS` arm above
        // returns `Close` otherwise). This pins the no-infinite-loop contract.
        debug_assert!(
            counter < MAX_LOOP_ITERATIONS,
            "ready_inner must only Continue when the loop settled within the iteration budget"
        );
        SessionResult::Continue
    }

    pub fn timeout_status(&self) -> TimeoutStatus {
        if self.request_stream.is_main_phase() {
            match &self.response_stream {
                ResponseStream::BackendAnswer(kawa) if kawa.is_initial() => {
                    TimeoutStatus::WaitingForResponse
                }
                _ => TimeoutStatus::Response,
            }
        } else if self.keepalive_count > 0 {
            TimeoutStatus::WaitingForNewRequest
        } else {
            TimeoutStatus::Request
        }
    }

    /// Cross-field invariant sweep for the H1 session state machine.
    ///
    /// Run-to-completion postcondition: called via a `#[cfg(debug_assertions)]`
    /// guard at the end of the [`SessionState::ready`] entry point, after the
    /// inner loop has settled the session into a stable, between-events state.
    /// It encodes relationships that must hold for ANY `Http` regardless of the
    /// path (request, response, default answer, backend retry) that drove the
    /// loop:
    ///
    /// - the circuit-breaker counter never exceeds the configured retry budget
    ///   (`check_circuit_breaker` short-circuits at `CONN_RETRIES`, so a higher
    ///   value would mean a retry slipped past the breaker);
    /// - a `Connected` backend always names a live socket — `close_backend`
    ///   takes the socket and drops back to `NotConnected` atomically, so any
    ///   `Connected`-without-socket state is a teardown-ordering bug that would
    ///   make `backend_writable`/`backend_readable` log "back socket not found";
    /// - while a default answer is queued the frontend is steered toward
    ///   WRITABLE and never asked to read more request bytes (the read path
    ///   `error!`s and bails if it sees a `DefaultAnswer`), and the backend is
    ///   never asked to read or write (we have nothing left to proxy);
    /// - `context.status` agrees with the queued default-answer status, so the
    ///   access-log / `save_http_status_metric` code never reports a different
    ///   code than the bytes on the wire.
    ///
    /// Network input never reaches a hard `assert!` here — invalid traffic is
    /// turned into a default answer or a close upstream; these `debug_assert!`s
    /// fire only on our own logic bugs and compile out in release.
    #[cfg(debug_assertions)]
    fn check_invariants(&self) {
        debug_assert!(
            self.connection_attempts <= CONN_RETRIES,
            "connection_attempts ({}) must stay within the circuit-breaker budget ({CONN_RETRIES})",
            self.connection_attempts
        );
        if self.backend_connection_status == BackendConnectionStatus::Connected {
            debug_assert!(
                self.backend_socket.is_some(),
                "a Connected backend must own a live socket (teardown-ordering bug)"
            );
        }
        if let ResponseStream::DefaultAnswer(status, _) = &self.response_stream {
            debug_assert!(
                !self.frontend_readiness.interest.is_readable(),
                "frontend must not be asked to read more request bytes while a default answer is queued"
            );
            debug_assert!(
                !self.backend_readiness.interest.is_readable()
                    && !self.backend_readiness.interest.is_writable(),
                "backend must be idle (no read/write interest) while a default answer is queued"
            );
            debug_assert_eq!(
                self.context.status,
                Some(*status),
                "context.status must match the queued default-answer status"
            );
        }
    }
}

impl<Front: SocketHandler, L: ListenerHandler + L7ListenerHandler> SessionState for Http<Front, L> {
    fn ready(
        &mut self,
        session: Rc<RefCell<dyn crate::ProxySession>>,
        proxy: Rc<RefCell<dyn L7Proxy>>,
        metrics: &mut SessionMetrics,
    ) -> SessionResult {
        let session_result = self.ready_inner(session, proxy, metrics);
        if session_result == SessionResult::Upgrade {
            let response_storage = match &mut self.response_stream {
                ResponseStream::BackendAnswer(response_stream) => &mut response_stream.storage,
                _ => return SessionResult::Close,
            };

            // sync the underlying Checkout buffers, if they contain remaining data
            // it will be processed once upgraded to websocket
            self.request_stream.storage.buffer.sync(
                self.request_stream.storage.end,
                self.request_stream.storage.head,
            );
            response_storage
                .buffer
                .sync(response_storage.end, response_storage.head);
        }
        // Run-to-completion postcondition: the session has settled into a
        // stable between-events state. Compiled out in release.
        #[cfg(debug_assertions)]
        self.check_invariants();
        session_result
    }

    fn update_readiness(&mut self, token: Token, events: Ready) {
        if self.frontend_token == token {
            self.frontend_readiness.event |= events;
        } else if self.backend_token == Some(token) {
            self.backend_readiness.event |= events;
        }
    }

    fn close(&mut self, proxy: Rc<RefCell<dyn L7Proxy>>, metrics: &mut SessionMetrics) {
        self.close_backend(proxy, metrics);
        self.frontend_socket.socket_close();
        let _ = self.frontend_socket.socket_write_vectored(&[]);

        //if the state was initial, the connection was already reset
        if !self.request_stream.is_initial() {
            gauge_add!(names::http::ACTIVE_REQUESTS, -1);

            if let Some(b) = self.backend.as_mut() {
                let mut backend = b.borrow_mut();
                backend.active_requests = backend.active_requests.saturating_sub(1);
            }
        }
    }

    fn timeout(&mut self, token: Token, metrics: &mut SessionMetrics) -> StateResult {
        //info!("got timeout for token: {:?}", token);
        if self.frontend_token == token {
            self.container_frontend_timeout.triggered();
            return match self.timeout_status() {
                // we do not have a complete answer
                TimeoutStatus::Request => {
                    self.context.access_log_message = Some("client_timeout");
                    self.set_answer(DefaultAnswer::Answer408 {
                        duration: self.container_frontend_timeout.to_string(),
                    });
                    self.writable(metrics)
                }
                // we have a complete answer but the response did not start
                TimeoutStatus::WaitingForResponse => {
                    // this case is ambiguous, as it is the frontend timeout that triggers while we were waiting for response
                    // the timeout responsibility should have switched before
                    self.context.access_log_message = Some("client_timeout_during_response");
                    self.set_answer(DefaultAnswer::Answer504 {
                        duration: self.container_backend_timeout.to_string(),
                    });
                    self.writable(metrics)
                }
                // we have a complete answer and the start of a response, but the request was not tagged as terminated
                // for now we place responsibility of timeout on the backend in those cases, so we ignore this
                TimeoutStatus::Response => StateResult::Continue,
                // timeout in keep-alive, simply close the connection
                TimeoutStatus::WaitingForNewRequest => StateResult::CloseSession,
            };
        }

        if self.backend_token == Some(token) {
            //info!("backend timeout triggered for token {:?}", token);
            self.container_backend_timeout.triggered();
            return match self.timeout_status() {
                TimeoutStatus::Request => {
                    error!(
                        "{} got backend timeout while waiting for a request, this should not happen",
                        log_context!(self)
                    );
                    // Operator-visible cause is the same as the regular
                    // backend-timeout arm — backend silent before we
                    // could send the request. The error! above keeps the
                    // internal-invariant signal for sozu maintainers.
                    self.context.access_log_message = Some("backend_timeout");
                    self.set_answer(DefaultAnswer::Answer504 {
                        duration: self.container_backend_timeout.to_string(),
                    });
                    self.writable(metrics)
                }
                TimeoutStatus::WaitingForResponse => {
                    self.context.access_log_message = Some("backend_timeout");
                    self.set_answer(DefaultAnswer::Answer504 {
                        duration: self.container_backend_timeout.to_string(),
                    });
                    self.writable(metrics)
                }
                TimeoutStatus::Response => {
                    error!(
                        "backend {:?} timeout while receiving response (cluster {:?})",
                        self.context.backend_id, self.context.cluster_id
                    );
                    self.context.access_log_message = Some("backend_response_timeout");
                    StateResult::CloseSession
                }
                // in keep-alive, we place responsibility of timeout on the frontend, so we ignore this
                TimeoutStatus::WaitingForNewRequest => StateResult::Continue,
            };
        }

        error!("{} Got timeout for an invalid token", log_context!(self));
        StateResult::CloseSession
    }

    fn cancel_timeouts(&mut self) {
        self.container_backend_timeout.cancel();
        self.container_frontend_timeout.cancel();
    }

    fn print_state(&self, context: &str) {
        error!(
            "\
{} {} Session(Kawa)
\tFrontend:
\t\ttoken: {:?}\treadiness: {:?}\tstate: {:?}
\tBackend:
\t\ttoken: {:?}\treadiness: {:?}",
            log_context!(self),
            context,
            self.frontend_token,
            self.frontend_readiness,
            self.request_stream.parsing_phase,
            self.backend_token,
            self.backend_readiness,
            // self.response_stream.parsing_phase
        );
    }

    fn shutting_down(&mut self) -> SessionIsToBeClosed {
        if self.request_stream.is_initial() && self.request_stream.storage.is_empty()
        // && self.response_stream.storage.is_empty()
        {
            true
        } else {
            self.context.closing = true;
            false
        }
    }
}

fn handle_connection_result(
    connection_result: Result<BackendConnectAction, BackendConnectionError>,
) -> Option<SessionResult> {
    match connection_result {
        // reuse connection or send a default answer, we can continue
        Ok(BackendConnectAction::Reuse) => None,
        Ok(BackendConnectAction::New) | Ok(BackendConnectAction::Replace) => {
            // we must wait for an event
            Some(SessionResult::Continue)
        }
        Err(_) => {
            // All BackendConnectionError already set a default answer
            // the session must continue to serve it
            // - NotFound: not used for http (only tcp)
            // - RetrieveClusterError: 301/400/401/404,
            // - MaxConnectionRetries: 503,
            // - Backend: 503,
            // - MaxSessionsMemory: not checked in connect_to_backend (TODO: check it?)
            None
        }
    }
}

/// Save the HTTP status code of the backend response.
///
/// Emits two counters per response with a known status:
///
/// 1. The bucket counter `http.status.{1xx,…,5xx,other}` (always),
/// 2. A per-code counter `http.status.{200,301,…}` when the code is on the
///    short-list maintained by [`crate::metrics::http_status_code_metric_name`].
///
/// Status-less responses still emit `http.status.none` upstream of this
/// function (see `generate_access_log` for the H2 path); the H1 entry point
/// only carries `Some(status)` so the `else` branch is unnecessary here.
fn save_http_status_metric(status: Option<u16>, context: LogContext) {
    if let Some(status) = status {
        // Every status reaching this bucketer originates either from a parsed
        // backend response status line or a validated answer template, both of
        // which are 3-digit HTTP codes. A value outside `100..=999` would fall
        // into the `STATUS_OTHER` catch-all below and signal a status that was
        // never a real HTTP code — a logic bug upstream, not hostile traffic.
        debug_assert!(
            (100..=999).contains(&status),
            "save_http_status_metric got a non-3-digit status: {status}"
        );
        match status {
            100..=199 => {
                incr!(
                    names::http::STATUS_1XX,
                    context.cluster_id,
                    context.backend_id
                );
            }
            200..=299 => {
                incr!(
                    names::http::STATUS_2XX,
                    context.cluster_id,
                    context.backend_id
                );
            }
            300..=399 => {
                incr!(
                    names::http::STATUS_3XX,
                    context.cluster_id,
                    context.backend_id
                );
            }
            400..=499 => {
                incr!(
                    names::http::STATUS_4XX,
                    context.cluster_id,
                    context.backend_id
                );
            }
            500..=599 => {
                incr!(
                    names::http::STATUS_5XX,
                    context.cluster_id,
                    context.backend_id
                );
            }
            _ => {
                // http responses with other codes (protocol error)
                incr!(names::http::STATUS_OTHER);
            }
        }

        if let Some(per_code) = crate::metrics::http_status_code_metric_name(status) {
            incr!(per_code, context.cluster_id, context.backend_id);
        }
    }
}