ibapi 5.5.0

//! This module implements a message bus for handling communications with TWS.
//! It provides functionality for routing requests from the Client to TWS,
//! and responses from TWS back to the Client.

use std::collections::{hash_map, HashMap, HashSet};
use std::io::prelude::*;
use std::net::TcpStream;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, MutexGuard, OnceLock, RwLock, RwLockReadGuard, RwLockWriteGuard};
use std::thread::{self, JoinHandle};
use std::time::Duration;

use crossbeam::channel::{self, Receiver, Sender};
use log::{debug, error, info, trace, warn};

use crate::accounts::types::AccountId;
use crate::client::handshake_state::HandshakeState;
use crate::client::ids::{OrderId, RequestId, WireId};
use crate::connection::sync::Connection;

use super::common::{lock, log_orphan, read_lock, report_unroutable_frame, validate_frame_length, write_lock, Lease, LeaseRef};
use super::order_taps::{NewTap, OrderTaps};
use super::rate_limiter::RateLimiter;
use super::raw_capture::RawFrameTap;
use super::routing::{
    classify_error, determine_routing, is_foreign, order_routing_strategy, order_update_notice, DecodedError, ErrorDisposition, OrderRoutingStrategy,
    RoutingDecision,
};
use super::{
    Admit, BoundState, BufferBound, InternalSubscription, MessageBus, Response, RoutedItem, SharedCounts, SharedTicket, Signal, SubscriptionBuilder,
};
use crate::messages::{
    is_cancel_frame, shared_channel_configuration, transport_reconnect_notice, IncomingMessages, Notice, OutgoingMessages, ResponseMessage,
};
use crate::subscriptions::notice_stream::sync_impl::NoticeStream;
use crate::Error;

const TWS_READ_TIMEOUT: Duration = Duration::from_secs(1);

/// How long a frame may stall after its first byte before the read gives up
/// and the dispatcher reconnects. A healthy gateway never pauses this long
/// inside a frame.
const MID_FRAME_STALL: Duration = Duration::from_secs(30);

/// [`MID_FRAME_STALL`] in read timeouts: [`FrameRead`] counts timeouts rather
/// than reading a clock.
const MID_FRAME_TIMEOUT_LIMIT: u32 = (MID_FRAME_STALL.as_millis() / TWS_READ_TIMEOUT.as_millis()) as u32;
const _: () = assert!(MID_FRAME_TIMEOUT_LIMIT > 0, "MID_FRAME_STALL must exceed TWS_READ_TIMEOUT");

/// How long connect waits for each startup frame (handshake ack, account
/// info) before giving up. A loaded gateway can take longer than one read
/// timeout to answer.
const STARTUP_STALL: Duration = Duration::from_secs(30);

/// [`STARTUP_STALL`] in read timeouts, counted like [`MID_FRAME_TIMEOUT_LIMIT`].
pub(crate) const STARTUP_TIMEOUT_LIMIT: u32 = (STARTUP_STALL.as_millis() / TWS_READ_TIMEOUT.as_millis()) as u32;
const _: () = assert!(STARTUP_TIMEOUT_LIMIT > 0, "STARTUP_STALL must exceed TWS_READ_TIMEOUT");

/// Queue depth at which (and at every further multiple of which) a growing
/// sync channel logs a warning. Sync channels are unbounded — they never drop,
/// so the failure mode of a stalled consumer is silent memory growth. The
/// watermark makes it loud without changing the lossless semantics; see
/// #779 and #896.
const BACKLOG_WATERMARK: usize = 10_000;

/// `true` exactly when `depth` sits on a watermark multiple — one warning per
/// 10k messages of backlog, not one per message. The dispatcher is the sole
/// producer and reads the depth after each single-message send, so an upward
/// crossing cannot skip past the multiple.
fn backlog_watermark_crossed(depth: usize) -> bool {
    depth > 0 && depth.is_multiple_of(BACKLOG_WATERMARK)
}

/// Warn when a queue's depth crosses a watermark; `label` names the queue.
pub(super) fn warn_if_backlogged(label: std::fmt::Arguments<'_>, depth: usize) {
    if backlog_watermark_crossed(depth) {
        warn!("{label} at {depth} messages and growing — consumer is stalling");
    }
}

// One live subscription to a shared (id-less) request. Registered under
// every response type of its request mapping; `sender` feeds its own queue.
#[derive(Debug)]
struct SharedSubscriber {
    request: OutgoingMessages,
    responses: &'static [IncomingMessages],
    sender: Sender<RoutedItem>,
    lease: LeaseRef,
}

impl SharedSubscriber {
    fn receives(&self, message_type: IncomingMessages) -> bool {
        self.responses.contains(&message_type)
    }
}

// For requests without an identifier, responses are routed by message type
// to every live subscription of a request mapped to that type. Each
// subscription owns its own queue (like the async broadcast `resubscribe`),
// so every subscriber sees every frame, nothing dispatched before it
// subscribed reaches it, and nothing is queued for a subscriber that no
// longer exists.
#[derive(Debug)]
struct SharedChannels {
    // Every response type of `CHANNEL_MAPPINGS`; a frame of one of these
    // types is a shared response whether or not anybody is subscribed.
    response_types: HashSet<IncomingMessages>,
    // Live subscriptions, added in `send_shared_request` and removed by the
    // cleanup thread on cancel or drop or, lazily, when a send finds the
    // queue gone.
    subscribers: Mutex<Vec<SharedSubscriber>>,
    // Live subscriptions per request type; see `SharedCounts`.
    counts: Mutex<SharedCounts>,
}

impl SharedChannels {
    pub fn new() -> Self {
        Self {
            response_types: shared_channel_configuration::CHANNEL_MAPPINGS
                .iter()
                .flat_map(|mapping| mapping.responses.iter().copied())
                .collect(),
            subscribers: Mutex::new(Vec::new()),
            counts: Mutex::new(SharedCounts::default()),
        }
    }

    fn subscribers(&self) -> MutexGuard<'_, Vec<SharedSubscriber>> {
        lock(&self.subscribers)
    }

    // Registers `sender` for every response type of `request`. Panics if
    // `request` has no mapping.
    fn add(&self, request: OutgoingMessages, sender: Sender<RoutedItem>, lease: LeaseRef) {
        let responses = shared_channel_configuration::response_types(request)
            .unwrap_or_else(|| panic!("unsupported request message {request:?}. check mapping in messages::shared_channel_configuration"));
        self.subscribers().push(SharedSubscriber {
            request,
            responses,
            sender,
            lease,
        });
    }

    // Removes the subscription holding `lease`, if still registered.
    fn remove(&self, lease: &LeaseRef) {
        let mut subscribers = self.subscribers();
        let before = subscribers.len();
        subscribers.retain(|subscriber| !subscriber.lease.is(lease));
        debug!("cleanup shared subscription: removed={}", before != subscribers.len());
    }

    fn is_shared_response(&self, message_type: IncomingMessages) -> bool {
        self.response_types.contains(&message_type)
    }

    // Runs `write` for a new subscription of `message_type` and counts it on
    // success. The lock spans the write so the count and the wire agree.
    // `account` is the account-updates account, checked before the write.
    fn subscribe(
        &self,
        message_type: OutgoingMessages,
        account: Option<&AccountId>,
        write: impl FnOnce() -> Result<(), Error>,
    ) -> Result<SharedTicket, Error> {
        let mut counts = lock(&self.counts);
        counts.check_account_updates(account)?;
        write()?;
        Ok(counts.subscribe(message_type, account))
    }

    // Uncounts `ticket`'s subscription; runs `write` (the cancel) only when
    // `SharedCounts::unsubscribe` says so.
    fn unsubscribe(&self, ticket: SharedTicket, write: impl FnOnce() -> Result<(), Error>) -> Result<(), Error> {
        let mut counts = lock(&self.counts);
        if counts.unsubscribe(ticket) {
            write()
        } else {
            Ok(())
        }
    }

    // Every live shared subscription has just been failed: start a new
    // generation so their later drops cannot touch the next session's counts.
    fn reset_counts(&self) {
        lock(&self.counts).reset();
    }

    // Sends `item()` to every subscriber selected by `filter`; returns how
    // many were selected. A send fails only when the subscriber's queue is
    // gone (its handle was dropped and the cleanup signal has not been
    // processed yet); it is removed here.
    fn send_to<F, I>(&self, filter: F, item: I) -> usize
    where
        F: Fn(&SharedSubscriber) -> bool,
        I: Fn() -> RoutedItem,
    {
        let mut selected = 0;
        self.subscribers().retain(|subscriber| {
            if !filter(subscriber) {
                return true;
            }
            selected += 1;
            match subscriber.sender.send(item()) {
                Ok(()) => {
                    warn_if_backlogged(format_args!("shared channel for {:?}", subscriber.request), subscriber.sender.len());
                    true
                }
                Err(_) => {
                    debug!("shared subscription {:?} dropped: removed", subscriber.request);
                    false
                }
            }
        });
        selected
    }

    // Deliver `message` to every live subscription registered for its type;
    // returns whether any was. With none, the frame is dropped: a shared
    // response nobody asked for (e.g. `OpenOrder` from another client, or
    // after the requester's drop).
    fn send_message(&self, message_type: IncomingMessages, message: &ResponseMessage) -> bool {
        let delivered = self.send_to(|subscriber| subscriber.receives(message_type), || message.clone().into()) > 0;
        if !delivered {
            trace!("no shared subscription for {message_type:?}: frame dropped");
        }
        delivered
    }

    // Deliver `message_fn()` once to every live subscription, then remove
    // every registration under the same lock, so one made meanwhile can't be
    // dropped unnotified. A failed handle's queue must not keep filling until
    // it is dropped.
    fn fail_all<F>(&self, message_fn: F)
    where
        F: Fn() -> RoutedItem,
    {
        let mut subscribers = self.subscribers();
        for subscriber in subscribers.iter() {
            let _ = subscriber.sender.send(message_fn());
        }
        subscribers.clear();
    }

    // Fail in-flight one-shot requests fast by delivering an error to the
    // one-shot subscriptions only. Used for request-less errors, which carry
    // no id to correlate. Streaming subscriptions are excluded so an
    // unrelated error can't terminate a live stream.
    fn fail_one_shot_channels<F>(&self, error_fn: F)
    where
        F: Fn() -> RoutedItem,
    {
        let one_shot = shared_channel_configuration::exclusive_one_shot_response_types();
        self.send_to(|subscriber| subscriber.responses.iter().any(|r| one_shot.contains(r)), error_fn);
    }
}

/// Fan-out for unrouted notices. Each subscriber gets its own unbounded
/// crossbeam channel, watermarked like the subscription queues; `broadcast`
/// lazily prunes subscribers whose receivers have been dropped
/// (`Sender::send` returns `Err` once the receiver is gone).
#[derive(Debug)]
pub(crate) struct NoticeBroadcaster {
    /// `None` once closed.
    senders: Mutex<Option<Vec<Sender<Notice>>>>,
}

impl super::common::NoticeSink for NoticeBroadcaster {
    fn deliver(&self, notice: Notice) {
        self.broadcast(notice);
    }
}

impl NoticeBroadcaster {
    pub(crate) fn new() -> Self {
        Self {
            senders: Mutex::new(Some(Vec::new())),
        }
    }

    /// After `close`, the returned receiver is already at end-of-stream,
    /// like the ones `close` ended.
    pub(crate) fn subscribe(&self) -> Receiver<Notice> {
        let (sender, receiver) = channel::unbounded();
        if let Some(senders) = lock(&self.senders).as_mut() {
            senders.push(sender);
        }
        receiver
    }

    pub(crate) fn broadcast(&self, notice: Notice) {
        if let Some(senders) = lock(&self.senders).as_mut() {
            senders.retain(|s| {
                let sent = s.send(notice.clone()).is_ok();
                if sent {
                    warn_if_backlogged(format_args!("notice stream"), s.len());
                }
                sent
            });
        }
    }

    /// Drop all senders so existing receivers see channel-closed, and end
    /// every later subscription on arrival.
    pub(crate) fn close(&self) {
        *lock(&self.senders) = None;
    }
}

#[derive(Debug)]
pub struct TcpMessageBus<S: Stream> {
    connection: Connection<S>,
    handles: Mutex<Vec<JoinHandle<()>>>,
    requests: SenderHash<RequestId, RoutedItem>,
    orders: SenderHash<OrderId, RoutedItem>,
    executions: SenderHash<String, RoutedItem>,
    shared_channels: SharedChannels,
    signals_send: Sender<Signal>,
    signals_recv: Receiver<Signal>,
    shutdown_send: Sender<()>,
    shutdown_recv: Receiver<()>,
    /// Shared with the connection so a reconnect in progress sees the request.
    shutdown: Arc<ShutdownSignal>,
    /// The client state a reconnect's handshake refreshes (order-id floor,
    /// server version). Installed once via [`Self::set_handshake_state`]
    /// before the dispatcher thread starts; absent in bus-only test fixtures,
    /// which never reconnect a client.
    handshake_state: OnceLock<HandshakeState>,
    /// See [`MessageBus::session_generation`]; bumped in `reset`.
    session_generation: AtomicU64,
    /// Outbound rate limiter, if the client was built with one. Installed
    /// once via [`Self::set_rate_limiter`] before any request is sent.
    rate_limiter: OnceLock<RateLimiter>,
    /// The order-update stream and the per-order status streams.
    order_taps: OrderTaps<Sender<RoutedItem>>,
    /// Session state, and what `wait_connected` blocks on.
    connection_state: ConnectionSignal,
}

impl<S: Stream> TcpMessageBus<S> {
    pub fn new(connection: Connection<S>) -> TcpMessageBus<S> {
        let order_taps = OrderTaps::new();
        let (signals_send, signals_recv) = channel::unbounded();
        let (shutdown_send, shutdown_recv) = channel::bounded(1);
        let shutdown = connection.shutdown_signal();

        TcpMessageBus {
            connection,
            handles: Mutex::new(Vec::default()),
            requests: SenderHash::new(),
            orders: SenderHash::new(),
            executions: SenderHash::new(),
            shared_channels: SharedChannels::new(),
            signals_send,
            signals_recv,
            shutdown_send,
            shutdown_recv,
            shutdown,
            handshake_state: OnceLock::new(),
            session_generation: AtomicU64::new(0),
            rate_limiter: OnceLock::new(),
            order_taps,
            connection_state: ConnectionSignal::default(),
        }
    }

    /// Installs the client state a successful reconnect refreshes from its
    /// handshake. Called exactly once, before the dispatcher thread starts.
    pub(crate) fn set_handshake_state(&self, state: HandshakeState) {
        self.handshake_state.set(state).expect("handshake state installed twice");
    }

    /// Installs the outbound rate limiter. Called at most once, before the
    /// client sends anything.
    pub(crate) fn set_rate_limiter(&self, limiter: RateLimiter) {
        self.rate_limiter.set(limiter).expect("rate limiter installed twice");
    }

    fn is_shutting_down(&self) -> bool {
        self.shutdown.is_requested()
    }

    fn request_shutdown(&self) {
        debug!("shutdown requested");

        self.requests.fail_all(|| Error::Shutdown.into());
        self.orders.fail_all(|| Error::Shutdown.into());
        self.shared_channels.fail_all(|| Error::Shutdown.into());
        // Aliases of the routes just failed.
        self.executions.clear();
        self.connection.notice_broadcaster.close();

        // Both latch: the connection signal releases every `wait_connected`
        // with `Error::Shutdown`, and the shutdown signal ends any backoff
        // wait `Connection::reconnect` is in.
        self.connection_state.shutdown();
        self.shutdown.request();

        self.order_taps.close();

        // bounded(1) + try_send: if a shutdown is already pending,
        // Err(Full) is the desired no-op (idempotent across duplicate calls).
        let _ = self.shutdown_send.try_send(());

        // Release the socket, which retained clients and subscriptions would
        // otherwise keep open. Also breaks the dispatcher's blocked read, so it
        // exits without waiting out the 1s socket-read timeout.
        self.connection.socket.close();
    }

    /// The send gate: every bus-originated write checks here first, and is
    /// refused unless the session is live and not shutting down - with
    /// `Error::ConnectionReset` while a reconnect may still bring it back,
    /// `Error::ConnectionFailed` once the reconnect gave up, and
    /// `Error::Shutdown` after a shutdown, so a caller can tell "wait" from
    /// "stop".
    ///
    /// Without it, a request sent while the connection is down would be lost
    /// on the dead socket - or, worse, interleave with the handshake the
    /// reconnect is writing on the new one - and the channel it registered
    /// would never be reset again, so its caller would block forever. Callers
    /// wait the reconnect out through `wait_connected` instead.
    ///
    /// The check-then-write window is not closed: a caller that passes here
    /// just before the dispatcher flips the state can still write, and in
    /// principle race the socket swap inside `reconnect`. Closing that needs a
    /// session-level gate held across the handshake.
    fn ensure_connected(&self) -> Result<(), Error> {
        self.connection_state.check()?;
        if self.is_shutting_down() {
            return Err(Error::Shutdown);
        }
        Ok(())
    }

    /// A refusal from the order taps, which close on shutdown, reported as
    /// why the session ended - `ConnectionFailed` when the client gave up on
    /// it, as the send gate does. Asked after the refusal: `fail` latches
    /// before the taps close, so the reason is always already set.
    fn ended_reason(&self, refusal: Error) -> Error {
        self.connection_state.check_ended().err().unwrap_or(refusal)
    }

    /// The one funnel for every bus-originated write, so no send can reach a
    /// socket the session no longer owns. The dispatcher's own reconnect
    /// handshake writes through `Connection`, not here.
    fn write_message(&self, message: &[u8]) -> Result<(), Error> {
        self.throttle(message);
        self.write_now(message)
    }

    /// Waits for the rate limiter, if any. Shared-channel paths call this
    /// before taking the `counts` lock, so a throttled request never holds up
    /// a cancel queued on that lock.
    fn throttle(&self, message: &[u8]) {
        if let Some(limiter) = self.rate_limiter.get() {
            limiter.acquire_blocking(is_cancel_frame(message));
        }
    }

    /// Writes without throttling; the connected check runs after any wait.
    fn write_now(&self, message: &[u8]) -> Result<(), Error> {
        self.ensure_connected()?;
        self.connection.write_message(message)
    }

    fn reset(&self) {
        debug!("reset message bus");
        // Before failing the routes, so a subscription that reads the reset
        // already sees the new generation.
        self.session_generation.fetch_add(1, Ordering::AcqRel);

        self.requests.fail_all(|| Error::ConnectionReset.into());
        self.orders.fail_all(|| Error::ConnectionReset.into());
        self.shared_channels.fail_all(|| Error::ConnectionReset.into());
        self.shared_channels.reset_counts();
        // Aliases of the routes just failed.
        self.executions.clear();
        self.order_taps.reset();
    }

    // The three cleanup handlers below remove a registration only when it
    // holds the cancelled or dropped subscription's lease: a signal can be
    // processed arbitrarily late, and unconditional removal would take out a
    // newer registration under the same key (place then cancel on one order
    // id, or an order update stream recreated after a reconnect reset).
    // Identity alone, not liveness: a sync subscription has one holder, and
    // its lease is released before its signal is sent, at cancel or drop.
    //
    // `clean_request` and `clean_order` also drop the subscription's
    // execution-id aliases, matched by lease rather than key, so a stale
    // signal still releases its own aliases and never a newer registration's.
    // Not gated on `removed`: a stale signal still owns aliases to release.

    fn clean_request(&self, request_id: RequestId, lease: &LeaseRef) {
        let removed = self.requests.remove_if_same(request_id, lease);
        let aliases = self.executions.remove_all_same(lease);
        debug!(
            "cleanup request_id {request_id}: removed={removed}, aliases={aliases}, requests.len()={}",
            self.requests.len()
        );
    }

    fn clean_order(&self, order_id: OrderId, lease: &LeaseRef) {
        let removed = self.orders.remove_if_same(order_id, lease);
        let aliases = self.executions.remove_all_same(lease);
        debug!(
            "cleanup order_id {order_id}: removed={removed}, aliases={aliases}, orders.len()={}",
            self.orders.len()
        );
    }

    fn clear_order_update_stream(&self, lease: &LeaseRef) {
        let removed = self.order_taps.release_updates(lease);
        debug!("cleanup order_update_stream: removed={removed}");
    }

    fn read_message(&self) -> Response {
        self.connection.read_message()
    }
    pub(crate) fn dispatch(&self) -> Result<(), Error> {
        match self.read_message() {
            Ok(message) => {
                if message.is_shutdown() {
                    self.request_shutdown();
                    Err(Error::Shutdown)
                } else {
                    self.dispatch_message(message);
                    Ok(())
                }
            }
            Err(ref err) if err.is_read_timeout() => {
                if self.is_shutting_down() {
                    debug!("dispatcher thread exiting");
                    return Err(Error::Shutdown);
                }
                Ok(())
            }
            Err(ref err) if err.is_connection_lost() => {
                if self.is_shutting_down() {
                    debug!("dispatcher thread exiting");
                    return Err(Error::Shutdown);
                }
                error!("error reading next message (will attempt reconnect): {err:?}");
                self.connection_state.set_disconnected();

                // Fail every registered channel before reconnecting, not
                // after: nothing they wait for can arrive on the dead
                // session, and the reconnect can run for minutes.
                self.reset();

                match self.connection.reconnect() {
                    Ok(()) => {}
                    // Shutdown was requested while reconnecting: not a failure,
                    // and the flag is already set, so just exit the dispatcher.
                    Err(Error::Shutdown) => {
                        debug!("shutdown requested during reconnect; dispatcher thread exiting");
                        return Err(Error::Shutdown);
                    }
                    Err(reconnect_err) => {
                        error!("failed to reconnect to TWS/Gateway: {reconnect_err:?}");
                        // Latched before the shutdown, which keeps it: later
                        // sends and waiters see `ConnectionFailed`, not `Shutdown`.
                        self.connection_state.fail();
                        self.request_shutdown();
                        return Err(Error::ConnectionFailed);
                    }
                }

                // Shutdown may have been requested while the connect was in
                // flight; the reconnect succeeded anyway, so check before
                // reporting the session as live.
                if self.is_shutting_down() {
                    debug!("shutdown requested during reconnect; dispatcher thread exiting");
                    return Err(Error::Shutdown);
                }

                // Before reporting the session as live; see
                // `HandshakeState::refresh`.
                if let Some(state) = self.handshake_state.get() {
                    state.refresh(&self.connection.connection_metadata());
                }

                info!("successfully reconnected to TWS/Gateway");
                self.connection_state.set_connected();

                // TWS never frames the socket reconnect itself and does not replay
                // restoration notices on the new connection, so the notice fan-out —
                // the carrier 1100/1101/1102 arrive on — learns the socket generation
                // changed from this synthesized notice alone; see
                // `TRANSPORT_RECONNECT_CODE`. Published after the session is live
                // so a consumer that resubscribes on it lands on the new session,
                // into maps the reset above can no longer wipe.
                self.connection.notice_broadcaster.broadcast(transport_reconnect_notice());
                Ok(())
            }
            Err(Error::Shutdown) => {
                debug!("socket closed; dispatcher thread exiting");
                self.request_shutdown();
                Err(Error::Shutdown)
            }
            Err(err) => {
                error!("error reading next message (shutting down): {err:?}");
                // Nobody asked for this shutdown: report the session as failed.
                self.connection_state.fail();
                self.request_shutdown();
                Err(err)
            }
        }
    }

    // Dispatcher thread reads messages from TWS and dispatches them to
    // appropriate channel.
    fn start_dispatcher_thread(self: &Arc<Self>) -> JoinHandle<()> {
        let message_bus = Arc::clone(self);
        thread::spawn(move || {
            loop {
                match message_bus.dispatch() {
                    Ok(_) => {}
                    Err(Error::Shutdown | Error::ConnectionFailed) => break,
                    Err(e) => {
                        error!("Dispatcher encountered an error: {e:?}");
                        break;
                    }
                }
            }
            debug!("Dispatcher thread finished.");
        })
    }

    fn dispatch_message(&self, message: ResponseMessage) {
        match determine_routing(&message) {
            RoutingDecision::Error(payload) => self.route_error_message(payload),
            RoutingDecision::ByOrderId(_) => {
                // Order-related messages
                self.process_orders(message);
            }
            RoutingDecision::ByRequestId(id) => {
                self.process_response_with_id(WireId::classify(id), message, false);
            }
            _ => {
                // All other messages
                self.process_response(message, false);
            }
        }
    }

    /// Route an error frame by severity and request id. Mirrors the async
    /// transport's `route_error_message`.
    fn route_error_message(&self, payload: DecodedError) {
        let sent_to_update_stream = order_update_notice(&payload).is_some_and(|notice| self.send_order_update_item(RoutedItem::Notice(notice)));
        match classify_error(payload) {
            ErrorDisposition::NoticeOnly(notice) => {
                notice.log();
                self.connection.notice_broadcaster.broadcast(notice);
            }
            ErrorDisposition::NoticeAndFailOneShots(notice, error) => {
                notice.log();
                self.connection.notice_broadcaster.broadcast(notice);
                self.shared_channels.fail_one_shot_channels(|| RoutedItem::Error(error.clone()));
            }
            ErrorDisposition::Route(id, item) => {
                self.order_taps.publish_rejection(id, &item);
                self.deliver(id, item, sent_to_update_stream);
            }
        }
    }

    fn process_response(&self, message: ResponseMessage, routed: bool) {
        let id = message.request_id().and_then(WireId::classify);
        self.process_response_with_id(id, message, routed);
    }

    /// The id's range picks the table: a request and an order can never be
    /// confused, whatever is registered under the number.
    fn process_response_with_id(&self, id: Option<WireId>, message: ResponseMessage, routed: bool) {
        match id {
            Some(WireId::Request(request_id)) if self.requests.contains(&request_id) => {
                let _ = self.requests.deliver(&request_id, message.into());
                return;
            }
            Some(WireId::Order(order_id)) if self.orders.contains(&order_id) => {
                let _ = self.orders.deliver(&order_id, message.into());
                return;
            }
            _ => {}
        }
        if self.shared_channels.is_shared_response(message.message_type()) {
            self.shared_channels.send_message(message.message_type(), &message);
        } else if !routed {
            report_unroutable_frame(&message, &*self.connection.notice_broadcaster);
        }
    }

    /// Deliver a pre-classified Notice or Error to the request or order
    /// subscription its id names.
    fn deliver(&self, id: WireId, item: RoutedItem, sent_to_update_stream: bool) {
        let unrouted = match id {
            WireId::Request(request_id) => self.requests.deliver(&request_id, item),
            WireId::Order(order_id) => self.orders.deliver(&order_id, item),
        };
        if let Err(item) = unrouted {
            if !sent_to_update_stream {
                log_orphan(id, &item);
            }
        }
    }

    fn process_orders(&self, message: ResponseMessage) {
        let foreign = is_foreign(&message, self.connection.client_id);
        let strategy = order_routing_strategy(message.message_type(), foreign);
        let message_order_id = message.order_id().map(OrderId::from);
        let message_request_id = message.request_id().and_then(RequestId::from_raw);

        if !foreign {
            self.order_taps.publish_status(&message);
        }

        match strategy {
            OrderRoutingStrategy::OrderUpdateOnly => {
                self.send_order_update(&message);
            }
            OrderRoutingStrategy::ExecutionData => {
                let sent_to_update_stream = self.send_order_update(&message);
                let execution_id = message.execution_id();
                if let Err(item) = self.deliver_to_order_or_request(message_order_id, message_request_id, message.into(), execution_id.as_ref()) {
                    if !sent_to_update_stream {
                        warn!("could not route message {item:?}");
                    }
                }
            }
            OrderRoutingStrategy::ExecutionDataEnd => {
                if let Err(item) = self.deliver_to_order_or_request(message_order_id, message_request_id, message.into(), None) {
                    warn!("could not route message {item:?}");
                }
            }
            OrderRoutingStrategy::OrderAndShared => {
                let sent_to_update_stream = self.send_order_update(&message);
                let shared = self.shared_channels.send_message(message.message_type(), &message);
                let unrouted = match message_order_id {
                    Some(order_id) => self.orders.deliver(&order_id, message.into()).err(),
                    None => Some(message.into()),
                };
                if let Some(item) = unrouted {
                    if !shared && !sent_to_update_stream {
                        warn!("could not route message {item:?}");
                    }
                }
            }
            OrderRoutingStrategy::ByExecutionId => {
                let sent_to_update_stream = self.send_order_update(&message);

                let unrouted = match message.execution_id() {
                    Some(execution_id) => self.executions.deliver(&execution_id, message.into()),
                    None => Err(message.into()),
                };
                if let Err(item) = unrouted {
                    if !sent_to_update_stream {
                        warn!("could not route commission report {item:?}");
                    }
                }
            }
            OrderRoutingStrategy::SharedOnly => {
                self.shared_channels.send_message(message.message_type(), &message);
            }
            OrderRoutingStrategy::ByRequestId => {
                let execution_id = message.execution_id();
                if let Err(item) = self.deliver_to_order_or_request(None, message_request_id, message.into(), execution_id.as_ref()) {
                    debug!("another client's execution answers no request: {item:?}");
                }
            }
            OrderRoutingStrategy::Ignore => {}
            OrderRoutingStrategy::ByOrderId => {
                warn!("unhandled order message type: {message:?}");
            }
        }
    }

    /// Register `request_id`'s channel, optionally with an unread-item cap,
    /// then write the request.
    fn open_request(&self, request_id: RequestId, message: &[u8], bound: Option<BufferBound>) -> Result<InternalSubscription, Error> {
        // Read before the gate: a reset after this read leaves the
        // subscription in the old session, which is where its request went.
        let session = self.session_generation();
        self.ensure_connected()?;

        let (sender, receiver) = channel::unbounded();
        let lease = Lease::new();
        let lease_ref = lease.downgrade();

        match bound {
            Some(bound) => self.requests.insert_bounded(request_id, sender.clone(), lease_ref.clone(), bound),
            None => self.requests.insert(request_id, sender.clone(), lease_ref.clone()),
        };

        // The gate can close between `ensure_connected` and the write, so take
        // the registration back out on failure rather than leave a channel no
        // reset will clear. `remove_if_same` so a newer registration under the
        // same id survives.
        if let Err(e) = self.write_message(message) {
            self.requests.remove_if_same(request_id, &lease_ref);
            return Err(e);
        }

        let subscription = SubscriptionBuilder::new()
            .receiver(receiver)
            .sender(sender)
            .signaler(self.signals_send.clone())
            .lease(lease)
            .request_id(request_id)
            .session(session)
            .build();

        Ok(subscription)
    }

    /// Deliver to `order_id`'s route, else `request_id`'s, aliasing the route
    /// under `execution_id` for the commission report that follows. Hands
    /// the item back when neither is registered.
    fn deliver_to_order_or_request(
        &self,
        order_id: Option<OrderId>,
        request_id: Option<RequestId>,
        item: RoutedItem,
        execution_id: Option<&String>,
    ) -> Result<(), RoutedItem> {
        match order_id {
            Some(id) => self.orders.deliver_aliased(&id, item, execution_id, &self.executions),
            None => Err(item),
        }
        .or_else(|item| match request_id {
            Some(id) => self.requests.deliver_aliased(&id, item, execution_id, &self.executions),
            None => Err(item),
        })
    }

    // Sends an order update message to the order update stream if it exists.
    // Returns true if the message was sent to the order update stream.
    fn send_order_update(&self, message: &ResponseMessage) -> bool {
        self.send_order_update_item(message.clone().into())
    }

    fn send_order_update_item(&self, item: RoutedItem) -> bool {
        self.order_taps.send_update(item)
    }

    // The cleanup thread receives signals as subscribers are cancelled or
    // dropped and releases the sender channels. Exits promptly on shutdown via
    // select! over the signal channel and the shutdown-notify channel — no
    // polling.
    fn start_cleanup_thread(self: &Arc<Self>) -> JoinHandle<()> {
        let message_bus = Arc::clone(self);

        thread::spawn(move || {
            let signal_recv = message_bus.signals_recv.clone();
            let shutdown_recv = message_bus.shutdown_recv.clone();

            loop {
                crossbeam::select! {
                    recv(signal_recv) -> signal => match signal {
                        Ok(Signal::Request(request_id, lease)) => message_bus.clean_request(request_id, &lease),
                        Ok(Signal::Order(order_id, lease)) => message_bus.clean_order(order_id, &lease),
                        Ok(Signal::OrderUpdateStream(lease)) => message_bus.clear_order_update_stream(&lease),
                        Ok(Signal::OrderStatus(order_id, lease)) => message_bus.order_taps.release_status(order_id, &lease),
                        Ok(Signal::Shared(lease)) => message_bus.shared_channels.remove(&lease),
                        Err(_) => {
                            debug!("cleanup signal channel closed");
                            return;
                        }
                    },
                    recv(shutdown_recv) -> _ => {
                        debug!("cleanup thread exiting");
                        return;
                    }
                }
            }
        })
    }

    pub(crate) fn process_messages(self: &Arc<Self>) -> Result<(), Error> {
        let handle = self.start_dispatcher_thread();
        self.add_join_handle(handle);

        let handle = self.start_cleanup_thread();
        self.add_join_handle(handle);

        Ok(())
    }

    fn add_join_handle(&self, handle: JoinHandle<()>) {
        let mut handles = lock(&self.handles);
        handles.push(handle);
    }

    pub fn join(&self) {
        let mut handles = lock(&self.handles);

        for handle in handles.drain(..) {
            if let Err(e) = handle.join() {
                warn!("could not join thread: {e:?}");
            }
        }
    }

    // Registers a subscription of `message_type` and writes its request;
    // `account` for account updates (see `SharedCounts`).
    fn send_shared(&self, message_type: OutgoingMessages, account: Option<&AccountId>, message: &[u8]) -> Result<InternalSubscription, Error> {
        // See `open_request`.
        let session = self.session_generation();
        self.ensure_connected()?;

        // A queue of its own, registered before the write so no response can
        // arrive ahead of it. A failed write or a refused account takes the
        // registration with it.
        let (sender, receiver) = channel::unbounded();
        let lease = Lease::new();
        let lease_ref = lease.downgrade();
        self.shared_channels.add(message_type, sender.clone(), lease_ref.clone());
        self.throttle(message);
        let ticket = match self.shared_channels.subscribe(message_type, account, || self.write_now(message)) {
            Ok(ticket) => ticket,
            Err(e) => {
                self.shared_channels.remove(&lease_ref);
                return Err(e);
            }
        };

        // The lease is the drop signal's identity: `Signal::Shared` removes
        // exactly this registration.
        let subscription = SubscriptionBuilder::new()
            .receiver(receiver)
            .sender(sender)
            .signaler(self.signals_send.clone())
            .lease(lease)
            .shared(ticket)
            .session(session)
            .build();

        Ok(subscription)
    }
}

impl<S: Stream> MessageBus for TcpMessageBus<S> {
    fn send_request_capped(&self, request_id: RequestId, message: &[u8], bound: Option<BufferBound>) -> Result<InternalSubscription, Error> {
        self.open_request(request_id, message, bound)
    }

    fn send_order_request(&self, order_id: OrderId, message: &[u8]) -> Result<InternalSubscription, Error> {
        // See `open_request`.
        let session = self.session_generation();
        self.ensure_connected()?;

        let (sender, receiver) = channel::unbounded();
        let lease = Lease::new();
        let lease_ref = lease.downgrade();

        self.orders.insert(order_id, sender.clone(), lease_ref.clone());
        debug!("Registered order subscription for order_id={}", order_id);

        // See `send_request`: a failed write takes its registration with it.
        if let Err(e) = self.write_message(message) {
            self.orders.remove_if_same(order_id, &lease_ref);
            return Err(e);
        }

        let subscription = SubscriptionBuilder::new()
            .receiver(receiver)
            .sender(sender)
            .signaler(self.signals_send.clone())
            .lease(lease)
            .order_id(order_id)
            .session(session)
            .build();

        Ok(subscription)
    }

    fn send_message(&self, message: &[u8]) -> Result<(), Error> {
        self.write_message(message)?;
        Ok(())
    }

    fn send_market_data_type(&self, message: &[u8]) -> Result<(), Error> {
        // Remembered before the write: a reconnect between the two would
        // otherwise replay the previous type after this call returned Ok.
        self.connection.remember_market_data_type(message.to_vec());
        self.write_message(message)
    }

    fn create_order_status_subscription(&self, order_id: OrderId) -> Result<InternalSubscription, Error> {
        // Allowed mid-reconnect, like the order-update stream below.
        let NewTap { sender, receiver, lease } = self.order_taps.subscribe_status(order_id, 0).map_err(|e| self.ended_reason(e))?;
        Ok(SubscriptionBuilder::new()
            .receiver(receiver)
            .sender(sender)
            .signaler(self.signals_send.clone())
            .lease(lease)
            .order_status(order_id)
            .session(self.session_generation())
            .build())
    }

    fn create_order_update_subscription(&self) -> Result<InternalSubscription, Error> {
        // Not `ensure_connected`: nothing is written, and the stream may be
        // created while a reconnect is in progress. The lease gives the
        // subscription's drop signal its identity — see
        // `clear_order_update_stream`.
        let NewTap { sender, receiver, lease } = self.order_taps.subscribe_updates(0).map_err(|e| self.ended_reason(e))?;
        Ok(SubscriptionBuilder::new()
            .receiver(receiver)
            .sender(sender)
            .signaler(self.signals_send.clone())
            .lease(lease)
            .session(self.session_generation())
            .build())
    }

    fn send_shared_request(&self, message_type: OutgoingMessages, message: &[u8]) -> Result<InternalSubscription, Error> {
        self.send_shared(message_type, None, message)
    }

    fn send_account_updates_request(&self, account: &AccountId, message: &[u8]) -> Result<InternalSubscription, Error> {
        self.send_shared(OutgoingMessages::RequestAccountData, Some(account), message)
    }

    fn cancel_shared_subscription(&self, ticket: SharedTicket, message: Option<&[u8]>) -> Result<(), Error> {
        // Throttled before the lock: some cancels (account updates) are
        // requests with an off flag and do wait.
        if let Some(message) = message {
            self.throttle(message);
        }
        self.shared_channels
            .unsubscribe(ticket, || message.map_or(Ok(()), |message| self.write_now(message)))
    }

    fn notice_subscribe(&self) -> NoticeStream {
        NoticeStream::new(self.connection.notice_broadcaster.subscribe())
    }

    fn ensure_shutdown(&self) {
        self.request_shutdown();
        self.join();
    }

    fn wait_connected(&self) -> Result<(), Error> {
        self.connection_state.wait_connected()
    }

    fn is_connected(&self) -> bool {
        self.connection_state.is_connected() && !self.is_shutting_down()
    }

    fn session_generation(&self) -> u64 {
        self.session_generation.load(Ordering::Acquire)
    }
}

#[derive(Debug)]
struct Entry<V> {
    sender: Sender<V>,
    /// The subscription's lease: which subscription this is, and whether it lives.
    lease: LeaseRef,
    /// The unread-item cap of a route opened with a `send_request_capped` bound.
    bound: Option<BoundState>,
}

impl<V> Entry<V> {
    fn new(sender: Sender<V>, lease: LeaseRef) -> Self {
        Self { sender, lease, bound: None }
    }

    fn closed(&self) -> bool {
        self.bound.as_ref().is_some_and(BoundState::closed)
    }

    /// An unbounded registration sharing this one's channel and lease.
    fn alias(&self) -> Self {
        Self::new(self.sender.clone(), self.lease.clone())
    }
}

impl Entry<RoutedItem> {
    /// Send `item`, subject to a bounded route's cap ([`BoundState::admit`]).
    /// Never blocks.
    fn deliver(&self, id: &impl std::fmt::Debug, item: RoutedItem) {
        let item = match &self.bound {
            Some(bound) => match bound.admit(&item, self.sender.len()) {
                Admit::Deliver => item,
                Admit::Overflow => bound.overflow_error(),
                Admit::Discard => {
                    trace!("discarding item for closed route {id:?}");
                    return;
                }
            },
            None => item,
        };
        // An unbounded send fails only once the receiver is gone: the
        // subscription was dropped and its cleanup signal hasn't run yet.
        if self.sender.send(item).is_err() {
            trace!("discarding item for dropped subscription {id:?}");
        } else {
            warn_if_backlogged(format_args!("subscription queue for {id:?}"), self.sender.len());
        }
    }
}

#[derive(Debug)]
struct SenderHash<K, V> {
    senders: RwLock<HashMap<K, Entry<V>>>,
}

impl<K: std::hash::Hash + Eq + std::fmt::Debug, V: std::fmt::Debug> SenderHash<K, V> {
    pub fn new() -> Self {
        Self {
            senders: RwLock::new(HashMap::new()),
        }
    }

    fn read(&self) -> RwLockReadGuard<'_, HashMap<K, Entry<V>>> {
        read_lock(&self.senders)
    }

    fn write(&self) -> RwLockWriteGuard<'_, HashMap<K, Entry<V>>> {
        write_lock(&self.senders)
    }

    #[cfg(test)]
    pub fn copy_sender(&self, id: K) -> Option<Sender<V>> {
        self.with_entry(&id, |entry| entry.sender.clone())
    }

    #[cfg(test)]
    pub fn lease(&self, id: K) -> Option<LeaseRef> {
        self.with_entry(&id, |entry| entry.lease.clone())
    }

    /// Run `f` on `id`'s entry while holding the read lock, so no removal can
    /// land between the lookup and `f`.
    #[cfg(test)]
    fn with_entry<R>(&self, id: &K, f: impl FnOnce(&Entry<V>) -> R) -> Option<R> {
        let senders = self.read();
        senders.get(id).map(f)
    }

    pub fn insert(&self, id: K, sender: Sender<V>, lease: LeaseRef) {
        self.insert_entry(id, Entry::new(sender, lease));
    }

    /// Like [`insert`](Self::insert), with an unread-item cap: see [`BoundState::admit`].
    pub fn insert_bounded(&self, id: K, sender: Sender<V>, lease: LeaseRef, bound: BufferBound) {
        let entry = Entry {
            bound: Some(BoundState::new(bound)),
            ..Entry::new(sender, lease)
        };
        self.insert_entry(id, entry);
    }

    fn insert_entry(&self, id: K, entry: Entry<V>) {
        self.write().insert(id, entry);
    }

    /// Remove the entry for `id` only if it holds `lease`. Returns whether an
    /// entry was removed. Used by signal cleanup so a stale signal cannot
    /// remove a newer registration under the same key.
    pub fn remove_if_same(&self, id: K, lease: &LeaseRef) -> bool {
        match self.write().entry(id) {
            hash_map::Entry::Occupied(registered) if registered.get().lease.is(lease) => {
                registered.remove();
                true
            }
            _ => false,
        }
    }

    /// Remove every entry holding `lease`. Returns how many were removed.
    pub fn remove_all_same(&self, lease: &LeaseRef) -> usize {
        let mut senders = self.write();
        let before = senders.len();
        senders.retain(|_, registered| !registered.lease.is(lease));
        before - senders.len()
    }

    pub fn contains(&self, id: &K) -> bool {
        let senders = self.read();
        senders.contains_key(id)
    }

    pub fn len(&self) -> usize {
        let senders = self.read();
        senders.len()
    }

    pub fn clear(&self) {
        let mut senders = self.write();
        senders.clear();
    }
}

impl<K: std::hash::Hash + Eq + std::fmt::Debug> SenderHash<K, RoutedItem> {
    /// Deliver `item` to `id`'s route; hands it back when nothing is
    /// registered under `id`.
    pub fn deliver(&self, id: &K, item: RoutedItem) -> Result<(), RoutedItem> {
        self.deliver_then(id, item, |_| {})
    }

    /// Deliver `item` to `id`'s route, then alias that route under `alias`
    /// in `aliases` while still holding `id`'s read lock, so a concurrent
    /// cleanup either removes the registration first (no alias is stored) or
    /// prunes the alias after it. Lock order: `self` (read), then `aliases`
    /// (write); nothing takes them the other way round. Hands the item back
    /// when nothing is registered under `id`.
    pub fn deliver_aliased<A: std::hash::Hash + Eq + Clone + std::fmt::Debug>(
        &self,
        id: &K,
        item: RoutedItem,
        alias: Option<&A>,
        aliases: &SenderHash<A, RoutedItem>,
    ) -> Result<(), RoutedItem> {
        self.deliver_then(id, item, |entry| {
            if let Some(alias) = alias {
                aliases.insert_entry(alias.clone(), entry.alias());
            }
        })
    }

    /// Deliver `item` to `id`'s route and run `then` on it, both under the
    /// read lock.
    fn deliver_then(&self, id: &K, item: RoutedItem, then: impl FnOnce(&Entry<RoutedItem>)) -> Result<(), RoutedItem> {
        let senders = self.read();
        let Some(entry) = senders.get(id) else {
            return Err(item);
        };
        entry.deliver(id, item);
        then(entry);
        Ok(())
    }

    /// Send `message_fn()` to every route, skipping closed bounded ones (their
    /// stream already ended), then clear them all under the same write lock,
    /// so a route registered meanwhile can't be cleared unnotified. A send
    /// fails only for a dropped subscription not yet cleaned up, which has
    /// nobody to tell.
    pub fn fail_all<F>(&self, message_fn: F)
    where
        F: Fn() -> RoutedItem,
    {
        let mut senders = self.write();
        for entry in senders.values().filter(|entry| !entry.closed()) {
            let _ = entry.sender.send(message_fn());
        }
        senders.clear();
    }
}

/// What `close` shuts down, and whether it has.
///
/// Lock order: take this after `reader` / `writer`, never wait on them while
/// holding it. `reconnect` and the error path of `read_message` / `write_all`
/// follow it, and `close` takes only this, so a reconnect waiting behind a
/// stuck write can never hold up the close that unsticks that write.
#[derive(Debug)]
struct CloseHandle {
    /// A clone of the active stream; `reconnect` refreshes it.
    stream: TcpStream,
    /// Latched by `close`; `reconnect` re-checks it under the same lock, so a
    /// stream it installs is either refused or closed after it.
    closed: bool,
}

#[derive(Debug)]
pub(crate) struct TcpSocket {
    reader: Mutex<TcpStream>,
    writer: Mutex<TcpStream>,
    close_handle: Mutex<CloseHandle>,
    connection_url: String,
    tcp_no_delay: bool,
    /// Byte-level capture of the inbound stream. Disabled unless
    /// `IBAPI_RAW_CAPTURE_DIR` is set; see [`RawFrameTap`].
    tap: RawFrameTap,
}
impl TcpSocket {
    pub fn connect(address: &str, tcp_no_delay: bool) -> Result<Self, Error> {
        let stream = TcpStream::connect(address)?;
        Self::new(stream, address, tcp_no_delay)
    }

    pub fn new(stream: TcpStream, connection_url: &str, tcp_no_delay: bool) -> Result<Self, Error> {
        let writer = stream.try_clone()?;
        let close_handle = CloseHandle {
            stream: stream.try_clone()?,
            closed: false,
        };

        stream.set_read_timeout(Some(TWS_READ_TIMEOUT))?;
        stream.set_nodelay(tcp_no_delay)?;

        Ok(Self {
            reader: Mutex::new(stream),
            writer: Mutex::new(writer),
            close_handle: Mutex::new(close_handle),
            connection_url: connection_url.to_string(),
            tcp_no_delay,
            tap: RawFrameTap::from_env(),
        })
    }

    /// An I/O failure after `close` is the close, not the peer.
    fn closed_or(&self, err: Error) -> Error {
        if lock(&self.close_handle).closed {
            Error::Shutdown
        } else {
            err
        }
    }
}

impl Reconnect for TcpSocket {
    fn reconnect(&self) -> Result<(), Error> {
        let stream = TcpStream::connect(&self.connection_url)?;
        stream.set_read_timeout(Some(TWS_READ_TIMEOUT))?;
        stream.set_nodelay(self.tcp_no_delay)?;

        // The close handle last; see `CloseHandle` for the order.
        let mut reader = self.reader.lock()?;
        let mut writer = self.writer.lock()?;
        let mut handle = self.close_handle.lock()?;
        if handle.closed {
            // Dropping `stream` closes the new connection.
            return Err(Error::Shutdown);
        }
        *reader = stream.try_clone()?;
        *writer = stream.try_clone()?;
        handle.stream = stream;

        // One capture file per TCP stream: splicing two of them would
        // read back as a desync at the seam that never happened.
        self.tap.start_new_segment();

        Ok(())
    }
    fn sleep(&self, duration: std::time::Duration, shutdown: &ShutdownSignal) {
        shutdown.wait_timeout(duration)
    }
}

pub(crate) trait Reconnect {
    /// Replace the connection. Refused with [`Error::Shutdown`] after
    /// [`Io::close`].
    fn reconnect(&self) -> Result<(), Error>;
    /// Wait out the reconnect backoff, returning early once `shutdown` is
    /// requested. In-memory test streams return immediately.
    fn sleep(&self, duration: std::time::Duration, shutdown: &ShutdownSignal);
}

pub(crate) trait Stream: Io + Reconnect + Sync + Send + 'static + std::fmt::Debug {}
impl Stream for TcpSocket {}

/// Reads one frame's bytes, keeping partial progress across read timeouts.
///
/// `read_exact` drops the bytes it consumed when a read times out, and the
/// dispatcher treats a timeout as idle, so a frame straddling the socket's
/// read timeout used to desync the stream for good (#892). Here a timeout
/// before the frame's first byte is returned as-is (idle: the dispatcher
/// polls its shutdown flag, the handshake fails fast); after it, timeouts are
/// waited out, up to [`MID_FRAME_TIMEOUT_LIMIT`] in a row for the frame.
struct FrameRead<'a, R> {
    reader: &'a mut R,
    /// Any byte of this frame consumed.
    started: bool,
    /// Consecutive timeouts since the last progress, across prefix and body.
    timeouts: u32,
}

impl<'a, R: Read> FrameRead<'a, R> {
    fn new(reader: &'a mut R) -> Self {
        Self {
            reader,
            started: false,
            timeouts: 0,
        }
    }

    fn fill(&mut self, buf: &mut [u8]) -> Result<(), Error> {
        let mut filled = 0;
        while filled < buf.len() {
            match self.reader.read(&mut buf[filled..]) {
                // `read_exact`'s message, which the handshake surfaces in `ConnectionRejected`.
                Ok(0) => return Err(std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "failed to fill whole buffer").into()),
                Ok(n) => {
                    filled += n;
                    self.started = true;
                    self.timeouts = 0;
                }
                Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {}
                Err(e) if matches!(e.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut) => {
                    if !self.started {
                        return Err(e.into());
                    }
                    self.timeouts += 1;
                    if self.timeouts > MID_FRAME_TIMEOUT_LIMIT {
                        return Err(Error::InvalidFrame(format!("frame stalled mid-read for {} read timeouts", self.timeouts)));
                    }
                }
                Err(e) => return Err(e.into()),
            }
        }
        Ok(())
    }
}

/// Read the 4-byte big-endian length prefix, taps it, then validates it.
///
/// The tap runs *before* [`validate_frame_length`] on purpose — a prefix that
/// fails validation is exactly the byte sequence a framing desync leaves
/// behind, so it has to reach the capture even though it never reaches a
/// caller. See [`RawFrameTap`].
fn read_header(frame: &mut FrameRead<'_, impl Read>, tap: &RawFrameTap) -> Result<usize, Error> {
    let mut buffer = [0_u8; 4];
    frame.fill(&mut buffer)?;
    tap.record_length_prefix(&buffer);
    validate_frame_length(u32::from_be_bytes(buffer) as usize)
}

pub(crate) fn read_message(reader: &mut impl Read, tap: &RawFrameTap) -> Result<Vec<u8>, Error> {
    let mut frame = FrameRead::new(reader);
    let message_size = read_header(&mut frame, tap)?;
    let mut data = vec![0_u8; message_size];
    frame.fill(&mut data)?;
    tap.record_body(&data);
    Ok(data)
}

impl Io for TcpSocket {
    fn read_message(&self) -> Result<Vec<u8>, Error> {
        let mut reader = self.reader.lock()?;
        read_message(&mut *reader, &self.tap).map_err(|err| self.closed_or(err))
    }

    fn write_all(&self, buf: &[u8]) -> Result<(), Error> {
        let mut writer = self.writer.lock()?;
        writer.write_all(buf).map_err(|err| self.closed_or(err.into()))
    }

    fn close(&self) {
        let mut handle = lock(&self.close_handle);
        if std::mem::replace(&mut handle.closed, true) {
            return;
        }
        // Socket-level, not per clone: sends FIN, and fails pending and later
        // reads and writes on every clone, so retained owners can't hold the
        // session open.
        if let Err(e) = handle.stream.shutdown(std::net::Shutdown::Both) {
            debug!("socket shutdown returned: {e:?}");
        }
    }
}

pub(crate) trait Io {
    fn read_message(&self) -> Result<Vec<u8>, Error>;
    fn write_all(&self, buf: &[u8]) -> Result<(), Error>;
    /// Release the physical stream, even while owners still hold it: pending
    /// and later reads and writes fail with [`Error::Shutdown`], and
    /// `reconnect` refuses. Idempotent. Twin of `AsyncIo::close`.
    fn close(&self);
}

mod connection_signal;
mod shutdown;
pub(crate) use connection_signal::ConnectionSignal;
pub(crate) use shutdown::ShutdownSignal;

#[cfg(test)]
mod memory;
#[cfg(test)]
pub(crate) use memory::MemoryStream;

#[cfg(test)]
pub(crate) mod test_listener;

#[cfg(test)]
#[path = "sync_tests.rs"]
mod tests;