gnitz-core 0.1.7

//! The protocol session: a sans-io connection state machine. Nothing in it
//! waits; whoever drives [`Session::step`] does.
//!
//! A frame is of what its [`WireLane`] names. Replies arrive in request order,
//! each the head slot's until its last train terminates; a pushed train and
//! the answer to a sync arrive whole between two replies.

use std::borrow::Cow;
use std::collections::hash_map::Entry;
use std::collections::{HashMap, VecDeque};
use std::future::Future;
use std::os::fd::{BorrowedFd, RawFd};
use std::pin::Pin;
use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
use std::task::{Context, Poll, Waker};
use std::time::{Duration, Instant};

use crate::error::ClientError;
use crate::protocol::wal_block::decode_wal_block_into;
use crate::{
    encode_ddl_txn, encode_frame, encode_push_txn, sys_schema, ClientTransport, ProtocolError, PushFamily, Schema,
    ZSetBatch,
};
use gnitz_wire::control::{peek_control_block, ControlHeader, DecodedControl};
use gnitz_wire::txn_frame::{self, kept_as};
use gnitz_wire::CONNECT_TIMEOUT;
use gnitz_wire::{ClientVerb, WireConflictMode, WireLane};
use gnitz_wire::{RelClass, RelDescriptorBlob, RelIndex};

/// Requests one connection may hold in flight; `submit` raises past it. A bound
/// on the memory a driver that never waits can pin, not a throughput knob.
pub const MAX_IN_FLIGHT: usize = 4096;

/// Unwritten frame bytes one connection may hold; `submit` raises past it. An
/// encoded push is a full copy of its batch, so the count above bounds no
/// memory on its own. Checked before queueing, so one frame up to the peer's
/// egress limit always goes through — what it bounds is a driver that submits
/// without flushing.
pub const MAX_QUEUED_BYTES: usize = 64 << 20;

/// One relation's read result.
#[derive(Debug)]
pub struct ScanReply {
    pub schema: Arc<Schema>,
    pub batch: ZSetBatch,
    /// The server LSN the read was served at. `None` for rows no server read
    /// produced: an answer off a mirrored copy, a constant, `EXPLAIN` or
    /// `RETURNING`.
    pub lsn: Option<u64>,
}

/// One relation, as a RESOLVE describes it.
#[derive(Debug)]
pub struct RelDescriptor {
    pub tid: u64,
    pub class: RelClass,
    /// A stream, or a view whose planner set [`gnitz_wire::sys_rows::ViewTabRow::pk_repeats`].
    pub pk_repeats: bool,
    /// [`gnitz_wire::TableProps::serial`].
    pub serial: bool,
    pub schema: Arc<Schema>,
    pub indexes: Vec<RelIndex>,
    /// The RESOLVE answer's descriptor token: what a request built from this
    /// descriptor carries, for the server to refuse it once the relation
    /// resolves differently.
    pub token: u64,
}

pub use gnitz_wire::control::Target;

impl From<&RelDescriptor> for Target {
    fn from(rel: &RelDescriptor) -> Self {
        Target { tid: rel.tid, token: rel.token }
    }
}

/// One reply frame's data block, kept undecoded: the owned frame buffer and the
/// block's extent within it. `block()` is the block itself.
#[derive(Debug)]
pub(crate) struct RawBlock {
    frame: Vec<u8>,
    block: std::ops::Range<usize>,
}

impl RawBlock {
    /// The data block's bytes, ready to decode against the reply schema.
    pub(crate) fn block(&self) -> &[u8] {
        &self.frame[self.block.clone()]
    }
}

pub use gnitz_wire::txn_frame::DeltaCursor;

// ── The request vocabulary ───────────────────────────────────────────────────

/// What a driver should wait for before its next `step`, and what it hands
/// back: two booleans on the connection.
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
pub struct Interest {
    pub read: bool,
    pub write: bool,
}

impl Interest {
    pub const NONE: Interest = Interest { read: false, write: false };
    pub const READ: Interest = Interest { read: true, write: false };
    pub const WRITE: Interest = Interest { read: false, write: true };

    pub fn is_empty(self) -> bool {
        self == Interest::NONE
    }

    /// The `poll(2)` event mask to wait for.
    pub fn poll_events(self) -> libc::c_short {
        (if self.read { libc::POLLIN } else { 0 }) | (if self.write { libc::POLLOUT } else { 0 })
    }

    /// What a `poll(2)` wakeup says is ready, as the set to hand `step`.
    /// `POLLHUP` / `POLLERR` fold into `read` — they arrive whether requested
    /// or not, and the step that follows reads the EOF or reset and surfaces
    /// it; a driver that ignored them would park forever on a dead peer.
    pub fn from_revents(revents: libc::c_short) -> Interest {
        Interest {
            read: revents & (libc::POLLIN | libc::POLLHUP | libc::POLLERR) != 0,
            write: revents & libc::POLLOUT != 0,
        }
    }
}

/// A request answered by one value — an LSN, or the base id of a run. It
/// borrows its inputs; the borrow ends at [`Session::submit`].
pub enum Request<'a> {
    /// A run of this many catalog object ids.
    AllocIds(u64),
    /// A run of `count` ids from the SERIAL sequence of `table`.
    AllocSerial { table: Target, count: u64 },
    /// An atomic DDL transaction: system-table batches, each named by its table
    /// id, under one durable SAL zone.
    DdlTxn(&'a [(u64, ZSetBatch)]),
    /// An atomic user-table push transaction: the server refuses it if some
    /// family's relation was written after that family's `basis`.
    PushTxn { families: &'a [PushFamily] },
    /// PUSH. The frame always carries `schema`'s record.
    Push {
        target: Target,
        schema: &'a Schema,
        batch: &'a ZSetBatch,
        mode: WireConflictMode,
    },
}

impl Request<'_> {
    /// Validate and encode: the frame, and the relation its ACK names.
    fn encode(self) -> Result<(Vec<u8>, u64), ClientError> {
        Ok(match self {
            Request::AllocIds(n) => {
                let hdr = ControlHeader::naming(ClientVerb::AllocIds, Target::from(0), n);
                (encode_frame(hdr, &[], None, None), 0)
            }
            Request::AllocSerial { table, count } => {
                let hdr = ControlHeader::naming(ClientVerb::AllocSerialRange, table, count);
                (encode_frame(hdr, &[], None, None), table.tid)
            }
            Request::DdlTxn(families) => {
                for (tid, batch) in families {
                    if gnitz_wire::sys_family_index(*tid).is_none() {
                        return Err(ClientError::from(format!("DDL family {tid} is not a system table")));
                    }
                    batch.validate(sys_schema(*tid))?;
                }
                (encode_ddl_txn(families), 0)
            }
            Request::PushTxn { families } => {
                for f in families {
                    f.batch.validate(&f.schema)?;
                }
                (encode_push_txn(families), 0)
            }
            Request::Push { target, schema, batch, mode } => {
                // In-process, so a convenience and never a trust boundary; the
                // server checks the same things. Here so no driver has to
                // remember to.
                batch.validate(schema)?;
                // The push verb marks the frame as a push independent of data
                // presence, so an empty batch (a legitimate empty Z-set delta)
                // is ACKed as a no-op push instead of being mistaken for a scan.
                let mut hdr = ControlHeader::naming(ClientVerb::Push, target, 0);
                hdr.flags.conflict_mode = mode;
                (
                    encode_frame(hdr, &[], Some(&schema.to_block()), Some(batch)),
                    target.tid,
                )
            }
        })
    }
}

/// One request in flight: how its reply decodes, the reply read so far, and
/// where it goes. A slot is answered only while it heads its queue, so its
/// state is the one train in progress, and it goes when the slot does.
enum Slot {
    /// A control-only ACK naming `tid`, its value in `arg0`.
    Ack { tid: u64, to: Promise<u64> },
    /// RESOLVE: the descriptor, or `None` when no such relation exists.
    /// Uncorrelated on the wire, because the request names no id.
    Resolve { to: Promise<Option<Arc<RelDescriptor>>> },
    Scan {
        tid: u64,
        reply_schema: Arc<Schema>,
        /// The train's rows so far.
        data: Option<ZSetBatch>,
        to: Promise<ScanReply>,
    },
    /// One view's delta read: a scan's train, whose terminal carries the cursor
    /// to read from next.
    Delta {
        tid: u64,
        reply_schema: Arc<Schema>,
        data: Option<ZSetBatch>,
        /// The subscription id the view is kept as from that cursor.
        keep: Option<u64>,
        to: Promise<(ScanReply, DeltaCursor)>,
    },
    /// One train per relation, each decoded under the schema paired with it.
    /// The train in progress is `rels[replies.len()]`.
    Multi {
        rels: Vec<(u64, Arc<Schema>)>,
        replies: Vec<ScanReply>,
        data: Option<ZSetBatch>,
        to: Promise<Vec<ScanReply>>,
    },
    /// One train per view, its blocks kept raw; the train in progress is
    /// `views[at]`, and goes to its [`Taker`]. A pushed train is one of these.
    DeltaPoll {
        views: Vec<u64>,
        at: usize,
        /// The poll this request is, see [`Polls::live`]; `None` for one that
        /// never had a reader.
        poll: Option<u64>,
        /// The subscription id `views[0]` is kept as, the rest counting up.
        first: u64,
        /// The train in progress, when a [`Taker::Holder`] takes it.
        blocks: Vec<RawBlock>,
    },
    /// A SYNC_PUSHED, queued among the syncs.
    Sync { to: Promise<()> },
}

/// Where a reply is, between its writer and its reader.
#[derive(Default)]
enum Awaited<T> {
    #[default]
    Outstanding,
    Arrived(Result<T, ClientError>),
    /// Handed on as it arrives.
    Routed(Box<dyn FnOnce(Result<T, ClientError>) + Send>),
    /// Awaited by a task its arrival wakes.
    Watched(Waker),
}

struct Cell<T>(Mutex<Awaited<T>>);

impl<T> Cell<T> {
    /// A panic under this lock leaves what it guards whole.
    fn lock(&self) -> MutexGuard<'_, Awaited<T>> {
        self.0.lock().unwrap_or_else(PoisonError::into_inner)
    }
}

/// A reply on its way, borrowing nothing. One to a request arrives as the
/// session is stepped: [`GnitzClient::wait`](crate::GnitzClient::wait) steps
/// until it has, and awaiting it directly leaves the stepping to someone else.
pub struct Sent<T>(Arc<Cell<T>>);

/// The writer of a [`Sent`]. Dropped unfulfilled, the reply is `Closed`.
pub struct Promise<T>(Option<Arc<Cell<T>>>);

/// A reply yet to arrive, and its writer.
pub fn promise<T>() -> (Promise<T>, Sent<T>) {
    let cell = Arc::new(Cell(Mutex::new(Awaited::Outstanding)));
    (Promise(Some(Arc::clone(&cell))), Sent(cell))
}

impl<T> Promise<T> {
    /// The reply arrives; a second one goes nowhere.
    pub fn fulfil(&mut self, reply: Result<T, ClientError>) {
        let Some(cell) = self.0.take() else {
            return;
        };
        let mut state = cell.lock();
        match std::mem::take(&mut *state) {
            Awaited::Routed(route) => {
                drop(state);
                route(reply)
            }
            Awaited::Watched(waiter) => {
                *state = Awaited::Arrived(reply);
                drop(state);
                waiter.wake()
            }
            Awaited::Outstanding | Awaited::Arrived(_) => *state = Awaited::Arrived(reply),
        }
    }
}

impl<T> Drop for Promise<T> {
    fn drop(&mut self) {
        self.fulfil(Err(ClientError::Closed));
    }
}

impl<T> Sent<T> {
    /// A reply that needed no request.
    pub fn ready(value: Result<T, ClientError>) -> Self {
        Sent(Arc::new(Cell(Mutex::new(Awaited::Arrived(value)))))
    }

    /// The reply, if it has arrived. It is handed out once.
    pub fn try_take(&mut self) -> Option<Result<T, ClientError>> {
        let mut state = self.0.lock();
        match std::mem::take(&mut *state) {
            Awaited::Arrived(reply) => Some(reply),
            other => {
                *state = other;
                None
            }
        }
    }

    /// Hand the reply to `route` — now if it has arrived, else from wherever
    /// it does: for a request's, inside the step that reads it.
    pub fn then(self, route: impl FnOnce(Result<T, ClientError>) + Send + 'static) {
        let mut state = self.0.lock();
        match std::mem::take(&mut *state) {
            Awaited::Arrived(reply) => {
                drop(state);
                route(reply)
            }
            _ => *state = Awaited::Routed(Box::new(route)),
        }
    }
}

impl<T> Future for Sent<T> {
    type Output = Result<T, ClientError>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let mut state = self.0.lock();
        match std::mem::take(&mut *state) {
            Awaited::Arrived(reply) => Poll::Ready(reply),
            _ => {
                *state = Awaited::Watched(cx.waker().clone());
                Poll::Pending
            }
        }
    }
}

/// How one view's train ends: the cursor its terminal carries, or the fault
/// that ended its position.
pub(crate) type PollEnd = Result<DeltaCursor, ClientError>;

/// What a delta poll hands its reader as each frame arrives.
pub(crate) enum Polled {
    /// One data block of the view's train.
    Block(RawBlock),
    /// The view's terminal; the position is answered.
    End(PollEnd),
}

/// Where the trains of a delta poll go.
#[derive(Default)]
struct Polls {
    /// The poll whose reader is live.
    live: u64,
    /// The live poll's results, in item order: each view's blocks, then its
    /// one [`Polled::End`], however the poll ends.
    queue: VecDeque<Polled>,
    /// What was pushed for each subscription the session holds, by its id,
    /// kept until its owner takes it. An id with no entry is one nobody holds.
    subs: HashMap<u64, Parked>,
    /// The server may hold a subscription of this connection: a read it was
    /// asked to keep was answered with a cursor since the last sync, or that
    /// sync named one.
    server_holds: bool,
}

/// The blocks pushed for one subscription and not yet taken, and the cursor
/// past everything pushed — or the fault that ended the subscription.
type Parked = Result<(Vec<RawBlock>, DeltaCursor), ClientError>;

/// Who takes one train of a delta poll.
enum Taker {
    /// The poll's reader, off [`Polls::queue`].
    Reader,
    /// Whoever holds this subscription, off [`Polls::subs`].
    Holder,
    Nobody,
}

impl Polls {
    /// The taker of a train of the poll `poll` whose view is kept as `kept`.
    fn taker(&self, poll: Option<u64>, kept: u64) -> Taker {
        if poll == Some(self.live) {
            Taker::Reader
        } else if self.subs.contains_key(&kept) {
            Taker::Holder
        } else {
            Taker::Nobody
        }
    }

    /// One view of the poll `poll`, kept as `kept`, ended with `end` after
    /// the `blocks` held for a [`Taker::Holder`].
    fn end(&mut self, poll: Option<u64>, kept: u64, blocks: &mut Vec<RawBlock>, end: PollEnd) {
        // A cursor is what the server keeps a subscription from, whoever
        // takes this end.
        self.server_holds |= end.is_ok();
        let train = std::mem::take(blocks);
        match self.taker(poll, kept) {
            Taker::Reader => {
                // Held before the reader takes the end: a train pushed for
                // the id may be right behind this reply.
                if let Ok(cursor) = &end {
                    self.subs.insert(kept, Ok((Vec::new(), *cursor)));
                }
                self.queue.push_back(Polled::End(end));
            }
            Taker::Holder => match (self.subs.get_mut(&kept), end) {
                (Some(Ok((held, at))), Ok(cursor)) => {
                    held.extend(train);
                    *at = cursor;
                }
                (Some(parked @ Ok(_)), Err(why)) => *parked = Err(why),
                _ => {}
            },
            Taker::Nobody => {}
        }
    }

    /// Whether a subscription holds a train or an end nobody took yet.
    fn untaken(&self) -> bool {
        let quiet = |parked: &Parked| matches!(parked, Ok((blocks, _)) if blocks.is_empty());
        self.subs.values().any(|parked| !quiet(parked))
    }
}

/// A protocol session: the transport plus all per-connection protocol state
/// (the pending queue, and the continuation reassembly and status→error policy
/// that read/write it). Exactly one owner of that state: a
/// [`crate::GnitzClient`], whose host decides how it waits.
pub struct Session {
    transport: ClientTransport,
    pending: VecDeque<Slot>,
    submitted: u64,
    /// The error every request gets once the session has ended: what ended
    /// it. `None` while it is open.
    ended: Option<ClientError>,
    /// The last read filled its window, so more may be waiting.
    unread: bool,
    polls: Polls,
    /// The pushed train being read.
    pushed: Option<Slot>,
    /// The syncs sent and not yet answered, oldest first.
    syncing: VecDeque<Slot>,
}

impl Session {
    /// `target` is an AF_UNIX socket path or a `tls://` target (see
    /// `ClientTransport::connect`).
    pub fn connect(target: &str) -> Result<Self, ClientError> {
        Ok(Self::over(ClientTransport::connect(
            target,
            Instant::now() + CONNECT_TIMEOUT,
        )?))
    }

    /// A session over a connected transport.
    pub(crate) fn over(transport: ClientTransport) -> Self {
        Session {
            transport,
            pending: VecDeque::new(),
            submitted: 0,
            ended: None,
            unread: false,
            polls: Polls::default(),
            pushed: None,
            syncing: VecDeque::new(),
        }
    }

    /// Requests submitted since this session connected — one per `submit`,
    /// however the bytes were batched. HELLO predates the session, uncounted.
    pub fn requests_sent(&self) -> u64 {
        self.submitted
    }

    /// The connection's socket, as a number: valid while this session lives.
    pub fn as_raw_fd(&self) -> RawFd {
        self.transport.as_raw_fd()
    }

    /// The connection's socket, for a [`Host`](crate::Host) to register.
    pub fn as_fd(&self) -> BorrowedFd<'_> {
        self.transport.as_fd()
    }

    // ── Submitting and stepping ────────────────────────────────────────────

    /// Encode and queue `req`. A request this session refuses — one that
    /// does not validate, or at either cap — is answered with the refusal, as
    /// every submit's is.
    pub fn submit(&mut self, req: Request<'_>) -> Sent<u64> {
        let (frame, tid) = match req.encode() {
            Ok(encoded) => encoded,
            Err(why) => return Sent::ready(Err(why)),
        };
        let (to, sent) = promise();
        self.enqueue(frame, Slot::Ack { tid, to });
        sent
    }

    /// RESOLVE — describe the relation named by the canonical
    /// `"schema_name.relation_name"`; `None` when there is none.
    pub fn submit_resolve(&mut self, qname: &str) -> Sent<Option<Arc<RelDescriptor>>> {
        let hdr = ControlHeader::naming(ClientVerb::Resolve, Target::from(0), 0);
        let (to, sent) = promise();
        self.enqueue(encode_frame(hdr, qname.as_bytes(), None, None), Slot::Resolve { to });
        sent
    }

    /// SCAN_SPEC, replied in `reply_schema`'s layout.
    pub fn submit_scan(
        &mut self,
        target: Target,
        spec: &gnitz_wire::ReadSpec,
        reply_schema: &Arc<Schema>,
    ) -> Sent<ScanReply> {
        let hdr = ControlHeader::naming(ClientVerb::ScanSpec, target, reply_schema.layout().layout_digest());
        let (to, sent) = promise();
        let slot = Slot::Scan {
            tid: target.tid,
            reply_schema: Arc::clone(reply_schema),
            data: None,
            to,
        };
        self.enqueue(encode_frame(hdr, &spec.encode(), None, None), slot);
        sent
    }

    /// SCAN_MULTI: N [`Self::submit_scan`]s as one request, replied in order.
    pub fn submit_scan_multi(
        &mut self,
        scans: &[(Target, &gnitz_wire::ReadSpec, &Arc<Schema>)],
    ) -> Sent<Vec<ScanReply>> {
        if scans.is_empty() {
            return Sent::ready(Err(ClientError::from("a multi-read names no relation".to_string())));
        }
        let specs: Vec<Vec<u8>> = scans.iter().map(|(_, spec, _)| spec.encode()).collect();
        let items: Vec<txn_frame::ScanItem> = scans
            .iter()
            .zip(&specs)
            .map(|(&(target, _, schema), spec)| txn_frame::ScanItem {
                target,
                reply_layout: schema.layout().layout_digest(),
                spec,
            })
            .collect();
        let (to, sent) = promise();
        let slot = Slot::Multi {
            replies: Vec::with_capacity(scans.len()),
            rels: scans
                .iter()
                .map(|&(target, _, schema)| (target.tid, Arc::clone(schema)))
                .collect(),
            data: None,
            to,
        };
        self.enqueue(txn_frame::encode_scan_multi(&items), slot);
        sent
    }

    /// DELTA_POLL of one view after `from` — the view whole with none — under
    /// `spec`, replied in `reply_schema`'s layout. With `keep`, a view answered
    /// with a cursor stays subscribed from it under that id.
    pub fn submit_delta_read(
        &mut self,
        view: Target,
        from: Option<DeltaCursor>,
        reply_schema: &Arc<Schema>,
        spec: &[u8],
        keep: Option<u64>,
    ) -> Sent<(ScanReply, DeltaCursor)> {
        let (to, sent) = promise();
        let slot = Slot::Delta {
            tid: view.tid,
            reply_schema: Arc::clone(reply_schema),
            data: None,
            keep,
            to,
        };
        let item = txn_frame::DeltaPollItem {
            view,
            from,
            reply_layout: reply_schema.layout().layout_digest(),
            spec,
        };
        self.enqueue(txn_frame::encode_delta_poll(&[item], keep), slot);
        sent
    }

    /// SYNC_PUSHED: bring the subscriptions `held` names up to date, held up
    /// to `wait` while none has anything new, and end every other one of this
    /// connection.
    pub(crate) fn submit_sync(&mut self, held: &[u64], wait: Duration) -> Sent<()> {
        self.polls.subs.retain(|id, _| held.contains(id));
        if !self.polls.server_holds {
            return Sent::ready(Ok(()));
        }
        // An unanswered sync may bring trains nobody is left to take.
        let wait = if self.polls.untaken() || !self.syncing.is_empty() {
            Duration::ZERO
        } else {
            wait
        };
        let hdr = ControlHeader::naming(ClientVerb::SyncPushed, Target::from(0), wait_ms(wait));
        let named = txn_frame::encode_held(self.polls.subs.keys().copied());
        let (to, sent) = promise();
        if self.enqueue(encode_frame(hdr, &named, None, None), Slot::Sync { to }) {
            self.polls.server_holds = !self.polls.subs.is_empty();
        }
        sent
    }

    /// What was pushed for subscription `id` since it was last taken: the
    /// blocks in arrival order, and the cursor past everything pushed.
    ///
    /// `Err` is the end of the subscription, which the session then holds no
    /// more: the server ended it, or this connection never held it.
    pub(crate) fn take_pushed(&mut self, id: u64) -> Result<(Vec<RawBlock>, DeltaCursor), ClientError> {
        let Entry::Occupied(mut parked) = self.polls.subs.entry(id) else {
            return Err(ClientError::from(format!(
                "subscription {id} is not held on this connection; read from its cursor and subscribe again"
            )));
        };
        match parked.get_mut() {
            Ok((blocks, cursor)) => Ok((std::mem::take(blocks), *cursor)),
            Err(_) => parked.remove(),
        }
    }

    /// DELTA_POLL of `views`, the items of the live poll: each one's train
    /// queues for [`Self::next_polled`], in order, and a request the session
    /// refuses ends each with the refusal. A view answered with a cursor stays
    /// subscribed from it, item `i` as [`kept_as`]`(first, i)`.
    pub(crate) fn submit_delta_poll(&mut self, first: u64, views: &[txn_frame::DeltaPollItem]) {
        if views.is_empty() {
            return;
        }
        let slot = Slot::DeltaPoll {
            views: views.iter().map(|v| v.view.tid).collect(),
            at: 0,
            poll: Some(self.polls.live),
            first,
            blocks: Vec::new(),
        };
        self.enqueue(txn_frame::encode_delta_poll(views, Some(first)), slot);
    }

    /// Queue `frame`, whose reply `slot` reads; whether it was queued. A
    /// request this session refuses is answered with the refusal.
    fn enqueue(&mut self, frame: Vec<u8>, mut slot: Slot) -> bool {
        let (total, limit) = (frame.len(), gnitz_wire::MAX_FRAME_PAYLOAD);
        // A frame past the ceiling is refused here rather than by the server's
        // ingress cap, which would drop the connection.
        let refusal = if total > limit {
            Some(ClientError::from(format!(
                "request frame is {total} bytes, exceeding the {limit}-byte server ingress cap; \
                 split the request"
            )))
        } else if let Some(why) = &self.ended {
            Some(why.clone())
        } else {
            // The predicate a driver reads for back-pressure, so the two can
            // never disagree; only the message re-derives which cap was hit.
            self.at_capacity().then(|| {
                let queued = self.queued_bytes();
                ClientError::from(if self.in_flight() >= MAX_IN_FLIGHT {
                    format!("connection has {MAX_IN_FLIGHT} requests in flight")
                } else {
                    format!("connection has {queued} unwritten bytes queued, at the {MAX_QUEUED_BYTES}-byte cap")
                })
            })
        };
        if let Some(why) = refusal {
            slot.fail(why, &mut self.polls);
            return false;
        }
        self.transport.enqueue(frame);
        self.submitted += 1;
        match slot {
            Slot::Sync { .. } => self.syncing.push_back(slot),
            _ => self.pending.push_back(slot),
        }
        true
    }

    /// Requests sent and not yet answered.
    fn in_flight(&self) -> usize {
        self.pending.len() + self.syncing.len()
    }

    /// Do the I/O `ready` allows — one read, then one flush; every reply it
    /// completes arrives. Afterwards a driver may park on `interest()` unless
    /// [`Self::unread`]: nothing buffered can advance, and bytes still queued
    /// are ones the fd refused.
    ///
    /// A failure ends the session: every reply still owed arrives as
    /// [`ClientError::ConnectionLost`], after those that completed.
    pub fn step(&mut self, ready: Interest) {
        if ready.read {
            self.read();
        }
        // Last, so the ciphertext a read queues goes out with this flush.
        if ready.write && self.ended.is_none() {
            if let Err(e) = self.transport.flush() {
                // A peer gone after answering is still readable.
                self.read();
                while self.unread {
                    self.read();
                }
                self.end(ClientError::ConnectionLost(e));
            }
        }
    }

    /// One read of the socket, each frame it completes fed to the head slot. A
    /// failure ends the session.
    fn read(&mut self) {
        self.unread = false;
        if self.ended.is_some() {
            return;
        }
        let Session {
            transport,
            pending,
            polls,
            pushed,
            syncing,
            ..
        } = self;
        match transport.read(|buf| feed(pending, pushed, syncing, buf, polls)) {
            Ok(more) => self.unread = more,
            Err(e) => self.end(ClientError::ConnectionLost(e)),
        }
    }

    /// The live delta poll's next result.
    pub(crate) fn next_polled(&mut self) -> Option<Polled> {
        self.polls.queue.pop_front()
    }

    /// Everything the last step queued has been handed out.
    pub(crate) fn polled_out(&self) -> bool {
        self.polls.queue.is_empty()
    }

    /// The live poll's reader is gone: drop what it was handed and what its
    /// slots are still to be fed.
    pub(crate) fn abandon_poll(&mut self) {
        self.polls.live += 1;
        self.polls.queue.clear();
    }

    /// The last step left a read unfinished: step `READ` again without waiting
    /// for the socket, which has no new edge to report.
    pub(crate) fn unread(&self) -> bool {
        self.unread
    }

    /// Frame bytes queued and not yet written, against [`MAX_QUEUED_BYTES`].
    fn queued_bytes(&self) -> usize {
        self.transport.queued_bytes()
    }

    /// Either of the caps [`Self::submit`] refuses on — in-flight slots or
    /// queued bytes — reached. A driver that can wait leaves the request where
    /// it is, which is what turns a cap into back-pressure rather than an error.
    /// Each cap implies its own interest bit, so gating on it loses no wakeup.
    pub fn at_capacity(&self) -> bool {
        self.in_flight() >= MAX_IN_FLIGHT || self.queued_bytes() >= MAX_QUEUED_BYTES
    }

    /// `READ` while any request is unanswered; `WRITE` while bytes remain queued
    /// or rustls has ciphertext to ship. Nothing once the session has ended.
    pub fn interest(&self) -> Interest {
        if self.ended.is_some() {
            return Interest::NONE;
        }
        Interest {
            read: self.in_flight() > 0,
            write: self.transport.wants_write(),
        }
    }

    /// Whether this session has ended, and refuses work.
    pub fn is_closed(&self) -> bool {
        self.ended.is_some()
    }

    /// Fail every unanswered request with `why` and refuse further work. The shutdown
    /// shows the server EOF now rather than when the session drops.
    pub(crate) fn end(&mut self, why: ClientError) {
        if self.ended.is_some() {
            return;
        }
        self.transport.close();
        let Session { pending, syncing, polls, .. } = self;
        for mut slot in pending.drain(..).chain(syncing.drain(..)) {
            slot.fail(why.clone(), polls);
        }
        self.ended = Some(why);
    }
}

/// `wait` in the wire's milliseconds, rounded up: a wait shorter than one is
/// still a wait.
fn wait_ms(wait: Duration) -> u64 {
    u64::try_from(wait.as_nanos().div_ceil(1_000_000)).unwrap_or(u64::MAX)
}

/// Feed one frame to the slot it is of, whose reply arrives if that completes it.
fn feed(
    pending: &mut VecDeque<Slot>,
    pushed: &mut Option<Slot>,
    syncing: &mut VecDeque<Slot>,
    buf: Cow<'_, [u8]>,
    polls: &mut Polls,
) -> Result<(), ProtocolError> {
    let ctrl = peek_control_block(&buf).map_err(ProtocolError::DecodeError)?;
    let in_train = pushed.is_some();
    let queue = match ctrl.hdr.flags.lane {
        WireLane::Reply => pending,
        _ if in_train => {
            return Err(ProtocolError::DecodeError(
                "a pushed train split by a frame that is not of it".into(),
            ))
        }
        WireLane::PushedTrain => {
            *pushed = Some(Slot::DeltaPoll {
                views: vec![ctrl.hdr.target_id],
                at: 0,
                poll: None,
                first: ctrl.hdr.arg0,
                blocks: Vec::new(),
            });
            return Ok(());
        }
        WireLane::SyncAnswer => syncing,
    };
    let Some(slot) = pushed.as_mut().or(queue.front_mut()) else {
        return Err(ProtocolError::DecodeError("reply frame with no request pending".into()));
    };
    if slot.feed(ctrl, buf, polls)? {
        match in_train {
            true => *pushed = None,
            false => drop(queue.pop_front()),
        }
    }
    Ok(())
}

impl Slot {
    /// Take one reply frame; whether that answered the slot, with its reply
    /// or with the refusal this frame is. `Err` is a frame this slot cannot
    /// have been sent, which ends the session.
    fn feed(&mut self, ctrl: DecodedControl, mut buf: Cow<'_, [u8]>, polls: &mut Polls) -> Result<bool, ProtocolError> {
        let named = ctrl.hdr.target_id;
        // The relation this frame must name. Replies arrive in request order,
        // so a frame naming another would decode under the wrong schema
        // silently; make it loud.
        let want = match &*self {
            Slot::Ack { tid, .. } | Slot::Scan { tid, .. } | Slot::Delta { tid, .. } => Some(*tid),
            Slot::Sync { .. } => Some(0),
            Slot::Multi { rels, replies, .. } => Some(rels[replies.len()].0),
            Slot::DeltaPoll { views, at, .. } => Some(views[*at]),
            Slot::Resolve { .. } => None,
        };
        if let Some(fault) = ctrl.fault(&buf) {
            let refused = ClientError::Refused(fault);
            // A DELTA_POLL fault naming a view ends that view's position alone;
            // every other fault ends the request.
            return match (&mut *self, want) {
                (Slot::DeltaPoll { views, at, poll, first, blocks }, Some(view)) if named != 0 => {
                    if named != view {
                        return Err(out_of_order(view, named));
                    }
                    polls.end(*poll, kept_as(*first, *at), blocks, Err(refused));
                    *at += 1;
                    Ok(*at == views.len())
                }
                _ => {
                    self.fail(refused, polls);
                    Ok(true)
                }
            };
        }
        if let Some(want) = want.filter(|&want| want != named) {
            return Err(out_of_order(want, named));
        }
        // Only a RESOLVE is answered in the server's schema; every read decodes
        // under the schema its request named.
        if ctrl.schema.is_some() && !matches!(self, Slot::Resolve { .. }) {
            return Err(ProtocolError::DecodeError(
                "a schema block on a reply whose request named its schema".into(),
            ));
        }

        // Rows are decoded straight into the slot: a train carries one data
        // frame per worker, and a per-frame batch would be copied in and dropped.
        let decode = |data: &mut Option<ZSetBatch>, schema: &Arc<Schema>, block: &[u8]| {
            decode_wal_block_into(data.get_or_insert_with(|| ZSetBatch::new(schema)), block, schema)
        };
        match (&mut *self, ctrl.data.clone()) {
            (_, None) => {}
            (Slot::DeltaPoll { poll, first, at, blocks, .. }, Some(block)) => {
                let mut raw = || RawBlock {
                    frame: std::mem::take(&mut buf).into_owned(),
                    block: block.clone(),
                };
                match polls.taker(*poll, kept_as(*first, *at)) {
                    Taker::Reader => polls.queue.push_back(Polled::Block(raw())),
                    Taker::Holder => blocks.push(raw()),
                    Taker::Nobody => {}
                }
            }
            (Slot::Scan { reply_schema, data, .. } | Slot::Delta { reply_schema, data, .. }, Some(r)) => {
                decode(data, reply_schema, &buf[r])?
            }
            (Slot::Multi { rels, replies, data, .. }, Some(r)) => decode(data, &rels[replies.len()].1, &buf[r])?,
            (Slot::Ack { .. } | Slot::Resolve { .. } | Slot::Sync { .. }, Some(_)) => {
                return Err(ProtocolError::DecodeError(
                    "a data block on a reply that carries no rows".into(),
                ))
            }
        }

        if ctrl.hdr.flags.continuation {
            return Ok(false);
        }

        // The train terminated.
        let scan_reply = |schema: &Arc<Schema>, data: &mut Option<ZSetBatch>| ScanReply {
            batch: data.take().unwrap_or_else(|| ZSetBatch::new(schema)),
            schema: Arc::clone(schema),
            lsn: Some(ctrl.hdr.arg0),
        };
        let cursor = || {
            DeltaCursor::from_pair(ctrl.hdr.arg1, ctrl.hdr.arg0)
                .ok_or_else(|| ProtocolError::DecodeError("a delta-poll terminal at round 0".into()))
        };
        match self {
            Slot::Ack { to, .. } => to.fulfil(Ok(ctrl.hdr.arg0)),
            Slot::Resolve { to } => to.fulfil(Ok(resolve_descriptor(&ctrl, &buf)?)),
            Slot::Scan { reply_schema, data, to, .. } => to.fulfil(Ok(scan_reply(reply_schema, data))),
            Slot::Delta { reply_schema, data, keep, to, .. } => {
                let cursor = cursor()?;
                // Held before the reader takes the reply: a train pushed for
                // the id may be right behind it.
                if let Some(id) = *keep {
                    polls.server_holds = true;
                    polls.subs.insert(id, Ok((Vec::new(), cursor)));
                }
                let reply = ScanReply {
                    lsn: None,
                    ..scan_reply(reply_schema, data)
                };
                to.fulfil(Ok((reply, cursor)))
            }
            Slot::Multi { rels, replies, data, to } => {
                let reply = scan_reply(&rels[replies.len()].1, data);
                replies.push(reply);
                if replies.len() < rels.len() {
                    return Ok(false);
                }
                to.fulfil(Ok(std::mem::take(replies)))
            }
            Slot::DeltaPoll { views, at, poll, first, blocks } => {
                polls.end(*poll, kept_as(*first, *at), blocks, Ok(cursor()?));
                *at += 1;
                if *at < views.len() {
                    return Ok(false);
                }
            }
            Slot::Sync { to } => to.fulfil(Ok(())),
        }
        Ok(true)
    }

    /// The request ended unanswered: `why` is its reply, and the end of each
    /// view a delta poll had yet to answer.
    fn fail(&mut self, why: ClientError, polls: &mut Polls) {
        match self {
            Slot::Ack { to, .. } => to.fulfil(Err(why)),
            Slot::Resolve { to } => to.fulfil(Err(why)),
            Slot::Scan { to, .. } => to.fulfil(Err(why)),
            Slot::Delta { to, .. } => to.fulfil(Err(why)),
            Slot::Multi { to, .. } => to.fulfil(Err(why)),
            Slot::DeltaPoll { views, at, poll, first, blocks } => {
                for at in *at..views.len() {
                    polls.end(*poll, kept_as(*first, at), blocks, Err(why.clone()));
                }
            }
            Slot::Sync { to } => to.fulfil(Err(why)),
        }
    }
}

/// A reply frame naming a relation the head slot's position does not expect.
fn out_of_order(want: u64, got: u64) -> ProtocolError {
    ProtocolError::DecodeError(format!("reply out of order: expected target {want}, got {got}"))
}

/// A RESOLVE train as its descriptor; `None` when the reply names no relation.
fn resolve_descriptor(ctrl: &DecodedControl, frame: &[u8]) -> Result<Option<Arc<RelDescriptor>>, ProtocolError> {
    if ctrl.hdr.target_id == 0 {
        return Ok(None);
    }
    let schema = ctrl
        .schema
        .clone()
        .ok_or_else(|| ProtocolError::DecodeError("RESOLVE reply carries no schema".into()))?;
    let schema = Arc::new(Schema::from_block(&frame[schema]).map_err(ProtocolError::DecodeError)?);
    let desc = RelDescriptorBlob::decode(&frame[ctrl.blob.clone()]).map_err(ProtocolError::DecodeError)?;
    Ok(Some(Arc::new(RelDescriptor {
        tid: ctrl.hdr.target_id,
        class: desc.class,
        pk_repeats: desc.pk_repeats,
        serial: desc.serial,
        schema,
        indexes: desc.indexes,
        token: ctrl.hdr.arg0,
    })))
}

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

#[cfg(test)]
#[path = "benches/connection.rs"]
mod bench;