questdb/

ingress.rs

/*******************************************************************************
 *     ___                  _   ____  ____
 *    / _ \ _   _  ___  ___| |_|  _ \| __ )
 *   | | | | | | |/ _ \/ __| __| | | |  _ \
 *   | |_| | |_| |  __/\__ \ |_| |_| | |_) |
 *    \__\_\\__,_|\___||___/\__|____/|____/
 *
 *  Copyright (c) 2014-2019 Appsicle
 *  Copyright (c) 2019-2025 QuestDB
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *  http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 *
 ******************************************************************************/

#![cfg_attr(feature = "_sync-sender", doc = include_str!("ingress/mod.md"))]
#![cfg_attr(
    not(feature = "_sync-sender"),
    doc = "Shared data types used by the egress reader. Enable a `sync-sender-*` \
           feature to expose the sender APIs and their full module documentation."
)]

#[cfg(feature = "_sender-qwp-ws")]
pub(crate) use self::conf::QwpWsManagedSlotExclusion;
pub use self::ndarr::{ArrayElement, NdArrayView};
pub use self::timestamp::*;
use crate::error::Result;
#[cfg(feature = "_sync-sender")]
use crate::error::{self, fmt};
#[cfg(feature = "_sync-sender")]
use crate::ingress::conf::ConfigSetting;
#[cfg(feature = "_sync-sender")]
use core::time::Duration;
#[cfg(feature = "_sync-sender")]
use std::collections::HashMap;
#[cfg(feature = "_sender-qwp-ws")]
use std::collections::HashSet;
#[cfg(feature = "_sync-sender")]
use std::fmt::Write;
use std::fmt::{Debug, Display, Formatter};

#[cfg(feature = "_sync-sender")]
use std::ops::Deref;
#[cfg(feature = "_sender-qwp-ws")]
use std::path::Path;
#[cfg(feature = "_sync-sender")]
use std::path::PathBuf;
#[cfg(feature = "_sync-sender")]
use std::str::FromStr;

#[cfg(feature = "_sync-sender")]
mod tls;

#[cfg(all(feature = "_sender-tcp", feature = "aws-lc-crypto"))]
use aws_lc_rs::signature::{ECDSA_P256_SHA256_FIXED_SIGNING, EcdsaKeyPair};

#[cfg(all(feature = "_sender-tcp", feature = "ring-crypto"))]
use ring::{
    rand::SystemRandom,
    signature::{ECDSA_P256_SHA256_FIXED_SIGNING, EcdsaKeyPair},
};

#[cfg(feature = "_sync-sender")]
mod conf;

#[cfg(feature = "_sender-qwp-ws")]
pub mod conn_events;
#[cfg(feature = "_sender-qwp-ws")]
pub use conn_events::{
    ConnectionEvent, ConnectionEventDispatcher, ConnectionEventKind, ConnectionListener,
};

#[cfg(feature = "_sender-qwp-ws")]
pub(crate) mod rejection_events;

pub(crate) mod ndarr;

mod timestamp;

mod buffer;
pub use buffer::*;

#[cfg(feature = "_sync-sender")]
pub(crate) mod sender;
#[cfg(feature = "_sender-qwp-ws")]
pub(crate) use sender::QwpWsRoleReject;
#[cfg(feature = "polars-ingress")]
pub(crate) use sender::ReconnectPolicy;
#[cfg(feature = "sync-sender-qwp-ws")]
pub(crate) use sender::ReconnectReason;
#[cfg(feature = "_sync-sender")]
pub use sender::*;
#[cfg(feature = "sync-sender-qwp-ws")]
pub(crate) use sender::{reconnect_backoff_step, reconnect_error_is_terminal};

mod decimal;
pub use decimal::DecimalView;

#[cfg(feature = "sync-sender-qwp-ws")]
pub mod column_sender;

/// Acknowledgement level shared by the column-major and row-major QWP/WebSocket
/// senders' `wait` / `sync` APIs.
#[cfg(feature = "sync-sender-qwp-ws")]
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
#[non_exhaustive]
pub enum AckLevel {
    /// Wait for the server to accept every published frame.
    #[default]
    Ok,
    /// Wait for durable-ACK coverage. This level requires QuestDB Enterprise
    /// and the `request_durable_ack=on` connection-string setting.
    Durable,
}

/// Precision of a timestamp column, selecting the QWP wire type used by
/// [`Chunk::column_ts`](crate::ingress::column_sender::Chunk::column_ts):
/// [`TimestampUnit::Micros`] maps to `TIMESTAMP` and [`TimestampUnit::Nanos`]
/// to `TIMESTAMP_NANOS`. Column values are Unix-epoch integers in the chosen
/// unit.
#[cfg(feature = "sync-sender-qwp-ws")]
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum TimestampUnit {
    /// Microseconds since the Unix epoch (QWP wire type `TIMESTAMP`).
    Micros,
    /// Nanoseconds since the Unix epoch (QWP wire type `TIMESTAMP_NANOS`).
    Nanos,
}

#[cfg(feature = "polars-ingress")]
pub mod polars;

const MAX_NAME_LEN_DEFAULT: usize = 127;

/// The maximum allowed dimensions for arrays.
pub const MAX_ARRAY_DIMS: usize = 32;
pub const MAX_ARRAY_BUFFER_SIZE: usize = 512 * 1024 * 1024; // 512MiB
pub const MAX_ARRAY_DIM_LEN: usize = 0x0FFF_FFFF; // 1 << 28 - 1

/// Maximum element count of a single ndarray row payload (`prod(shape)`).
/// Bounds the per-row reservation (`leaf_count * 8` bytes) well below
/// `isize::MAX` so allocator-OOM cannot abort the host under
/// `panic = "abort"`. Enforced on both the FFI and pure-Rust entry
/// points to keep the contract uniform across API surfaces.
pub const MAX_NDARRAY_LEAF_ELEMS: usize = 1 << 24;

pub(crate) const ARRAY_BINARY_FORMAT_TYPE: u8 = 14;
pub(crate) const DOUBLE_BINARY_FORMAT_TYPE: u8 = 16;
pub const DECIMAL_BINARY_FORMAT_TYPE: u8 = 23;

/// Transport-scoped protocol version identifier used by the ingestion APIs.
///
/// Interpret this value together with the transport protocol.
/// The same version number may correspond to different wire formats or feature
/// sets on different transports.
#[derive(Debug, Copy, Clone, PartialEq, PartialOrd)]
pub enum ProtocolVersion {
    /// Version 1.
    V1 = 1,

    /// Version 2.
    V2 = 2,

    /// Version 3.
    V3 = 3,
}

/// List of supported ILP protocol versions, in order of preference (highest to lowest).
#[cfg(feature = "_sender-http")]
const SUPPORTED_PROTOCOL_VERSIONS: [ProtocolVersion; 3] = [
    ProtocolVersion::V3,
    ProtocolVersion::V2,
    ProtocolVersion::V1,
];

impl Display for ProtocolVersion {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        match self {
            ProtocolVersion::V1 => write!(f, "v1"),
            ProtocolVersion::V2 => write!(f, "v2"),
            ProtocolVersion::V3 => write!(f, "v3"),
        }
    }
}

#[cfg(feature = "_sender-tcp")]
fn map_io_to_socket_err(prefix: &str, io_err: std::io::Error) -> error::Error {
    fmt!(SocketError, "{}{}", prefix, io_err)
}

/// Possible sources of the root certificates used to validate the server's TLS
/// certificate.
#[derive(PartialEq, Debug, Clone, Copy)]
pub enum CertificateAuthority {
    /// Use the root certificates provided by the
    /// [`webpki-roots`](https://crates.io/crates/webpki-roots) crate.
    #[cfg(feature = "tls-webpki-certs")]
    WebpkiRoots,

    /// Use the root certificates provided by the OS
    #[cfg(feature = "tls-native-certs")]
    OsRoots,

    /// Combine the root certificates provided by the OS and the `webpki-roots` crate.
    #[cfg(all(feature = "tls-webpki-certs", feature = "tls-native-certs"))]
    WebpkiAndOsRoots,

    /// Use the root certificates provided in a PEM-encoded file.
    PemFile,
}

/// A `u16` port number or `String` port service name as is registered with
/// `/etc/services` or equivalent.
///
/// ```
/// use questdb::ingress::Port;
/// use std::convert::Into;
///
/// let service: Port = 9009.into();
/// ```
///
/// or
///
/// ```
/// use questdb::ingress::Port;
/// use std::convert::Into;
///
/// // Assuming the service name is registered.
/// let service: Port = "qdb_ilp".into();  // or with a String too.
/// ```
#[cfg(feature = "_sync-sender")]
pub struct Port(String);

#[cfg(feature = "_sync-sender")]
impl From<String> for Port {
    fn from(s: String) -> Self {
        Port(s)
    }
}

#[cfg(feature = "_sync-sender")]
impl From<&str> for Port {
    fn from(s: &str) -> Self {
        Port(s.to_owned())
    }
}

#[cfg(feature = "_sync-sender")]
impl From<u16> for Port {
    fn from(p: u16) -> Self {
        Port(p.to_string())
    }
}

#[cfg(feature = "_sync-sender")]
fn validate_auto_flush_params(params: &HashMap<String, String>) -> Result<()> {
    if let Some(auto_flush) = params.get("auto_flush")
        && auto_flush.as_str() != "off"
    {
        return Err(error::fmt!(
            ConfigError,
            "Invalid auto_flush value '{auto_flush}'. This client does not \
            support auto-flush, so the only accepted value is 'off'"
        ));
    }

    for &param in ["auto_flush_rows", "auto_flush_bytes", "auto_flush_interval"].iter() {
        if params.contains_key(param) {
            return Err(error::fmt!(
                ConfigError,
                "Invalid configuration parameter {:?}. This client does not support auto-flush",
                param
            ));
        }
    }
    Ok(())
}

/// Protocol used to communicate with the QuestDB server.
///
/// `#[non_exhaustive]` so new wire protocols can be added without breaking
/// exhaustive matches in downstream code (the surface already covers ILP/TCP,
/// ILP/HTTP, QWP/UDP, and QWP/WS, and is expected to grow).
#[derive(PartialEq, Debug, Clone, Copy)]
#[non_exhaustive]
#[cfg(feature = "_sync-sender")]
pub enum Protocol {
    #[cfg(feature = "_sender-tcp")]
    /// ILP over TCP (streaming).
    Tcp,

    #[cfg(feature = "_sender-tcp")]
    /// TCP + TLS
    Tcps,

    #[cfg(feature = "_sender-http")]
    /// ILP over HTTP (request-response)
    /// Version 1 is compatible with the InfluxDB Line Protocol.
    Http,

    #[cfg(feature = "_sender-http")]
    /// HTTP + TLS
    Https,

    #[cfg(feature = "_sender-qwp-udp")]
    /// Quest Wire Protocol over UDP datagrams (IPv4-only).
    Udp,

    #[cfg(feature = "_sender-qwp-ws")]
    /// Quest Wire Protocol over WebSocket (RFC 6455).
    Ws,

    #[cfg(feature = "_sender-qwp-ws")]
    /// Quest Wire Protocol over WebSocket Secure (TLS).
    Wss,
}

#[cfg(feature = "_sync-sender")]
impl Display for Protocol {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::result::Result<(), std::fmt::Error> {
        f.write_str(self.schema())
    }
}

#[cfg(feature = "_sync-sender")]
impl Protocol {
    fn default_port(&self) -> &str {
        match *self {
            #[cfg(feature = "_sender-tcp")]
            Protocol::Tcp | Protocol::Tcps => "9009",
            #[cfg(feature = "_sender-http")]
            Protocol::Http | Protocol::Https => "9000",
            #[cfg(feature = "_sender-qwp-udp")]
            Protocol::Udp => "9007",
            #[cfg(feature = "_sender-qwp-ws")]
            Protocol::Ws | Protocol::Wss => "9000",
        }
    }

    fn tls_enabled(&self) -> bool {
        match *self {
            #[cfg(feature = "_sender-tcp")]
            Protocol::Tcp => false,
            #[cfg(feature = "_sender-tcp")]
            Protocol::Tcps => true,
            #[cfg(feature = "_sender-http")]
            Protocol::Http => false,
            #[cfg(feature = "_sender-http")]
            Protocol::Https => true,
            #[cfg(feature = "_sender-qwp-udp")]
            Protocol::Udp => false,
            #[cfg(feature = "_sender-qwp-ws")]
            Protocol::Ws => false,
            #[cfg(feature = "_sender-qwp-ws")]
            Protocol::Wss => true,
        }
    }

    #[cfg(feature = "_sender-tcp")]
    fn is_tcpx(&self) -> bool {
        match self {
            Protocol::Tcp | Protocol::Tcps => true,
            #[cfg(feature = "_sender-http")]
            Protocol::Http | Protocol::Https => false,
            #[cfg(feature = "_sender-qwp-udp")]
            Protocol::Udp => false,
            #[cfg(feature = "_sender-qwp-ws")]
            Protocol::Ws | Protocol::Wss => false,
        }
    }

    #[cfg(feature = "_sender-http")]
    fn is_httpx(&self) -> bool {
        match self {
            #[cfg(feature = "_sender-tcp")]
            Protocol::Tcp | Protocol::Tcps => false,
            Protocol::Http | Protocol::Https => true,
            #[cfg(feature = "_sender-qwp-udp")]
            Protocol::Udp => false,
            #[cfg(feature = "_sender-qwp-ws")]
            Protocol::Ws | Protocol::Wss => false,
        }
    }

    #[cfg(feature = "_sender-qwp-udp")]
    fn is_qwp_udp(&self) -> bool {
        matches!(self, Protocol::Udp)
    }

    #[cfg(feature = "_sender-qwp-ws")]
    fn is_qwp_ws(&self) -> bool {
        matches!(self, Protocol::Ws | Protocol::Wss)
    }

    /// True if the protocol authenticates via HTTP-style headers
    /// (basic / bearer-token), i.e. ILP/HTTP or QWP/WebSocket.
    #[cfg(any(feature = "_sender-http", feature = "_sender-qwp-ws"))]
    fn accepts_http_auth(&self) -> bool {
        let mut accepts = false;
        #[cfg(feature = "_sender-http")]
        if self.is_httpx() {
            accepts = true;
        }
        #[cfg(feature = "_sender-qwp-ws")]
        if self.is_qwp_ws() {
            accepts = true;
        }
        accepts
    }

    fn schema(&self) -> &str {
        match *self {
            #[cfg(feature = "_sender-tcp")]
            Protocol::Tcp => "tcp",
            #[cfg(feature = "_sender-tcp")]
            Protocol::Tcps => "tcps",
            #[cfg(feature = "_sender-http")]
            Protocol::Http => "http",
            #[cfg(feature = "_sender-http")]
            Protocol::Https => "https",
            #[cfg(feature = "_sender-qwp-udp")]
            Protocol::Udp => "udp",
            #[cfg(feature = "_sender-qwp-ws")]
            Protocol::Ws => "ws",
            #[cfg(feature = "_sender-qwp-ws")]
            Protocol::Wss => "wss",
        }
    }

    fn from_schema(schema: &str) -> Result<Self> {
        #[cfg(feature = "_sender-tcp")]
        if schema.eq_ignore_ascii_case("tcp") {
            return Ok(Protocol::Tcp);
        }
        #[cfg(feature = "_sender-tcp")]
        if schema.eq_ignore_ascii_case("tcps") {
            return Ok(Protocol::Tcps);
        }
        #[cfg(feature = "_sender-http")]
        if schema.eq_ignore_ascii_case("http") {
            return Ok(Protocol::Http);
        }
        #[cfg(feature = "_sender-http")]
        if schema.eq_ignore_ascii_case("https") {
            return Ok(Protocol::Https);
        }
        #[cfg(feature = "_sender-qwp-udp")]
        if schema.eq_ignore_ascii_case("udp") {
            return Ok(Protocol::Udp);
        }
        #[cfg(feature = "_sender-qwp-udp")]
        if schema.eq_ignore_ascii_case("udps") {
            return Err(error::fmt!(ConfigError, "TLS is not supported for UDP."));
        }
        #[cfg(feature = "_sender-qwp-ws")]
        if schema.eq_ignore_ascii_case("ws") {
            return Ok(Protocol::Ws);
        }
        #[cfg(feature = "_sender-qwp-ws")]
        if schema.eq_ignore_ascii_case("wss") {
            return Ok(Protocol::Wss);
        }
        Err(error::fmt!(ConfigError, "Unsupported protocol: {}", schema))
    }
}

#[cfg(any(feature = "_sender-qwp-ws", feature = "_egress"))]
pub(crate) struct QwpWsAddrScan {
    pub(crate) addr_values: Vec<String>,
    pub(crate) sanitized_conf: String,
}

/// Resolved, reusable ingredients for opening QWP/WebSocket connections to a
/// rotating set of endpoints. Built once by
/// [`SenderBuilder::build_qwp_ws_connector`] and held by the ingestion pool,
/// which drives it through a
/// shared [`sender::qwp_ws::QwpWsHostHealthTracker`] so every borrow lands on
/// a live, writable endpoint (skipping unhealthy / role-rejecting ones)
/// without re-parsing the connect string.
#[cfg(feature = "sync-sender-qwp-ws")]
pub(crate) struct QwpWsConnector {
    host: String,
    port: String,
    endpoints: std::sync::Arc<[conf::QwpWsEndpoint]>,
    use_tls: bool,
    tls_settings: Option<tls::TlsSettings>,
    qwp_ws: conf::QwpWsConfig,
    auth_header: Option<String>,
    max_buf_size: usize,
}

#[cfg(feature = "sync-sender-qwp-ws")]
impl QwpWsConnector {
    /// Number of configured endpoints — the pool sizes its shared health
    /// tracker to this.
    pub(crate) fn endpoint_count(&self) -> usize {
        self.endpoints.len()
    }

    /// Endpoint at `idx`, for event narration. `None` when out of range.
    pub(crate) fn endpoint(&self, idx: usize) -> Option<&conf::QwpWsEndpoint> {
        self.endpoints.get(idx)
    }

    pub(crate) fn max_buf_size(&self) -> usize {
        self.max_buf_size
    }

    pub(crate) fn request_durable_ack(&self) -> bool {
        *self.qwp_ws.request_durable_ack
    }

    pub(crate) fn sender_id(&self) -> &str {
        self.qwp_ws.sender_id.as_str()
    }

    pub(crate) fn sf_dir(&self) -> Option<&Path> {
        self.qwp_ws.sf_dir.as_deref()
    }

    /// Per-call request timeout parsed from the connect string. The direct
    /// column backend arms this as the socket read/write timeout; the
    /// store-and-forward backend uses it as the no-progress deadline in its
    /// `sync` poll loop so a silent-but-alive peer cannot block the caller
    /// forever.
    pub(crate) fn request_timeout(&self) -> Duration {
        *self.qwp_ws.request_timeout
    }

    /// Bound on how long the store-and-forward ingestion pool waits for a
    /// connection's background runner to deliver its queued frames before the
    /// connection is dropped (pool close / shutdown return). `Duration::ZERO`
    /// disables the wait. Parsed from `close_flush_timeout_millis`.
    pub(crate) fn close_flush_timeout(&self) -> Duration {
        *self.qwp_ws.close_flush_timeout
    }

    /// Reconnect backoff budget parsed from the connect string's
    /// `reconnect_*` keys. Only the retry-capable borrow paths consume it
    /// (the polars `reborrow_with_retry` and the FFI owned `*_with_retry`
    /// entry points); keep it compiled (so the `ReconnectPolicy` re-export it
    /// returns stays live) but quiet the dead-code lint when neither is built.
    #[cfg(feature = "sync-sender-qwp-ws")]
    #[cfg_attr(
        not(any(
            feature = "polars-ingress",
            feature = "polars-egress",
            feature = "ffi-support"
        )),
        allow(dead_code)
    )]
    pub(crate) fn reconnect_policy(&self) -> sender::ReconnectPolicy {
        sender::ReconnectPolicy::bounded(
            *self.qwp_ws.reconnect_max_duration,
            *self.qwp_ws.reconnect_initial_backoff,
            *self.qwp_ws.reconnect_max_backoff,
        )
    }

    /// Pool path: drive the connect round against the *shared* health tracker
    /// behind `health`, locking it only per tracker operation. The health lock
    /// is **never** held across
    /// the blocking TCP/TLS/WS-upgrade handshake, so concurrent cold-start
    /// borrows no longer serialize end-to-end and dead-sender returns that
    /// need the same lock are not stalled behind one slow / black-holed
    /// connect.
    pub(crate) fn connect_round_pooled(
        &self,
        health: &std::sync::Mutex<sender::qwp_ws::QwpWsHostHealthTracker>,
        events: Option<&conn_events::ConnectionEventSource>,
    ) -> Result<RawQwpWsRoundStream> {
        self.connect_round_with(sender::qwp_ws::LockedQwpWsHealth::new(health), events)
    }

    fn connect_round_with<A: sender::qwp_ws::QwpWsHealthAccess>(
        &self,
        health: A,
        events: Option<&conn_events::ConnectionEventSource>,
    ) -> Result<RawQwpWsRoundStream> {
        let mut previous_idx = None;
        let connected = sender::qwp_ws::connect_qwp_ws_endpoint_round(
            &self.endpoints,
            health,
            &mut previous_idx,
            None,
            self.use_tls,
            self.tls_settings.clone(),
            sender::qwp_ws::QwpWsConnectKind::Foreground,
            &self.qwp_ws,
            self.auth_header.as_deref(),
            events,
            None,
        )?;
        // The per-frame cap is the negotiated one: the configured
        // max_buf_size clamped to the server's advertised
        // X-QWP-Max-Batch-Size (0 = not advertised), matching the row
        // sender's effective_qwp_ws_max_buf_size.
        let max_buf_size = if connected.server_max_batch_size > 0 {
            self.max_buf_size.min(connected.server_max_batch_size)
        } else {
            self.max_buf_size
        };
        let raw = RawQwpWsRoundStream {
            endpoint_idx: connected.endpoint_idx,
            stream: connected.stream,
            leftover: connected.leftover,
            max_buf_size,
            request_timeout: *self.qwp_ws.request_timeout,
            durable_ack_opt_in: *self.qwp_ws.request_durable_ack,
        };
        if let Some(events) = events
            && let Some(endpoint) = self.endpoints.get(raw.endpoint_idx)
        {
            // Publish success only after the negotiated stream state,
            // including the server frame cap, has been committed locally.
            events.connect_succeeded(&endpoint.host, &endpoint.port);
        }
        Ok(raw)
    }

    pub(crate) fn connect_sfa_background_with_pool_slot(
        &self,
        sender_id: Option<&str>,
        managed_exclusions: &[conf::QwpWsManagedSlotExclusion],
        extra_orphan_slots: &[PathBuf],
        conn_events: std::sync::Arc<conn_events::ConnectionEventSource>,
        rejection_sink: std::sync::Arc<rejection_events::RejectionEventSource>,
        force_async_initial_connect: bool,
    ) -> Result<sender::qwp_ws::SyncQwpWsHandlerState> {
        let mut qwp_ws = self.qwp_ws.clone();
        // Reconnect-to-sync promotion applies only to standalone
        // `SenderBuilder::build()`; pools honor only an explicitly set mode.
        // Recovery pre-opens still override that mode for this one connect.
        if force_async_initial_connect {
            qwp_ws.force_async_initial_connect();
        }
        // The pool's shared sources exist (handlers already attached, or
        // permanently defaulted) before connect-time recovery senders are
        // pre-opened, so every runner narrates through them from its first
        // connect.
        qwp_ws.conn_events = Some(conn_events);
        qwp_ws.rejection_sink = Some(rejection_sink);
        configure_qwp_ws_pool_slot(
            &mut qwp_ws,
            sender_id,
            managed_exclusions,
            extra_orphan_slots,
        )?;
        sender::qwp_ws::connect_qwp_ws_background_state(
            self.host.as_str(),
            self.port.as_str(),
            self.use_tls,
            self.tls_settings.clone(),
            &qwp_ws,
            self.auth_header.clone(),
        )
    }
}

#[cfg(feature = "sync-sender-qwp-ws")]
fn configure_qwp_ws_pool_slot(
    qwp_ws: &mut conf::QwpWsConfig,
    sender_id: Option<&str>,
    managed_exclusions: &[conf::QwpWsManagedSlotExclusion],
    extra_orphan_slots: &[PathBuf],
) -> Result<()> {
    if let Some(sender_id) = sender_id {
        if !conf::is_valid_qwp_ws_sender_id(sender_id) {
            return Err(error::fmt!(
                ConfigError,
                "invalid pool-managed sender_id [value={sender_id}, allowed-chars=[A-Za-z0-9_-]]"
            ));
        }
        qwp_ws.sender_id = ConfigSetting::new_specified(sender_id.to_owned());
    }
    qwp_ws.orphan_exclude_managed_slots = managed_exclusions.to_vec();
    qwp_ws.orphan_extra_slots = extra_orphan_slots.to_vec();
    qwp_ws.pool_managed_slot = sender_id.is_some();
    Ok(())
}

/// One connection opened by `QwpWsConnector::connect_round_pooled`, tagged with the
/// endpoint index it landed on so the pool can mark that endpoint unhealthy if
/// the connection later dies.
#[cfg(feature = "sync-sender-qwp-ws")]
pub(crate) struct RawQwpWsRoundStream {
    pub(crate) endpoint_idx: usize,
    pub(crate) stream: sender::qwp_ws::WsStream,
    pub(crate) leftover: Vec<u8>,
    pub(crate) max_buf_size: usize,
    pub(crate) request_timeout: Duration,
    pub(crate) durable_ack_opt_in: bool,
}

/// Pre-scan a raw connect string for repeated `addr=...` params. Returns the
/// full list of addr values and a sanitized conf with duplicate `addr=` params
/// removed (the first one is kept so the downstream `questdb_confstr` parser
/// still sees a value).
///
/// Triggered when the schema is one of `ws` or `wss`; for
/// any other schema (or a malformed conf), returns `None` and the caller
/// should fall back to the standard `params.get("addr")` flow.
#[cfg(any(feature = "_sender-qwp-ws", feature = "_egress"))]
pub(crate) fn scan_qwp_ws_addr_params(conf: &str) -> Result<Option<QwpWsAddrScan>> {
    let Some((service, params)) = conf.split_once("::") else {
        return Ok(None);
    };
    if !service.eq_ignore_ascii_case("ws") && !service.eq_ignore_ascii_case("wss") {
        return Ok(None);
    }

    let mut addr_values = Vec::new();
    let mut sanitized_conf = String::with_capacity(conf.len());
    sanitized_conf.push_str(service);
    sanitized_conf.push_str("::");

    let params_offset = service.len() + 2;
    let mut pos = 0usize;
    while pos < params.len() {
        let param_start = pos;
        let Some(eq_rel) = params[pos..].find('=') else {
            return Ok(None);
        };
        let key_start = pos;
        let key_end = pos + eq_rel;
        let key = &params[key_start..key_end];
        pos = key_end + 1;

        let mut value = String::new();
        while pos < params.len() {
            let rest = &params[pos..];
            let mut chars = rest.char_indices();
            let (_, ch) = chars.next().expect("pos is within params");
            if ch == ';' {
                let next_pos = pos + ch.len_utf8();
                if params[next_pos..].starts_with(';') {
                    value.push(';');
                    pos = next_pos + 1;
                    continue;
                }
                pos = next_pos;
                break;
            }
            value.push(ch);
            pos += ch.len_utf8();
        }

        let param_end = pos;
        if key.eq_ignore_ascii_case("addr") {
            if addr_values.is_empty() {
                sanitized_conf
                    .push_str(&conf[params_offset + param_start..params_offset + param_end]);
            }
            addr_values.push(value);
        } else {
            sanitized_conf.push_str(&conf[params_offset + param_start..params_offset + param_end]);
        }
    }

    Ok(Some(QwpWsAddrScan {
        addr_values,
        sanitized_conf,
    }))
}

#[cfg(feature = "_sender-qwp-ws")]
fn parse_qwp_ws_endpoints(
    addr_values: &[String],
    default_port: &str,
) -> Result<Vec<conf::QwpWsEndpoint>> {
    let mut endpoints = Vec::new();
    let mut seen = HashSet::new();
    for addr in addr_values {
        for raw_entry in addr.split(',') {
            let entry = raw_entry.trim();
            if entry.is_empty() {
                return Err(error::fmt!(
                    ConfigError,
                    "invalid QWP/WebSocket addr list: empty entry"
                ));
            }
            let (host, port) = if let Some(rest) = entry.strip_prefix('[') {
                let (host, after) = rest.split_once(']').ok_or_else(|| {
                    error::fmt!(
                        ConfigError,
                        "invalid QWP/WebSocket addr entry {:?}: missing ']'",
                        entry
                    )
                })?;
                let port = match after.strip_prefix(':') {
                    Some(port) => port.trim(),
                    None if after.is_empty() => default_port,
                    None => {
                        return Err(error::fmt!(
                            ConfigError,
                            "invalid QWP/WebSocket addr entry {:?}: \
                             expected ':port' after ']'",
                            entry
                        ));
                    }
                };
                (host.trim(), port)
            } else if entry.matches(':').count() > 1 {
                return Err(error::fmt!(
                    ConfigError,
                    "invalid QWP/WebSocket addr entry {:?}: bracket IPv6 \
                     addresses, e.g. [::1]:9000",
                    entry
                ));
            } else {
                match entry.split_once(':') {
                    Some((host, port)) => (host.trim(), port.trim()),
                    None => (entry, default_port),
                }
            };
            if host.is_empty() {
                return Err(error::fmt!(
                    ConfigError,
                    "invalid QWP/WebSocket addr entry {:?}: empty host",
                    entry
                ));
            }
            if port.is_empty() {
                return Err(error::fmt!(
                    ConfigError,
                    "invalid QWP/WebSocket addr entry {:?}: empty port",
                    entry
                ));
            }
            let parsed_port = port.parse::<u16>().map_err(|_| {
                error::fmt!(
                    ConfigError,
                    "invalid QWP/WebSocket addr entry {:?}: invalid port {:?}",
                    entry,
                    port
                )
            })?;
            if parsed_port == 0 {
                return Err(error::fmt!(
                    ConfigError,
                    "invalid QWP/WebSocket addr entry {:?}: invalid port {:?}",
                    entry,
                    port
                ));
            }
            let normalized_port = parsed_port.to_string();
            let key = (host.to_string(), normalized_port.clone());
            if !seen.insert(key.clone()) {
                return Err(error::fmt!(
                    ConfigError,
                    "duplicate QWP/WebSocket addr endpoint {}:{}",
                    host,
                    normalized_port
                ));
            }
            endpoints.push(conf::QwpWsEndpoint::new(key.0, key.1));
        }
    }
    if endpoints.is_empty() {
        return Err(error::fmt!(
            ConfigError,
            "Missing \"addr\" parameter in config string"
        ));
    }
    Ok(endpoints)
}

/// Accumulates parameters for a new `Sender` instance.
///
/// You can also create the builder from a config string.
///
/// ```no_run
/// # use questdb::Result;
/// use questdb::ingress::SenderBuilder;
///
/// # fn main() -> Result<()> {
/// let mut sender = SenderBuilder::from_conf("https::addr=localhost:9000;")?.build()?;
/// # Ok(())
/// # }
/// ```
///
/// Or create it from the `QDB_CLIENT_CONF` environment variable.
///
/// ```no_run
/// # use questdb::Result;
/// use questdb::ingress::SenderBuilder;
///
/// # fn main() -> Result<()> {
/// // export QDB_CLIENT_CONF="https::addr=localhost:9000;"
/// let mut sender = SenderBuilder::from_env()?.build()?;
/// # Ok(())
/// # }
/// ```
///
/// The `SenderBuilder` can also be built programmatically.
/// The minimum required parameters are the protocol, host, and port.
///
/// ```no_run
/// # use questdb::Result;
/// use questdb::ingress::SenderBuilder;
/// use questdb::ingress::Protocol;
///
/// # fn main() -> Result<()> {
/// # #[cfg(feature = "sync-sender-http")] {
/// let mut sender = SenderBuilder::new(Protocol::Http, "localhost", 9000).build()?;
/// # }
/// # #[cfg(all(not(feature = "sync-sender-http"), feature = "sync-sender-tcp"))] {
/// let mut sender = SenderBuilder::new(Protocol::Tcp, "localhost", 9009).build()?;
/// # }
/// # Ok(())
/// # }
/// ```
#[derive(Debug, Clone)]
#[cfg(feature = "_sync-sender")]
pub struct SenderBuilder {
    protocol: Protocol,
    host: ConfigSetting<String>,
    port: ConfigSetting<String>,
    net_interface: ConfigSetting<Option<String>>,
    init_buf_size: ConfigSetting<usize>,
    max_buf_size: ConfigSetting<usize>,
    max_name_len: ConfigSetting<usize>,
    auth_timeout: ConfigSetting<Duration>,
    username: ConfigSetting<Option<String>>,
    password: ConfigSetting<Option<String>>,
    token: ConfigSetting<Option<String>>,

    #[cfg(feature = "_sender-tcp")]
    token_x: ConfigSetting<Option<String>>,

    #[cfg(feature = "_sender-tcp")]
    token_y: ConfigSetting<Option<String>>,

    protocol_version: ConfigSetting<Option<ProtocolVersion>>,

    #[cfg(feature = "insecure-skip-verify")]
    tls_verify: ConfigSetting<bool>,

    tls_ca: ConfigSetting<CertificateAuthority>,
    tls_roots: ConfigSetting<Option<PathBuf>>,

    /// Password unlocking a JKS / PKCS#12 keystore named by
    /// `tls_roots`. QWP/WebSocket only — other transports keep PEM
    /// as the sole `tls_roots` format.
    #[cfg(feature = "_sender-qwp-ws")]
    tls_roots_password: ConfigSetting<Option<String>>,

    #[cfg(feature = "_sender-http")]
    http: Option<conf::HttpConfig>,

    #[cfg(feature = "_sender-qwp-udp")]
    qwp_udp: Option<conf::QwpUdpConfig>,

    #[cfg(feature = "_sender-qwp-ws")]
    qwp_ws: Option<conf::QwpWsConfig>,

    #[cfg(feature = "_sender-qwp-ws")]
    qwp_ws_error_handler: QwpWsErrorHandler,
}

#[cfg(feature = "_sync-sender")]
impl SenderBuilder {
    /// Create a new `SenderBuilder` instance from the configuration string.
    ///
    /// The format of the string is: `"http::addr=host:port;key=value;...;"`.
    ///
    /// Instead of `"http"`, you can also specify `"https"`, `"tcp"`, `"tcps"`,
    /// `"udp"`, and the QWP/WebSocket schemes `"ws"` / `"wss"` when the
    /// corresponding sender features are enabled.
    ///
    /// We recommend HTTP for most cases because it provides more features, like
    /// reporting errors to the client and supporting transaction control. TCP can
    /// sometimes be faster in higher-latency networks, but misses a number of
    /// features.
    ///
    /// Many accepted keys match one-for-one with the methods on `SenderBuilder`.
    /// For example, this is a valid configuration string:
    ///
    /// "https::addr=host:port;username=alice;password=secret;"
    ///
    /// and there are matching methods [SenderBuilder::username] and
    /// [SenderBuilder::password]. The value of `addr=` is supplied directly to the
    /// `SenderBuilder` constructor, so there's no matching method for that.
    ///
    /// Some QWP/WebSocket configuration keys are accepted only through the
    /// configuration string, primarily for compatibility with Java-style
    /// configuration names and settings without a public Rust builder method.
    /// These include `sf_dir`, `sender_id`, `sf_max_segment_bytes`,
    /// `sf_max_total_bytes`, `sf_durability`, `sf_sync_interval_millis`,
    /// `sf_append_deadline_millis`, `auth_timeout_ms`, `close_flush_timeout_millis`,
    /// `request_durable_ack`,
    /// `durable_ack_keepalive_interval_millis`, `drain_orphans`,
    /// `max_background_drainers`, and `error_inbox_capacity`.
    ///
    /// `sf_max_segment_bytes` defaults to 4 MiB. Smaller disk-backed segments
    /// release acknowledged space more granularly, but rotate more often and
    /// therefore increase crash-consistency synchronization and file-operation
    /// overhead.
    ///
    /// You can also load the configuration from an environment variable. See
    /// [`SenderBuilder::from_env`].
    ///
    /// Once you have a `SenderBuilder` instance, you can further customize it
    /// before calling [`SenderBuilder::build`], but you can't change any settings
    /// that are already set in the config string.
    pub fn from_conf<T: AsRef<str>>(conf: T) -> Result<Self> {
        let conf = conf.as_ref();
        #[cfg(feature = "_sender-qwp-ws")]
        let qwp_ws_addr_scan = scan_qwp_ws_addr_params(conf)?;
        #[cfg(feature = "_sender-qwp-ws")]
        let conf_to_parse = qwp_ws_addr_scan
            .as_ref()
            .map(|scan| scan.sanitized_conf.as_str())
            .unwrap_or(conf);
        #[cfg(not(feature = "_sender-qwp-ws"))]
        let conf_to_parse = conf;

        let conf = questdb_confstr::parse_conf_str(conf_to_parse)
            .map_err(|e| error::fmt!(ConfigError, "Config parse error: {}", e))?;
        let service = conf.service();
        let params = conf.params();

        let protocol = Protocol::from_schema(service)?;
        #[cfg(feature = "_sender-qwp-ws")]
        let conf_is_qwp_ws = protocol.is_qwp_ws();

        let Some(addr) = params.get("addr") else {
            return Err(error::fmt!(
                ConfigError,
                "Missing \"addr\" parameter in config string"
            ));
        };
        #[cfg(feature = "_sender-qwp-ws")]
        let qwp_ws_endpoints = if protocol.is_qwp_ws() {
            let addr_values = qwp_ws_addr_scan
                .as_ref()
                .map(|scan| scan.addr_values.as_slice())
                .unwrap_or_else(|| std::slice::from_ref(addr));
            Some(parse_qwp_ws_endpoints(
                addr_values,
                protocol.default_port(),
            )?)
        } else {
            None
        };
        let (host, port) = {
            #[cfg(feature = "_sender-qwp-ws")]
            if let Some(endpoints) = qwp_ws_endpoints.as_ref() {
                let first = endpoints.first().ok_or_else(|| {
                    error::fmt!(ConfigError, "Missing \"addr\" parameter in config string")
                })?;
                (first.host.as_str(), first.port.as_str())
            } else {
                match addr.split_once(':') {
                    Some((h, p)) => (h, p),
                    None => (addr.as_str(), protocol.default_port()),
                }
            }

            #[cfg(not(feature = "_sender-qwp-ws"))]
            {
                match addr.split_once(':') {
                    Some((h, p)) => (h, p),
                    None => (addr.as_str(), protocol.default_port()),
                }
            }
        };
        let mut builder = SenderBuilder::new(protocol, host, port);
        #[cfg(feature = "_sender-qwp-ws")]
        if let Some(endpoints) = qwp_ws_endpoints {
            builder = builder.qwp_ws_endpoints(endpoints)?;
        }

        validate_auto_flush_params(params)?;

        // Connect-string keys valid on a `ws::` / `wss::` string that are not
        // matched by an arm below: `addr` (consumed before this loop), the
        // auto-flush keys (validated by `validate_auto_flush_params`), the
        // egress query-client keys (a single connect string drives both the
        // sender and the `QwpQueryClient`), and the ingestion-pool keys
        // (`pool_*`, consumed by `QuestDb::connect` before it opens each
        // per-slot `SenderBuilder::from_conf` connection). Any other key on a
        // QWP/WebSocket connect string is rejected as unknown. Keys added to
        // any of these directions MUST be reflected here or a shared connect
        // string breaks.
        #[cfg(feature = "_sender-qwp-ws")]
        const QWP_WS_PORTABLE_CONFIG_KEYS: &[&str] = &[
            "addr",
            "auth",
            "auto_flush",
            "auto_flush_bytes",
            "auto_flush_interval",
            "auto_flush_rows",
            "buffer_pool_size",
            "client_id",
            "compression",
            "compression_level",
            "failover",
            "failover_backoff_initial_ms",
            "failover_backoff_max_ms",
            "failover_max_attempts",
            "failover_max_duration_ms",
            "max_batch_rows",
            "max_version",
            "on_internal_error",
            "on_parse_error",
            "on_schema_error",
            "on_security_error",
            "on_server_error",
            "on_write_error",
            "path",
            "acquire_timeout_ms",
            "idle_timeout_ms",
            "lazy_connect",
            "pool_reap",
            "query_pool_max",
            "query_pool_min",
            "sender_pool_max",
            "sender_pool_min",
            "target",
            "zone",
        ];

        for (key, val) in params.iter().map(|(k, v)| (k.as_str(), v.as_str())) {
            builder = match key {
                "username" => builder.username(val)?,
                "password" => builder.password(val)?,
                "token" => builder.token(val)?,
                "token_x" => builder.token_x(val)?,
                "token_y" => builder.token_y(val)?,
                "bind_interface" => builder.bind_interface(val)?,
                #[cfg(feature = "_sender-qwp-udp")]
                "max_datagram_size" => builder.max_datagram_size(parse_conf_value(key, val)?)?,
                #[cfg(feature = "_sender-qwp-udp")]
                "multicast_ttl" => builder.multicast_ttl(parse_conf_value(key, val)?)?,
                #[cfg(feature = "_sender-qwp-ws")]
                "qwp_ws_progress" => builder.qwp_ws_progress(parse_qwp_ws_progress_value(val)?)?,
                #[cfg(feature = "_sender-qwp-ws")]
                "sf_dir" => builder.store_and_forward_dir(PathBuf::from(val))?,
                #[cfg(feature = "_sender-qwp-ws")]
                "sender_id" => builder.sender_id(val)?,
                #[cfg(feature = "_sender-qwp-ws")]
                "sf_max_segment_bytes" => {
                    builder.store_and_forward_max_bytes(parse_size_conf_value(key, val)?)?
                }
                #[cfg(feature = "_sender-qwp-ws")]
                "sf_max_total_bytes" => {
                    builder.store_and_forward_max_total_bytes(parse_size_conf_value(key, val)?)?
                }
                #[cfg(feature = "_sender-qwp-ws")]
                "sf_durability" => {
                    builder.store_and_forward_durability(parse_sf_durability_value(val)?)?
                }
                #[cfg(feature = "_sender-qwp-ws")]
                "sf_sync_interval_millis" => builder.store_and_forward_sync_interval_millis(val)?,
                #[cfg(feature = "_sender-qwp-ws")]
                "sf_append_deadline_millis" => builder.store_and_forward_append_deadline(
                    Duration::from_millis(parse_conf_value(key, val)?),
                )?,
                #[cfg(feature = "_sender-qwp-ws")]
                "reconnect_max_duration_millis" => builder
                    .reconnect_max_duration(Duration::from_millis(parse_conf_value(key, val)?))?,
                #[cfg(feature = "_sender-qwp-ws")]
                "reconnect_initial_backoff_millis" => builder.reconnect_initial_backoff(
                    Duration::from_millis(parse_conf_value(key, val)?),
                )?,
                #[cfg(feature = "_sender-qwp-ws")]
                "reconnect_max_backoff_millis" => builder
                    .reconnect_max_backoff(Duration::from_millis(parse_conf_value(key, val)?))?,
                #[cfg(feature = "_sender-qwp-ws")]
                "initial_connect_retry" => {
                    builder.qwp_ws_initial_connect_mode(parse_initial_connect_retry_value(val)?)?
                }
                #[cfg(feature = "_sender-qwp-ws")]
                "auth_timeout_ms" => builder.qwp_ws_auth_timeout_millis(val)?,
                #[cfg(feature = "_sender-qwp-ws")]
                "connect_timeout" => builder.qwp_ws_connect_timeout_millis(val)?,
                #[cfg(feature = "_sender-qwp-ws")]
                "close_flush_timeout_millis" => builder.close_flush_timeout_millis(val)?,
                #[cfg(feature = "_sender-qwp-ws")]
                "request_durable_ack" => builder.request_durable_ack(val)?,
                #[cfg(feature = "_sender-qwp-ws")]
                "durable_ack_keepalive_interval_millis" => {
                    builder.durable_ack_keepalive_interval_millis(val)?
                }
                #[cfg(feature = "_sender-qwp-ws")]
                "drain_orphans" => builder.drain_orphans(val)?,
                #[cfg(feature = "_sender-qwp-ws")]
                "max_background_drainers" => builder.max_background_drainers(val)?,
                #[cfg(feature = "_sender-qwp-ws")]
                "error_inbox_capacity" => builder.error_inbox_capacity(val)?,
                #[cfg(feature = "_sender-qwp-ws")]
                "max_frame_rejections" => {
                    builder.max_frame_rejections(parse_conf_value(key, val)?)?
                }
                #[cfg(feature = "_sender-qwp-ws")]
                "poison_min_escalation_window_millis" => builder.poison_min_escalation_window(
                    Duration::from_millis(parse_conf_value(key, val)?),
                )?,
                "protocol_version" => match val {
                    "1" => builder.protocol_version(ProtocolVersion::V1)?,
                    "2" => builder.protocol_version(ProtocolVersion::V2)?,
                    "3" => builder.protocol_version(ProtocolVersion::V3)?,
                    "auto" => builder,
                    invalid => {
                        return Err(error::fmt!(
                            ConfigError,
                            "invalid \"protocol_version\" [value={invalid}, allowed-values=[auto, 1, 2, 3]]"
                        ));
                    }
                },
                "max_name_len" => builder.max_name_len(parse_conf_value(key, val)?)?,

                "init_buf_size" => builder.init_buf_size(parse_conf_value(key, val)?)?,

                "max_buf_size" => builder.max_buf_size(parse_conf_value(key, val)?)?,

                "auth_timeout" => {
                    builder.auth_timeout(Duration::from_millis(parse_conf_value(key, val)?))?
                }

                "tls_verify" => {
                    let verify = match val {
                        "on" => true,
                        "unsafe_off" => false,
                        _ => {
                            return Err(fmt!(
                                ConfigError,
                                r##"Config parameter "tls_verify" must be either "on" or "unsafe_off".'"##,
                            ));
                        }
                    };

                    #[cfg(not(feature = "insecure-skip-verify"))]
                    {
                        if !verify {
                            return Err(fmt!(
                                ConfigError,
                                r##"The "insecure-skip-verify" feature is not enabled, so "tls_verify=unsafe_off" is not supported"##,
                            ));
                        }
                        builder
                    }

                    #[cfg(feature = "insecure-skip-verify")]
                    builder.tls_verify(verify)?
                }

                "tls_ca" => {
                    #[allow(unreachable_code, unused_variables)]
                    {
                        let ca = match val {
                            #[cfg(feature = "tls-webpki-certs")]
                            "webpki_roots" => CertificateAuthority::WebpkiRoots,

                            #[cfg(not(feature = "tls-webpki-certs"))]
                            "webpki_roots" => {
                                return Err(error::fmt!(
                                    ConfigError,
                                    "Config parameter \"tls_ca=webpki_roots\" requires the \"tls-webpki-certs\" feature"
                                ));
                            }

                            #[cfg(feature = "tls-native-certs")]
                            "os_roots" => CertificateAuthority::OsRoots,

                            #[cfg(not(feature = "tls-native-certs"))]
                            "os_roots" => {
                                return Err(error::fmt!(
                                    ConfigError,
                                    "Config parameter \"tls_ca=os_roots\" requires the \"tls-native-certs\" feature"
                                ));
                            }

                            #[cfg(all(feature = "tls-webpki-certs", feature = "tls-native-certs"))]
                            "webpki_and_os_roots" => CertificateAuthority::WebpkiAndOsRoots,

                            #[cfg(not(all(
                                feature = "tls-webpki-certs",
                                feature = "tls-native-certs"
                            )))]
                            "webpki_and_os_roots" => {
                                return Err(error::fmt!(
                                    ConfigError,
                                    "Config parameter \"tls_ca=webpki_and_os_roots\" requires both the \"tls-webpki-certs\" and \"tls-native-certs\" features"
                                ));
                            }

                            _ => {
                                return Err(error::fmt!(
                                    ConfigError,
                                    "Invalid value {val:?} for \"tls_ca\""
                                ));
                            }
                        };
                        builder.tls_ca(ca)?
                    }
                }

                "tls_roots" => {
                    let path = PathBuf::from_str(val).map_err(|e| {
                        error::fmt!(
                            ConfigError,
                            "Invalid path {:?} for \"tls_roots\": {}",
                            val,
                            e
                        )
                    })?;
                    builder.tls_roots(path)?
                }

                "tls_roots_password" => {
                    #[cfg(feature = "_sender-qwp-ws")]
                    {
                        builder.tls_roots_password(val.to_string())?
                    }
                    #[cfg(not(feature = "_sender-qwp-ws"))]
                    {
                        return Err(error::fmt!(
                            ConfigError,
                            "\"tls_roots_password\" is only supported for QWP/WebSocket \
                             (ws / wss). ILP/TCP and ILP/HTTP transports read \
                             unencrypted PEM via rustls."
                        ));
                    }
                }

                #[cfg(feature = "sync-sender-http")]
                "request_min_throughput" => {
                    builder.request_min_throughput(parse_conf_value(key, val)?)?
                }

                #[cfg(feature = "sync-sender-http")]
                "request_timeout" => {
                    builder.request_timeout(Duration::from_millis(parse_conf_value(key, val)?))?
                }

                #[cfg(feature = "sync-sender-http")]
                "retry_timeout" => {
                    builder.retry_timeout(Duration::from_millis(parse_conf_value(key, val)?))?
                }
                #[cfg(feature = "sync-sender-http")]
                "retry_max_backoff_millis" => {
                    builder.retry_max_backoff(Duration::from_millis(parse_conf_value(key, val)?))?
                }

                // QWP/WebSocket follows the connect-string spec: a key that is
                // neither matched above nor portable (QWP_WS_PORTABLE_CONFIG_KEYS)
                // is a typo or unsupported option and is rejected. Legacy ILP
                // transports keep ignoring unknown keys -- a parameter added to
                // one ILP client must not force a lock-step release of the others.
                #[cfg(feature = "_sender-qwp-ws")]
                other if conf_is_qwp_ws && !QWP_WS_PORTABLE_CONFIG_KEYS.contains(&other) => {
                    return Err(error::fmt!(ConfigError, "Unknown config key \"{}\"", other));
                }
                _ => builder,
            };
        }

        Ok(builder)
    }

    /// Create a new `SenderBuilder` instance from the configuration from the
    /// configuration stored in the `QDB_CLIENT_CONF` environment variable.
    ///
    /// The format of the string is the same as for [`SenderBuilder::from_conf`].
    pub fn from_env() -> Result<Self> {
        let conf = std::env::var("QDB_CLIENT_CONF").map_err(|_| {
            error::fmt!(ConfigError, "Environment variable QDB_CLIENT_CONF not set.")
        })?;
        Self::from_conf(conf)
    }

    /// Create a new `SenderBuilder` instance with the provided QuestDB
    /// server and port, using ILP over the specified protocol.
    ///
    /// ```no_run
    /// # use questdb::Result;
    /// use questdb::ingress::{Protocol, SenderBuilder};
    ///
    /// # fn main() -> Result<()> {
    /// # #[cfg(feature = "sync-sender-tcp")] {
    /// let mut sender = SenderBuilder::new(
    ///     Protocol::Tcp, "localhost", 9009).build()?;
    /// # }
    /// # #[cfg(all(not(feature = "sync-sender-tcp"), feature = "sync-sender-http"))] {
    /// let mut sender = SenderBuilder::new(
    ///     Protocol::Http, "localhost", 9000).build()?;
    /// # }
    /// # Ok(())
    /// # }
    /// ```
    pub fn new<H: Into<String>, P: Into<Port>>(protocol: Protocol, host: H, port: P) -> Self {
        let host = host.into();
        let port: Port = port.into();
        let port = port.0;

        #[cfg(feature = "tls-webpki-certs")]
        let tls_ca = CertificateAuthority::WebpkiRoots;

        #[cfg(all(not(feature = "tls-webpki-certs"), feature = "tls-native-certs"))]
        let tls_ca = CertificateAuthority::OsRoots;

        #[cfg(not(any(feature = "tls-webpki-certs", feature = "tls-native-certs")))]
        let tls_ca = CertificateAuthority::PemFile;

        Self {
            protocol,
            host: ConfigSetting::new_specified(host),
            port: ConfigSetting::new_specified(port),
            net_interface: ConfigSetting::new_default(None),
            init_buf_size: ConfigSetting::new_default(64 * 1024),
            max_buf_size: ConfigSetting::new_default(100 * 1024 * 1024),
            max_name_len: ConfigSetting::new_default(MAX_NAME_LEN_DEFAULT),
            auth_timeout: ConfigSetting::new_default(Duration::from_secs(15)),
            username: ConfigSetting::new_default(None),
            password: ConfigSetting::new_default(None),
            token: ConfigSetting::new_default(None),

            #[cfg(feature = "_sender-tcp")]
            token_x: ConfigSetting::new_default(None),

            #[cfg(feature = "_sender-tcp")]
            token_y: ConfigSetting::new_default(None),

            protocol_version: ConfigSetting::new_default(None),

            #[cfg(feature = "insecure-skip-verify")]
            tls_verify: ConfigSetting::new_default(true),

            tls_ca: ConfigSetting::new_default(tls_ca),
            tls_roots: ConfigSetting::new_default(None),

            #[cfg(feature = "_sender-qwp-ws")]
            tls_roots_password: ConfigSetting::new_default(None),

            #[cfg(feature = "sync-sender-http")]
            http: if protocol.is_httpx() {
                Some(conf::HttpConfig::default())
            } else {
                None
            },

            #[cfg(feature = "_sender-qwp-udp")]
            qwp_udp: if protocol.is_qwp_udp() {
                Some(conf::QwpUdpConfig::default())
            } else {
                None
            },

            #[cfg(feature = "_sender-qwp-ws")]
            qwp_ws: if protocol.is_qwp_ws() {
                Some(conf::QwpWsConfig::default())
            } else {
                None
            },

            #[cfg(feature = "_sender-qwp-ws")]
            qwp_ws_error_handler: QwpWsErrorHandler::log_default(),
        }
    }

    /// Install a producer-thread handler for structured QWP/WebSocket server
    /// diagnostics.
    ///
    /// The handler runs synchronously from sender API calls such as `flush`.
    /// It must not call methods on the same sender.
    #[cfg(feature = "_sender-qwp-ws")]
    pub fn qwp_ws_error_handler<F>(mut self, handler: F) -> Result<Self>
    where
        F: Fn(&QwpWsSenderError) + Send + Sync + 'static,
    {
        self.qwp_ws_error_handler = QwpWsErrorHandler::new(handler);
        Ok(self)
    }

    /// Select local outbound interface.
    ///
    /// This may be relevant if your machine has multiple network interfaces.
    ///
    /// The default is `"0.0.0.0"`.
    pub fn bind_interface<I: Into<String>>(self, addr: I) -> Result<Self> {
        #[cfg(any(feature = "_sender-tcp", feature = "_sender-qwp-udp"))]
        {
            let mut builder = self;
            builder.ensure_supports_bind_interface("bind_interface")?;
            builder
                .net_interface
                .set_specified("bind_interface", Some(validate_value(addr.into())?))?;
            Ok(builder)
        }

        #[cfg(not(any(feature = "_sender-tcp", feature = "_sender-qwp-udp")))]
        {
            let _ = addr;
            Err(error::fmt!(
                ConfigError,
                "The \"bind_interface\" setting can only be used with the TCP protocol."
            ))
        }
    }

    /// Set the username for authentication.
    ///
    /// For TCP, this is the `kid` part of the ECDSA key set.
    /// The other fields are [`token`](SenderBuilder::token), [`token_x`](SenderBuilder::token_x),
    /// and [`token_y`](SenderBuilder::token_y).
    ///
    /// For HTTP, this is a part of basic authentication.
    /// See also: [`password`](SenderBuilder::password).
    pub fn username(mut self, username: &str) -> Result<Self> {
        #[cfg(feature = "_sender-qwp-udp")]
        self.reject_if_qwp_udp("username")?;
        self.username
            .set_specified("username", Some(validate_value(username.to_string())?))?;
        Ok(self)
    }

    /// Set the password for basic HTTP authentication.
    /// See also: [`username`](SenderBuilder::username).
    pub fn password(mut self, password: &str) -> Result<Self> {
        #[cfg(feature = "_sender-qwp-udp")]
        self.reject_if_qwp_udp("password")?;
        self.password
            .set_specified("password", Some(validate_value(password.to_string())?))?;
        Ok(self)
    }

    /// Set the bearer-token authentication parameter for HTTP or
    /// QWP/WebSocket, which requires QuestDB Enterprise, or set the ECDSA
    /// private key for TCP authentication.
    pub fn token(mut self, token: &str) -> Result<Self> {
        #[cfg(feature = "_sender-qwp-udp")]
        self.reject_if_qwp_udp("token")?;
        self.token
            .set_specified("token", Some(validate_value(token.to_string())?))?;
        Ok(self)
    }

    /// Set the ECDSA public key X for TCP authentication.
    pub fn token_x(self, token_x: &str) -> Result<Self> {
        #[cfg(feature = "_sender-qwp-udp")]
        self.reject_if_qwp_udp("token_x")?;
        #[cfg(feature = "_sender-tcp")]
        {
            let mut builder = self;
            builder
                .token_x
                .set_specified("token_x", Some(validate_value(token_x.to_string())?))?;
            Ok(builder)
        }

        #[cfg(not(feature = "_sender-tcp"))]
        {
            let _ = token_x;
            Err(error::fmt!(
                ConfigError,
                "cannot specify \"token_x\": ECDSA authentication is only available with ILP/TCP and not available with ILP/HTTP."
            ))
        }
    }

    /// Set the ECDSA public key Y for TCP authentication.
    pub fn token_y(self, token_y: &str) -> Result<Self> {
        #[cfg(feature = "_sender-qwp-udp")]
        self.reject_if_qwp_udp("token_y")?;
        #[cfg(feature = "_sender-tcp")]
        {
            let mut builder = self;
            builder
                .token_y
                .set_specified("token_y", Some(validate_value(token_y.to_string())?))?;
            Ok(builder)
        }

        #[cfg(not(feature = "_sender-tcp"))]
        {
            let _ = token_y;
            Err(error::fmt!(
                ConfigError,
                "cannot specify \"token_y\": ECDSA authentication is only available with ILP/TCP and not available with ILP/HTTP."
            ))
        }
    }

    /// Sets the protocol version for ILP transports.
    /// - HTTP transport automatically negotiates the protocol version by default(unset, **Strong Recommended**).
    ///   You can explicitly configure the protocol version to avoid the slight latency cost at connection time.
    /// - TCP transport does not negotiate the protocol version and uses [`ProtocolVersion::V1`] by
    ///   default. You must explicitly set [`ProtocolVersion::V2`] in order to ingest
    ///   arrays.
    /// - QWP/UDP does not support explicit `protocol_version` configuration.
    ///
    /// **Note**: QuestDB server version 9.0.0 or later is required for [`ProtocolVersion::V2`] support.
    pub fn protocol_version(mut self, protocol_version: ProtocolVersion) -> Result<Self> {
        #[cfg(feature = "_sender-qwp-udp")]
        self.reject_if_qwp_udp("protocol_version")?;
        self.protocol_version
            .set_specified("protocol_version", Some(protocol_version))?;
        Ok(self)
    }

    #[cfg(feature = "_sender-qwp-udp")]
    /// Set the maximum datagram size in bytes for QWP/UDP transport.
    ///
    /// `value` must be between 1 and 65,507 bytes, inclusive. The upper bound
    /// is the UDP/IPv4 payload limit, not a recommended operating size. The
    /// default is 1,400 bytes, leaving room for IPv4 and UDP headers under a
    /// common 1,500-byte Ethernet MTU. If you raise this value, keep it within
    /// the effective UDP payload budget for the path MTU. Oversized IPv4
    /// packets may be fragmented when fragmentation is allowed, or dropped when
    /// it is not; fragmented UDP is fragile because losing any fragment loses
    /// the whole datagram.
    pub fn max_datagram_size(mut self, value: usize) -> Result<Self> {
        if value == 0 {
            return Err(error::fmt!(
                ConfigError,
                "\"max_datagram_size\" must be greater than 0."
            ));
        }
        if value > 65507 {
            return Err(error::fmt!(
                ConfigError,
                "\"max_datagram_size\" must not exceed 65507 (UDP/IPv4 limit)."
            ));
        }
        let Some(qwp_udp) = &mut self.qwp_udp else {
            return Err(error::fmt!(
                ConfigError,
                "The \"max_datagram_size\" setting is only supported for QWP/UDP."
            ));
        };
        qwp_udp
            .max_datagram_size
            .set_specified("max_datagram_size", value)?;
        Ok(self)
    }

    #[cfg(feature = "_sender-qwp-udp")]
    /// Set the multicast TTL for QWP/UDP transport. The default is 1.
    ///
    /// Use a value greater than 0 when sending to a multicast address. A value
    /// of 0 prevents multicast datagrams from leaving the local host.
    pub fn multicast_ttl(mut self, value: u32) -> Result<Self> {
        if value > 255 {
            return Err(error::fmt!(
                ConfigError,
                "\"multicast_ttl\" must be between 0 and 255."
            ));
        }
        let Some(qwp_udp) = &mut self.qwp_udp else {
            return Err(error::fmt!(
                ConfigError,
                "The \"multicast_ttl\" setting is only supported for QWP/UDP."
            ));
        };
        qwp_udp
            .multicast_ttl
            .set_specified("multicast_ttl", value)?;
        Ok(self)
    }

    #[cfg(feature = "_sender-qwp-ws")]
    /// Register a connection lifecycle listener: one
    /// [`ConnectionEvent`] per
    /// connection-state transition of this sender's QWP/WebSocket
    /// connection (initial connect, per-endpoint attempt failures,
    /// disconnect, reconnect/failover, terminal auth rejection).
    /// Delivered on a dedicated dispatcher thread through a bounded inbox
    /// (`inbox_capacity`; `0` selects the default of 64) with a
    /// drop-oldest overflow policy. At most one listener per sender.
    pub fn connection_listener(
        mut self,
        listener: crate::ingress::ConnectionListener,
        inbox_capacity: usize,
    ) -> Result<Self> {
        let Some(qwp_ws) = &mut self.qwp_ws else {
            return Err(error::fmt!(
                ConfigError,
                "The \"connection_listener\" setting is only supported for QWP/WebSocket."
            ));
        };
        if qwp_ws.conn_events.is_some() {
            return Err(error::fmt!(
                ConfigError,
                "A connection listener is already registered on this builder."
            ));
        }
        qwp_ws.conn_events = Some(std::sync::Arc::new(
            conn_events::ConnectionEventSource::new(listener, inbox_capacity),
        ));
        Ok(self)
    }

    #[cfg(feature = "_sender-qwp-ws")]
    /// Control whether QWP/WebSocket progress is driven by a background thread
    /// or manually by the caller. The default is [`QwpWsProgress::Background`],
    /// matching the Java sender.
    pub fn qwp_ws_progress(mut self, progress: QwpWsProgress) -> Result<Self> {
        let Some(qwp_ws) = &mut self.qwp_ws else {
            return Err(error::fmt!(
                ConfigError,
                "The \"qwp_ws_progress\" setting is only supported for QWP/WebSocket."
            ));
        };
        qwp_ws.progress.set_specified("qwp_ws_progress", progress)?;
        Ok(self)
    }

    #[cfg(feature = "_sender-qwp-ws")]
    fn store_and_forward_dir(mut self, dir: PathBuf) -> Result<Self> {
        if dir.as_os_str().is_empty() {
            return Err(error::fmt!(ConfigError, "\"sf_dir\" cannot be empty."));
        }
        let Some(qwp_ws) = &mut self.qwp_ws else {
            return Err(error::fmt!(
                ConfigError,
                "The \"sf_dir\" setting is only supported for QWP/WebSocket."
            ));
        };
        qwp_ws.sf_dir.set_specified("sf_dir", Some(dir))?;
        Ok(self)
    }

    #[cfg(feature = "_sender-qwp-ws")]
    fn sender_id(mut self, sender_id: &str) -> Result<Self> {
        let sender_id = validate_value(sender_id)?;
        if !conf::is_valid_qwp_ws_sender_id(sender_id) {
            return Err(error::fmt!(
                ConfigError,
                "invalid sender_id [value={sender_id}, allowed-chars=[A-Za-z0-9_-]]"
            ));
        }
        let Some(qwp_ws) = &mut self.qwp_ws else {
            return Err(error::fmt!(
                ConfigError,
                "The \"sender_id\" setting is only supported for QWP/WebSocket."
            ));
        };
        qwp_ws
            .sender_id
            .set_specified("sender_id", sender_id.to_owned())?;
        Ok(self)
    }

    #[cfg(feature = "_sender-qwp-ws")]
    fn store_and_forward_max_bytes(mut self, value: u64) -> Result<Self> {
        if value == 0 {
            return Err(error::fmt!(
                ConfigError,
                "\"sf_max_segment_bytes\" must be greater than 0."
            ));
        }
        let Some(qwp_ws) = &mut self.qwp_ws else {
            return Err(error::fmt!(
                ConfigError,
                "The \"sf_max_segment_bytes\" setting is only supported for QWP/WebSocket."
            ));
        };
        qwp_ws
            .sf_max_segment_bytes
            .set_specified("sf_max_segment_bytes", value)?;
        Ok(self)
    }

    #[cfg(feature = "_sender-qwp-ws")]
    fn store_and_forward_max_total_bytes(mut self, value: u64) -> Result<Self> {
        if value == 0 {
            return Err(error::fmt!(
                ConfigError,
                "\"sf_max_total_bytes\" must be greater than 0."
            ));
        }
        let Some(qwp_ws) = &mut self.qwp_ws else {
            return Err(error::fmt!(
                ConfigError,
                "The \"sf_max_total_bytes\" setting is only supported for QWP/WebSocket."
            ));
        };
        qwp_ws
            .sf_max_total_bytes
            .set_specified("sf_max_total_bytes", Some(value))?;
        Ok(self)
    }

    #[cfg(feature = "_sender-qwp-ws")]
    fn store_and_forward_durability(mut self, durability: conf::SfDurability) -> Result<Self> {
        let Some(qwp_ws) = &mut self.qwp_ws else {
            return Err(error::fmt!(
                ConfigError,
                "The \"sf_durability\" setting is only supported for QWP/WebSocket."
            ));
        };
        qwp_ws
            .sf_durability
            .set_specified("sf_durability", durability)?;
        Ok(self)
    }

    #[cfg(feature = "_sender-qwp-ws")]
    fn store_and_forward_sync_interval_millis(mut self, value: &str) -> Result<Self> {
        const MAX_MILLIS: i64 = i64::MAX / 1_000_000;

        let Some(qwp_ws) = &mut self.qwp_ws else {
            return Err(error::fmt!(
                ConfigError,
                "The \"sf_sync_interval_millis\" setting is only supported for QWP/WebSocket."
            ));
        };
        let millis: i64 = parse_conf_value("sf_sync_interval_millis", value)?;
        if millis <= 0 {
            return Err(error::fmt!(
                ConfigError,
                "\"sf_sync_interval_millis\" must be greater than 0."
            ));
        }
        if millis > MAX_MILLIS {
            return Err(error::fmt!(
                ConfigError,
                "\"sf_sync_interval_millis\" must be at most {MAX_MILLIS}."
            ));
        }
        qwp_ws.sf_sync_interval.set_specified(
            "sf_sync_interval_millis",
            Some(Duration::from_millis(millis as u64)),
        )?;
        Ok(self)
    }

    #[cfg(feature = "_sender-qwp-ws")]
    fn store_and_forward_append_deadline(mut self, value: Duration) -> Result<Self> {
        if value.is_zero() {
            return Err(error::fmt!(
                ConfigError,
                "\"sf_append_deadline_millis\" must be greater than 0."
            ));
        }
        let Some(qwp_ws) = &mut self.qwp_ws else {
            return Err(error::fmt!(
                ConfigError,
                "The \"sf_append_deadline_millis\" setting is only supported for QWP/WebSocket."
            ));
        };
        qwp_ws
            .sf_append_deadline
            .set_specified("sf_append_deadline_millis", value)?;
        Ok(self)
    }

    #[cfg(feature = "_sender-qwp-ws")]
    /// Per-outage reconnect retry budget. Default 300s.
    pub fn reconnect_max_duration(mut self, value: Duration) -> Result<Self> {
        let Some(qwp_ws) = &mut self.qwp_ws else {
            return Err(error::fmt!(
                ConfigError,
                "The \"reconnect_max_duration_millis\" setting is only supported for QWP/WebSocket."
            ));
        };
        qwp_ws
            .reconnect_max_duration
            .set_specified("reconnect_max_duration_millis", value)?;
        Ok(self)
    }

    #[cfg(feature = "_sender-qwp-ws")]
    /// Maximum repeated same-head-FSN rejects or server close frames tolerated
    /// without ACK progress before the sender treats the frame as poison.
    /// Default 4, matching the Java QWP/WebSocket sender.
    pub fn max_frame_rejections(mut self, value: usize) -> Result<Self> {
        if value == 0 {
            return Err(error::fmt!(
                ConfigError,
                "\"max_frame_rejections\" must be greater than 0."
            ));
        }
        let Some(qwp_ws) = &mut self.qwp_ws else {
            return Err(error::fmt!(
                ConfigError,
                "The \"max_frame_rejections\" setting is only supported for QWP/WebSocket."
            ));
        };
        qwp_ws
            .max_frame_rejections
            .set_specified("max_frame_rejections", value)?;
        Ok(self)
    }

    #[cfg(feature = "_sender-qwp-ws")]
    /// Minimum dwell before repeated same-head-FSN rejects or server close
    /// frames can escalate to a poison-frame protocol violation. Default 5s.
    /// Set to zero to escalate immediately at `max_frame_rejections`.
    pub fn poison_min_escalation_window(mut self, value: Duration) -> Result<Self> {
        let Some(qwp_ws) = &mut self.qwp_ws else {
            return Err(error::fmt!(
                ConfigError,
                "The \"poison_min_escalation_window_millis\" setting is only supported for QWP/WebSocket."
            ));
        };
        qwp_ws
            .poison_min_escalation_window
            .set_specified("poison_min_escalation_window_millis", value)?;
        Ok(self)
    }

    #[cfg(feature = "_sender-qwp-ws")]
    /// Initial reconnect backoff. Default 100ms.
    pub fn reconnect_initial_backoff(mut self, value: Duration) -> Result<Self> {
        if value.is_zero() {
            return Err(error::fmt!(
                ConfigError,
                "\"reconnect_initial_backoff_millis\" must be greater than 0."
            ));
        }
        let Some(qwp_ws) = &mut self.qwp_ws else {
            return Err(error::fmt!(
                ConfigError,
                "The \"reconnect_initial_backoff_millis\" setting is only supported for QWP/WebSocket."
            ));
        };
        qwp_ws
            .reconnect_initial_backoff
            .set_specified("reconnect_initial_backoff_millis", value)?;
        Ok(self)
    }

    #[cfg(feature = "_sender-qwp-ws")]
    /// Cap on the reconnect backoff the retry loop doubles toward; the actual
    /// per-attempt delay is this value jittered to ~[half, 1.5x]. Default 5s.
    pub fn reconnect_max_backoff(mut self, value: Duration) -> Result<Self> {
        if value.is_zero() {
            return Err(error::fmt!(
                ConfigError,
                "\"reconnect_max_backoff_millis\" must be greater than 0."
            ));
        }
        let Some(qwp_ws) = &mut self.qwp_ws else {
            return Err(error::fmt!(
                ConfigError,
                "The \"reconnect_max_backoff_millis\" setting is only supported for QWP/WebSocket."
            ));
        };
        qwp_ws
            .reconnect_max_backoff
            .set_specified("reconnect_max_backoff_millis", value)?;
        Ok(self)
    }

    #[cfg(feature = "_sender-qwp-ws")]
    fn qwp_ws_endpoints(mut self, endpoints: Vec<conf::QwpWsEndpoint>) -> Result<Self> {
        let Some(qwp_ws) = &mut self.qwp_ws else {
            return Err(error::fmt!(
                ConfigError,
                "QWP/WebSocket endpoint lists are only supported for QWP/WebSocket."
            ));
        };
        qwp_ws.endpoints.set_specified("addr", endpoints)?;
        Ok(self)
    }

    #[cfg(feature = "_sender-qwp-ws")]
    fn qwp_ws_initial_connect_mode(mut self, mode: conf::QwpWsInitialConnectMode) -> Result<Self> {
        let Some(qwp_ws) = &mut self.qwp_ws else {
            return Err(error::fmt!(
                ConfigError,
                "The \"initial_connect_retry\" setting is only supported for QWP/WebSocket."
            ));
        };
        qwp_ws
            .initial_connect_retry
            .set_specified("initial_connect_retry", mode)?;
        Ok(self)
    }

    #[cfg(feature = "_sender-qwp-ws")]
    /// Retry the initial connection using the reconnect policy. Default false.
    ///
    /// The mode also governs the pool ([`crate::QuestDb::connect`]): the
    /// eager warm-minimum pre-open and every pool borrow that opens a new
    /// connection honor it, defaulting to fail-fast `off`. A `lazy_connect`
    /// pool always connects in the background and rejects an explicit
    /// blocking mode. Reconnect-to-sync promotion applies only to standalone
    /// [`SenderBuilder::build`]; pools honor only an explicitly set mode.
    pub fn initial_connect_retry(mut self, value: bool) -> Result<Self> {
        let Some(qwp_ws) = &mut self.qwp_ws else {
            return Err(error::fmt!(
                ConfigError,
                "The \"initial_connect_retry\" setting is only supported for QWP/WebSocket."
            ));
        };
        qwp_ws.initial_connect_retry.set_specified(
            "initial_connect_retry",
            if value {
                conf::QwpWsInitialConnectMode::Sync
            } else {
                conf::QwpWsInitialConnectMode::Off
            },
        )?;
        Ok(self)
    }

    #[cfg(feature = "_sender-qwp-ws")]
    fn qwp_ws_auth_timeout_millis(mut self, value: &str) -> Result<Self> {
        let Some(qwp_ws) = &mut self.qwp_ws else {
            return Err(error::fmt!(
                ConfigError,
                "The \"auth_timeout_ms\" setting is only supported for QWP/WebSocket."
            ));
        };
        let millis: i64 = parse_conf_value("auth_timeout_ms", value)?;
        if millis <= 0 {
            return Err(error::fmt!(
                ConfigError,
                "auth_timeout_ms must be > 0: {}",
                millis
            ));
        }
        qwp_ws
            .auth_timeout
            .set_specified("auth_timeout_ms", Duration::from_millis(millis as u64))?;
        Ok(self)
    }

    #[cfg(feature = "_sender-qwp-ws")]
    fn qwp_ws_connect_timeout_millis(mut self, value: &str) -> Result<Self> {
        let Some(qwp_ws) = &mut self.qwp_ws else {
            return Err(error::fmt!(
                ConfigError,
                "The \"connect_timeout\" setting is only supported for QWP/WebSocket."
            ));
        };
        let millis: i64 = parse_conf_value("connect_timeout", value)?;
        if millis <= 0 {
            return Err(error::fmt!(
                ConfigError,
                "connect_timeout must be > 0: {}",
                millis
            ));
        }
        qwp_ws.connect_timeout.set_specified(
            "connect_timeout",
            Some(Duration::from_millis(millis as u64)),
        )?;
        Ok(self)
    }

    #[cfg(feature = "_sender-qwp-ws")]
    fn close_flush_timeout_millis(mut self, value: &str) -> Result<Self> {
        let Some(qwp_ws) = &mut self.qwp_ws else {
            return Err(error::fmt!(
                ConfigError,
                "The \"close_flush_timeout_millis\" setting is only supported for QWP/WebSocket."
            ));
        };
        let millis: i64 = parse_conf_value("close_flush_timeout_millis", value)?;
        let timeout = if millis <= 0 {
            Duration::ZERO
        } else {
            Duration::from_millis(millis as u64)
        };
        qwp_ws
            .close_flush_timeout
            .set_specified("close_flush_timeout_millis", timeout)?;
        Ok(self)
    }

    #[cfg(feature = "_sender-qwp-ws")]
    fn request_durable_ack(mut self, value: &str) -> Result<Self> {
        let Some(qwp_ws) = &mut self.qwp_ws else {
            return Err(error::fmt!(
                ConfigError,
                "The \"request_durable_ack\" setting is only supported for QWP/WebSocket."
            ));
        };
        if value.eq_ignore_ascii_case("off") {
            qwp_ws
                .request_durable_ack
                .set_specified("request_durable_ack", false)?;
            return Ok(self);
        }
        if value.eq_ignore_ascii_case("on") {
            qwp_ws
                .request_durable_ack
                .set_specified("request_durable_ack", true)?;
            return Ok(self);
        }

        Err(error::fmt!(
            ConfigError,
            "invalid request_durable_ack [value={value}, allowed-values=[on, off]]"
        ))
    }

    #[cfg(feature = "_sender-qwp-ws")]
    fn drain_orphans(mut self, value: &str) -> Result<Self> {
        let Some(qwp_ws) = &mut self.qwp_ws else {
            return Err(error::fmt!(
                ConfigError,
                "The \"drain_orphans\" setting is only supported for QWP/WebSocket."
            ));
        };
        if value.eq_ignore_ascii_case("off") || value.eq_ignore_ascii_case("false") {
            qwp_ws.drain_orphans.set_specified("drain_orphans", false)?;
            return Ok(self);
        }
        if value.eq_ignore_ascii_case("on") || value.eq_ignore_ascii_case("true") {
            qwp_ws.drain_orphans.set_specified("drain_orphans", true)?;
            return Ok(self);
        }

        Err(error::fmt!(
            ConfigError,
            "invalid drain_orphans [value={value}, allowed-values=[on, off, true, false]]"
        ))
    }

    #[cfg(feature = "_sender-qwp-ws")]
    fn durable_ack_keepalive_interval_millis(mut self, value: &str) -> Result<Self> {
        let Some(qwp_ws) = &mut self.qwp_ws else {
            return Err(error::fmt!(
                ConfigError,
                "The \"durable_ack_keepalive_interval_millis\" setting is only supported for QWP/WebSocket."
            ));
        };
        let millis: i64 = parse_conf_value("durable_ack_keepalive_interval_millis", value)?;
        let interval = if millis <= 0 {
            Duration::ZERO
        } else {
            Duration::from_millis(millis as u64)
        };
        qwp_ws
            .durable_ack_keepalive_interval
            .set_specified("durable_ack_keepalive_interval_millis", interval)?;
        Ok(self)
    }

    #[cfg(feature = "_sender-qwp-ws")]
    fn max_background_drainers(mut self, value: &str) -> Result<Self> {
        let Some(qwp_ws) = &mut self.qwp_ws else {
            return Err(error::fmt!(
                ConfigError,
                "The \"max_background_drainers\" setting is only supported for QWP/WebSocket."
            ));
        };
        let value: i32 = parse_conf_value("max_background_drainers", value)?;
        if value < 0 {
            return Err(error::fmt!(
                ConfigError,
                "max_background_drainers must be >= 0: {value}"
            ));
        }
        qwp_ws
            .max_background_drainers
            .set_specified("max_background_drainers", value as usize)?;
        Ok(self)
    }

    #[cfg(feature = "_sender-qwp-ws")]
    fn error_inbox_capacity(mut self, value: &str) -> Result<Self> {
        let Some(qwp_ws) = &mut self.qwp_ws else {
            return Err(error::fmt!(
                ConfigError,
                "The \"error_inbox_capacity\" setting is only supported for QWP/WebSocket."
            ));
        };
        let value: usize = parse_conf_value("error_inbox_capacity", value)?;
        if value < conf::QWP_WS_MIN_ERROR_INBOX_CAPACITY {
            return Err(error::fmt!(
                ConfigError,
                "error_inbox_capacity must be >= {}: {value}",
                conf::QWP_WS_MIN_ERROR_INBOX_CAPACITY
            ));
        }
        qwp_ws
            .error_inbox_capacity
            .set_specified("error_inbox_capacity", value)?;
        Ok(self)
    }

    /// Configure how long to wait for messages from the QuestDB server during
    /// the TLS handshake and authentication process. For QWP/WebSocket this
    /// bounds only the HTTP upgrade response read. The default is 15 seconds.
    pub fn auth_timeout(mut self, value: Duration) -> Result<Self> {
        #[cfg(feature = "_sender-qwp-udp")]
        self.reject_if_qwp_udp("auth_timeout")?;
        #[cfg(feature = "_sender-qwp-ws")]
        if let Some(qwp_ws) = &mut self.qwp_ws {
            if value.is_zero() {
                return Err(error::fmt!(
                    ConfigError,
                    "\"auth_timeout\" must be greater than 0."
                ));
            }
            qwp_ws.auth_timeout.set_specified("auth_timeout", value)?;
            return Ok(self);
        }
        self.auth_timeout.set_specified("auth_timeout", value)?;
        Ok(self)
    }

    #[cfg(feature = "_sender-qwp-udp")]
    fn reject_if_qwp_udp(&self, setting: &str) -> Result<()> {
        if self.protocol.is_qwp_udp() {
            return Err(error::fmt!(
                ConfigError,
                "The \"{setting}\" setting is not supported for QWP/UDP."
            ));
        }
        Ok(())
    }

    /// Ensure that TLS is enabled for the protocol.
    pub fn ensure_tls_enabled(&self, property: &str) -> Result<()> {
        if !self.protocol.tls_enabled() {
            return Err(error::fmt!(
                ConfigError,
                "Cannot set {property:?}: TLS is not supported for protocol {}",
                self.protocol
            ));
        }
        Ok(())
    }

    /// Set to `false` to disable TLS certificate verification.
    /// This should only be used for debugging purposes as it reduces security.
    ///
    /// For testing, consider specifying a path to a `.pem` file instead via
    /// the [`tls_roots`](SenderBuilder::tls_roots) method.
    #[cfg(feature = "insecure-skip-verify")]
    pub fn tls_verify(mut self, verify: bool) -> Result<Self> {
        self.ensure_tls_enabled("tls_verify")?;
        self.tls_verify.set_specified("tls_verify", verify)?;
        Ok(self)
    }

    /// Specify where to find the root certificate used to validate the
    /// server's TLS certificate.
    pub fn tls_ca(mut self, ca: CertificateAuthority) -> Result<Self> {
        self.ensure_tls_enabled("tls_ca")?;
        self.tls_ca.set_specified("tls_ca", ca)?;
        Ok(self)
    }

    /// Set the path to a custom root certificate `.pem` file.
    /// This is used to validate the server's certificate during the TLS handshake.
    ///
    /// On QWP/WebSocket (`ws::` / `wss::`) the same path key
    /// also accepts a JKS or PKCS#12 keystore — see
    /// [`tls_roots_password`](SenderBuilder::tls_roots_password) for
    /// the unlock password.
    ///
    /// See notes on how to test with [self-signed
    /// certificates](https://github.com/questdb/c-questdb-client/tree/main/tls_certs).
    pub fn tls_roots<P: Into<PathBuf>>(self, path: P) -> Result<Self> {
        let mut builder = self.tls_ca(CertificateAuthority::PemFile)?;
        let path = path.into();
        // Attempt to read the file here to catch any issues early.
        let _file = std::fs::File::open(&path).map_err(|io_err| {
            error::fmt!(
                ConfigError,
                "Could not open root certificate file from path {:?}: {}",
                path,
                io_err
            )
        })?;
        builder.tls_roots.set_specified("tls_roots", Some(path))?;
        Ok(builder)
    }

    /// Set the password unlocking the JKS / PKCS#12 keystore named by
    /// [`tls_roots`](SenderBuilder::tls_roots). QWP/WebSocket only —
    /// other transports keep PEM as the sole `tls_roots` format.
    ///
    /// With this set, the `tls_roots` file is read as a Java
    /// KeyStore (auto-detected: JKS magic `0xFEEDFEED`, or PKCS#12
    /// ASN.1 SEQUENCE) and trusted-certificate entries become the
    /// rustls root store. Mirrors the Java reference client's
    /// `tls_roots_password` connect-string key.
    #[cfg(feature = "_sender-qwp-ws")]
    pub fn tls_roots_password<S: Into<String>>(mut self, password: S) -> Result<Self> {
        if !self.protocol.is_qwp_ws() {
            return Err(error::fmt!(
                ConfigError,
                "\"tls_roots_password\" is only supported for QWP/WebSocket \
                 (ws / wss). ILP/TCP and ILP/HTTP transports read \
                 unencrypted PEM via rustls."
            ));
        }
        self.ensure_tls_enabled("tls_roots_password")?;
        self.tls_roots_password
            .set_specified("tls_roots_password", Some(password.into()))?;
        Ok(self)
    }

    /// The initial buffered size that the client will pre-allocate for new
    /// [`Buffer`] instances returned by [`Sender::new_buffer`].
    /// The default is 64 KiB.
    ///
    /// For ILP / HTTP this pre-allocates the underlying byte vector to this
    /// size; the buffer then grows up to [`Self::max_buf_size`].
    /// For QWP/WebSocket the value is accepted and cross-validated against
    /// `max_buf_size`, but no flat byte buffer exists to pre-allocate
    /// — the columnar buffer allocates per-table on first row.
    /// For QWP/UDP the value is accepted but has no effect: datagrams are
    /// bounded by `max_datagram_size`.
    pub fn init_buf_size(mut self, value: usize) -> Result<Self> {
        let min = 1024;
        if value < min {
            return Err(error::fmt!(
                ConfigError,
                "\"init_buf_size\" must be at least {min} bytes."
            ));
        }
        self.init_buf_size.set_specified("init_buf_size", value)?;
        Ok(self)
    }

    /// The maximum buffered size that the client will flush to the server.
    /// The default is 100 MiB.
    ///
    /// For ILP this applies to the exact pending byte length.
    /// For QWP/UDP this applies to the buffer size hint exposed by [`Buffer::len`].
    /// For QWP/WebSocket this applies to the encoded replay message size.
    pub fn max_buf_size(mut self, value: usize) -> Result<Self> {
        let min = 1024;
        if value < min {
            return Err(error::fmt!(
                ConfigError,
                "max_buf_size\" must be at least {min} bytes."
            ));
        }
        self.max_buf_size.set_specified("max_buf_size", value)?;
        Ok(self)
    }

    /// The maximum length of a table or column name in bytes.
    /// Matches the `cairo.max.file.name.length` setting in the server.
    /// The default is 127 bytes.
    /// If running over HTTP and protocol version 2 is auto-negotiated, this
    /// value is picked up from the server.
    pub fn max_name_len(mut self, value: usize) -> Result<Self> {
        if value < 16 {
            return Err(error::fmt!(
                ConfigError,
                "max_name_len must be at least 16 bytes."
            ));
        }
        self.max_name_len.set_specified("max_name_len", value)?;
        Ok(self)
    }

    // Only consumed by the QWP/WebSocket-gated pool builder in `db.rs`.
    #[cfg(feature = "sync-sender-qwp-ws")]
    pub(crate) fn configured_max_name_len(&self) -> usize {
        *self.max_name_len
    }

    #[cfg(feature = "sync-sender-http")]
    /// Set the cumulative duration spent in retries.
    /// The value is in milliseconds, and the default is 10 seconds.
    pub fn retry_timeout(mut self, value: Duration) -> Result<Self> {
        if let Some(http) = &mut self.http {
            http.retry_timeout.set_specified("retry_timeout", value)?;
        } else {
            return Err(error::fmt!(
                ConfigError,
                "retry_timeout is supported only in ILP over HTTP."
            ));
        }
        Ok(self)
    }

    #[cfg(feature = "sync-sender-http")]
    /// Cap on per-attempt backoff in the HTTP retry loop.
    ///
    /// The retry loop starts at 10 ms, doubles each attempt with ±5 ms
    /// jitter, and is bounded by this value (default: 1 second; minimum
    /// 10 ms — a cap below the initial interval is incoherent). Total
    /// retry budget is independently capped by
    /// [`SenderBuilder::retry_timeout`]; this knob shapes how aggressively
    /// the loop hits the server while waiting out a transient failure.
    ///
    /// Mirrors Java's `LineSenderBuilder.maxBackoffMillis(int)`.
    pub fn retry_max_backoff(mut self, value: Duration) -> Result<Self> {
        if value < Duration::from_millis(10) {
            return Err(error::fmt!(
                ConfigError,
                "\"retry_max_backoff_millis\" must be at least 10."
            ));
        }
        if let Some(http) = &mut self.http {
            http.retry_max_backoff
                .set_specified("retry_max_backoff_millis", value)?;
        } else {
            return Err(error::fmt!(
                ConfigError,
                "retry_max_backoff_millis is supported only in ILP over HTTP."
            ));
        }
        Ok(self)
    }

    #[cfg(feature = "sync-sender-http")]
    /// Set the minimum acceptable throughput while sending a buffer to the server.
    /// The sender will divide the payload size by this number to determine for how
    /// long to keep sending the payload before timing out.
    /// The value is in bytes per second, and the default is 100 KiB/s.
    /// The timeout calculated from minimum throughput is adedd to the value of
    /// [`request_timeout`](SenderBuilder::request_timeout) to get the total timeout
    /// value.
    /// A value of 0 disables this feature, so it's similar to setting "infinite"
    /// minimum throughput. The total timeout will then be equal to `request_timeout`.
    pub fn request_min_throughput(mut self, value: u64) -> Result<Self> {
        if let Some(http) = &mut self.http {
            http.request_min_throughput
                .set_specified("request_min_throughput", value)?;
        } else {
            return Err(error::fmt!(
                ConfigError,
                "\"request_min_throughput\" is supported only in ILP over HTTP."
            ));
        }
        Ok(self)
    }

    #[cfg(feature = "sync-sender-http")]
    /// Additional time to wait on top of that calculated from the minimum throughput.
    /// This accounts for the fixed latency of the HTTP request-response roundtrip.
    /// The default is 10 seconds.
    /// See also: [`request_min_throughput`](SenderBuilder::request_min_throughput).
    pub fn request_timeout(mut self, value: Duration) -> Result<Self> {
        if let Some(http) = &mut self.http {
            if value.is_zero() {
                return Err(error::fmt!(
                    ConfigError,
                    "\"request_timeout\" must be greater than 0."
                ));
            }
            http.request_timeout
                .set_specified("request_timeout", value)?;
        } else {
            return Err(error::fmt!(
                ConfigError,
                "\"request_timeout\" is supported only in ILP over HTTP."
            ));
        }
        Ok(self)
    }

    #[cfg(feature = "sync-sender-http")]
    /// Internal API, do not use.
    /// This is exposed exclusively for the Python client.
    /// We (QuestDB) use this to help us debug which client is being used if we encounter issues.
    #[doc(hidden)]
    pub fn user_agent(mut self, value: &str) -> Result<Self> {
        let value = validate_value(value)?;
        if let Some(http) = &mut self.http {
            http.user_agent = value.to_string();
        }
        Ok(self)
    }

    fn build_auth(&self) -> Result<Option<conf::AuthParams>> {
        match (
            self.protocol,
            self.username.deref(),
            self.password.deref(),
            self.token.deref(),
            #[cfg(feature = "_sender-tcp")]
            self.token_x.deref(),
            #[cfg(not(feature = "_sender-tcp"))]
            None::<String>,
            #[cfg(feature = "_sender-tcp")]
            self.token_y.deref(),
            #[cfg(not(feature = "_sender-tcp"))]
            None::<String>,
        ) {
            (_, None, None, None, None, None) => Ok(None),

            #[cfg(feature = "_sender-tcp")]
            (protocol, Some(username), None, Some(token), Some(token_x), Some(token_y))
                if protocol.is_tcpx() =>
            {
                Ok(Some(conf::AuthParams::Ecdsa(conf::EcdsaAuthParams {
                    key_id: username.to_string(),
                    priv_key: token.to_string(),
                    pub_key_x: token_x.to_string(),
                    pub_key_y: token_y.to_string(),
                })))
            }

            #[cfg(feature = "_sender-tcp")]
            (protocol, Some(_username), Some(_password), None, None, None)
                if protocol.is_tcpx() =>
            {
                Err(error::fmt!(
                    ConfigError,
                    r##"The "basic_auth" setting can only be used with the ILP/HTTP protocol."##,
                ))
            }

            #[cfg(feature = "_sender-tcp")]
            (protocol, None, None, Some(_token), None, None) if protocol.is_tcpx() => {
                Err(error::fmt!(
                    ConfigError,
                    "Token authentication only be used with the ILP/HTTP protocol."
                ))
            }

            #[cfg(feature = "_sender-tcp")]
            (protocol, _username, None, _token, _token_x, _token_y) if protocol.is_tcpx() => {
                Err(error::fmt!(
                    ConfigError,
                    r##"Incomplete ECDSA authentication parameters. Specify either all or none of: "username", "token", "token_x", "token_y"."##,
                ))
            }
            #[cfg(any(feature = "_sender-http", feature = "_sender-qwp-ws"))]
            (protocol, Some(username), Some(password), None, None, None)
                if protocol.accepts_http_auth() =>
            {
                Ok(Some(conf::AuthParams::Basic(conf::BasicAuthParams {
                    username: username.to_string(),
                    password: password.to_string(),
                })))
            }
            #[cfg(any(feature = "_sender-http", feature = "_sender-qwp-ws"))]
            (protocol, Some(_username), None, None, None, None) if protocol.accepts_http_auth() => {
                Err(error::fmt!(
                    ConfigError,
                    r##"Basic authentication parameter "username" is present, but "password" is missing."##,
                ))
            }
            #[cfg(any(feature = "_sender-http", feature = "_sender-qwp-ws"))]
            (protocol, None, Some(_password), None, None, None) if protocol.accepts_http_auth() => {
                Err(error::fmt!(
                    ConfigError,
                    r##"Basic authentication parameter "password" is present, but "username" is missing."##,
                ))
            }
            #[cfg(any(feature = "_sender-http", feature = "_sender-qwp-ws"))]
            (protocol, None, None, Some(token), None, None) if protocol.accepts_http_auth() => {
                Ok(Some(conf::AuthParams::Token(conf::TokenAuthParams {
                    token: token.to_string(),
                })))
            }
            #[cfg(feature = "_sender-http")]
            (protocol, Some(_username), None, Some(_token), Some(_token_x), Some(_token_y))
                if protocol.is_httpx() =>
            {
                Err(error::fmt!(
                    ConfigError,
                    "ECDSA authentication is only available with ILP/TCP and not available with ILP/HTTP."
                ))
            }
            #[cfg(feature = "_sender-http")]
            (protocol, _username, _password, _token, None, None) if protocol.is_httpx() => {
                Err(error::fmt!(
                    ConfigError,
                    r##"Inconsistent HTTP authentication parameters. Specify either "username" and "password", or just "token"."##,
                ))
            }
            _ => Err(error::fmt!(
                ConfigError,
                r##"Incomplete authentication parameters. Check "username", "password", "token", "token_x" and "token_y" parameters are set correctly."##,
            )),
        }
    }

    #[cfg(feature = "_sync-sender")]
    /// Build the sender.
    ///
    /// In the case of TCP, this synchronously establishes the TCP connection, and
    /// returns once the connection is fully established. If the connection
    /// requires authentication or TLS, these will also be completed before
    /// returning.
    pub fn build(&self) -> Result<Sender> {
        // Fail fast on misconfigured buffer sizes before opening any sockets.
        // Only enforce the init-vs-max relationship when the user explicitly
        // set init_buf_size; a defaulted init_buf_size silently clamps to
        // max_buf_size below.
        if self.init_buf_size.is_specified() && *self.init_buf_size > *self.max_buf_size {
            return Err(error::fmt!(
                ConfigError,
                "init_buf_size ({}) cannot exceed max_buf_size ({})",
                *self.init_buf_size,
                *self.max_buf_size
            ));
        }

        let mut descr = format!("Sender[host={:?},port={:?},", self.host, self.port);

        if self.protocol.tls_enabled() {
            write!(descr, "tls=enabled,").unwrap();
        } else {
            write!(descr, "tls=disabled,").unwrap();
        }

        #[cfg(feature = "insecure-skip-verify")]
        let tls_verify = *self.tls_verify;

        #[cfg(feature = "_sender-qwp-ws")]
        let tls_roots_password = self.tls_roots_password.deref().as_deref();
        #[cfg(not(feature = "_sender-qwp-ws"))]
        let tls_roots_password: Option<&str> = None;

        // Pair validation: the password unlocks the keystore at
        // `tls_roots`. Without `tls_roots`, the password names no
        // file, so the trust source falls back to the default — not
        // what the caller asked for. Java enforces the same pairing.
        if tls_roots_password.is_some() && self.tls_roots.deref().is_none() {
            return Err(error::fmt!(
                ConfigError,
                "\"tls_roots_password\" requires \"tls_roots\" \
                 (the password unlocks the keystore at that path)"
            ));
        }

        #[allow(unused_variables)]
        let tls_settings = tls::TlsSettings::build(
            self.protocol.tls_enabled(),
            #[cfg(feature = "insecure-skip-verify")]
            tls_verify,
            *self.tls_ca,
            self.tls_roots.deref().as_deref(),
            tls_roots_password,
        )?;

        let auth = self.build_auth()?;

        let handler = match self.protocol {
            #[cfg(feature = "sync-sender-tcp")]
            Protocol::Tcp | Protocol::Tcps => connect_tcp(
                self.host.as_str(),
                self.port.as_str(),
                self.net_interface.deref().as_deref(),
                *self.auth_timeout,
                tls_settings,
                &auth,
            )?,
            #[cfg(feature = "sync-sender-http")]
            Protocol::Http | Protocol::Https => {
                use ureq::unversioned::transport::Connector;
                use ureq::unversioned::transport::TcpConnector;
                if self.net_interface.is_some() {
                    // See: https://github.com/algesten/ureq/issues/692
                    return Err(error::fmt!(
                        InvalidApiCall,
                        "net_interface is not supported for ILP over HTTP."
                    ));
                }

                let http_config = self.http.as_ref().unwrap();
                let user_agent = http_config.user_agent.as_str();
                let connector = TcpConnector::default();

                let agent_builder = ureq::Agent::config_builder()
                    .user_agent(user_agent)
                    .no_delay(true);

                let tls_config = match tls_settings {
                    Some(tls_settings) => Some(tls::configure_tls(tls_settings)?),
                    None => None,
                };

                let connector = connector.chain(TlsConnector::new(tls_config));

                let auth = match auth {
                    Some(conf::AuthParams::Basic(ref auth)) => Some(auth.to_header_string()),
                    Some(conf::AuthParams::Token(ref auth)) => Some(auth.to_header_string()?),

                    #[cfg(feature = "sync-sender-tcp")]
                    Some(conf::AuthParams::Ecdsa(_)) => {
                        return Err(fmt!(
                            AuthError,
                            "ECDSA authentication is not supported for ILP over HTTP. \
                            Please use basic or token authentication instead."
                        ));
                    }
                    None => None,
                };
                let agent_builder = agent_builder
                    .timeout_connect(Some(*http_config.request_timeout.deref()))
                    .http_status_as_error(false);
                let agent = ureq::Agent::with_parts(
                    agent_builder.build(),
                    connector,
                    ureq::unversioned::resolver::DefaultResolver::default(),
                );
                let proto = self.protocol.schema();
                let url = format!(
                    "{}://{}:{}/write",
                    proto,
                    self.host.deref(),
                    self.port.deref()
                );
                SyncProtocolHandler::SyncHttp(SyncHttpHandlerState {
                    agent,
                    url,
                    auth,
                    config: self.http.as_ref().unwrap().clone(),
                })
            }
            #[cfg(feature = "sync-sender-qwp-udp")]
            Protocol::Udp => {
                let Some(qwp_udp) = self.qwp_udp.as_ref() else {
                    return Err(error::fmt!(
                        ConfigError,
                        "QWP/UDP configuration is missing."
                    ));
                };
                connect_qwp_udp(
                    self.host.as_str(),
                    self.port.as_str(),
                    self.net_interface.deref().as_deref(),
                    qwp_udp,
                )?
            }
            #[cfg(feature = "sync-sender-qwp-ws")]
            Protocol::Ws | Protocol::Wss => {
                if self.net_interface.is_some() {
                    return Err(error::fmt!(
                        InvalidApiCall,
                        "net_interface is not supported for QWP over WebSocket."
                    ));
                }
                let Some(qwp_ws) = self.qwp_ws.as_ref() else {
                    return Err(error::fmt!(
                        ConfigError,
                        "QWP/WebSocket configuration is missing."
                    ));
                };
                // Resolve reconnect-implies-initial-retry only for this
                // standalone build. The builder retains the user's explicit
                // choice (or lack of one), so pool connector builds never see
                // this effective mode.
                let actual_initial_connect_retry = qwp_ws.resolve_initial_connect_retry();
                let mut qwp_ws = qwp_ws.clone();
                qwp_ws.initial_connect_retry =
                    ConfigSetting::Specified(actual_initial_connect_retry);
                let qwp_ws = &qwp_ws;
                reject_unsupported_qwp_ws_sf_config(qwp_ws)?;
                let basic_auth = qwp_ws_auth_header(&auth)?;
                if *qwp_ws.progress == QwpWsProgress::Manual {
                    if *qwp_ws.initial_connect_retry == conf::QwpWsInitialConnectMode::Async {
                        return Err(error::fmt!(
                            ConfigError,
                            "initial_connect_retry=async requires QWP/WebSocket background progress; use qwp_ws_progress=background or initial_connect_retry=sync"
                        ));
                    }
                    SyncProtocolHandler::ManualQwpWs(Box::new(open_manual_qwp_ws(
                        self.host.as_str(),
                        self.port.as_str(),
                        matches!(self.protocol, Protocol::Wss),
                        tls_settings,
                        qwp_ws,
                        basic_auth,
                    )?))
                } else {
                    connect_qwp_ws(
                        self.host.as_str(),
                        self.port.as_str(),
                        matches!(self.protocol, Protocol::Wss),
                        tls_settings,
                        qwp_ws,
                        basic_auth,
                    )?
                }
            }
        };

        #[allow(unused_mut)]
        let mut max_name_len = *self.max_name_len;

        let protocol_version = match self.protocol_version.deref() {
            Some(v) => *v,
            None => match self.protocol {
                #[cfg(feature = "sync-sender-tcp")]
                Protocol::Tcp | Protocol::Tcps => ProtocolVersion::V1,
                #[cfg(feature = "sync-sender-http")]
                Protocol::Http | Protocol::Https => {
                    #[allow(irrefutable_let_patterns)]
                    if let SyncProtocolHandler::SyncHttp(http_state) = &handler {
                        let settings_url = &format!(
                            "{}://{}:{}/settings",
                            self.protocol.schema(),
                            self.host.deref(),
                            self.port.deref()
                        );
                        let (protocol_versions, server_max_name_len) =
                            read_server_settings(http_state, settings_url, max_name_len)?;
                        max_name_len = server_max_name_len;
                        SUPPORTED_PROTOCOL_VERSIONS
                            .iter()
                            .find(|version| protocol_versions.contains(version))
                            .copied()
                            .ok_or_else(|| {
                                fmt!(
                                    ProtocolVersionError,
                                    "Server does not support any of the client protocol versions: {:?}",
                                    SUPPORTED_PROTOCOL_VERSIONS
                                )
                            })?
                    } else {
                        unreachable!("HTTP handler should be used for HTTP protocol");
                    }
                }
                #[cfg(feature = "sync-sender-qwp-udp")]
                Protocol::Udp => ProtocolVersion::V1,
                #[cfg(feature = "sync-sender-qwp-ws")]
                Protocol::Ws | Protocol::Wss => ProtocolVersion::V1,
            },
        };

        if auth.is_some() {
            descr.push_str("auth=on]");
        } else {
            descr.push_str("auth=off]");
        }

        // Defaulted init_buf_size clamps to max_buf_size when the cap is
        // smaller. The explicit-init-too-big check fires at the top of
        // build(); reaching here means init_buf_size is in range.
        let effective_init_buf_size = (*self.init_buf_size).min(*self.max_buf_size);

        let sender = Sender::new(
            descr,
            handler,
            effective_init_buf_size,
            *self.max_buf_size,
            self.protocol,
            protocol_version,
            max_name_len,
            #[cfg(feature = "_sender-qwp-ws")]
            self.qwp_ws_error_handler.clone(),
            #[cfg(feature = "_sender-qwp-ws")]
            self.qwp_ws
                .as_ref()
                .and_then(|qwp_ws| qwp_ws.conn_events.clone()),
        );

        Ok(sender)
    }

    /// Resolve the QWP/WebSocket connect ingredients used by
    /// [`Self::build_qwp_ws_connector`]: validate the protocol / buffer / TLS
    /// settings, build the TLS config and auth header, and clone the
    /// SF-vetted `QwpWsConfig`.
    #[cfg(feature = "sync-sender-qwp-ws")]
    fn resolve_qwp_ws_ingredients(
        &self,
    ) -> Result<(
        bool,
        Option<tls::TlsSettings>,
        conf::QwpWsConfig,
        Option<String>,
    )> {
        if self.init_buf_size.is_specified() && *self.init_buf_size > *self.max_buf_size {
            return Err(error::fmt!(
                ConfigError,
                "init_buf_size ({}) cannot exceed max_buf_size ({})",
                *self.init_buf_size,
                *self.max_buf_size
            ));
        }

        if !matches!(self.protocol, Protocol::Ws | Protocol::Wss) {
            return Err(error::fmt!(
                ConfigError,
                "Column-sender requires a QWP/WebSocket connect string \
                 (got protocol {:?})",
                self.protocol
            ));
        }
        if self.net_interface.is_some() {
            return Err(error::fmt!(
                InvalidApiCall,
                "net_interface is not supported for QWP over WebSocket."
            ));
        }
        let Some(qwp_ws) = self.qwp_ws.as_ref() else {
            return Err(error::fmt!(
                ConfigError,
                "QWP/WebSocket configuration is missing."
            ));
        };

        #[cfg(feature = "insecure-skip-verify")]
        let tls_verify = *self.tls_verify;
        let tls_roots_password = self.tls_roots_password.deref().as_deref();

        if tls_roots_password.is_some() && self.tls_roots.deref().is_none() {
            return Err(error::fmt!(
                ConfigError,
                "\"tls_roots_password\" requires \"tls_roots\" \
                 (the password unlocks the keystore at that path)"
            ));
        }

        let tls_settings = tls::TlsSettings::build(
            self.protocol.tls_enabled(),
            #[cfg(feature = "insecure-skip-verify")]
            tls_verify,
            *self.tls_ca,
            self.tls_roots.deref().as_deref(),
            tls_roots_password,
        )?;

        let auth = self.build_auth()?;
        let auth_header = qwp_ws_auth_header(&auth)?;
        let qwp_ws = qwp_ws.clone();
        reject_unsupported_qwp_ws_sf_config(&qwp_ws)?;
        if *qwp_ws.progress == QwpWsProgress::Manual
            && *qwp_ws.initial_connect_retry == conf::QwpWsInitialConnectMode::Async
        {
            return Err(error::fmt!(
                ConfigError,
                "initial_connect_retry=async requires QWP/WebSocket background progress; use qwp_ws_progress=background or initial_connect_retry=sync"
            ));
        }

        let use_tls = matches!(self.protocol, Protocol::Wss);
        Ok((use_tls, tls_settings, qwp_ws, auth_header))
    }

    /// Force the pool connector's baked-in initial connect mode to background.
    #[cfg(feature = "sync-sender-qwp-ws")]
    pub(crate) fn force_async_initial_connect(&mut self) {
        if let Some(qwp_ws) = self.qwp_ws.as_mut() {
            qwp_ws.force_async_initial_connect();
        }
    }

    /// Build a reusable [`QwpWsConnector`] capturing the full configured
    /// endpoint list — the pooled QWP ingress path's entry point into the
    /// network. The pool drives it through a shared health tracker so each
    /// connect rotates across endpoints, skips unhealthy ones, and follows
    /// the writable primary on a role reject; the resulting `WsStream` does
    /// its own synchronous frame I/O and does not use the standalone
    /// [`Sender`]'s replay encoder or transaction ownership.
    #[cfg(feature = "sync-sender-qwp-ws")]
    pub(crate) fn build_qwp_ws_connector(&self) -> Result<QwpWsConnector> {
        let (use_tls, tls_settings, qwp_ws, auth_header) = self.resolve_qwp_ws_ingredients()?;
        let endpoints = sender::qwp_ws::qwp_ws_configured_endpoints(
            self.host.as_str(),
            self.port.as_str(),
            &qwp_ws,
        );
        Ok(QwpWsConnector {
            host: self.host.to_string(),
            port: self.port.to_string(),
            endpoints,
            use_tls,
            tls_settings,
            qwp_ws,
            auth_header,
            max_buf_size: *self.max_buf_size,
        })
    }

    #[cfg(any(feature = "_sender-tcp", feature = "_sender-qwp-udp"))]
    fn ensure_supports_bind_interface(&self, param_name: &str) -> Result<()> {
        #[cfg(feature = "_sender-tcp")]
        if self.protocol.is_tcpx() {
            return Ok(());
        }

        #[cfg(feature = "_sender-qwp-udp")]
        if self.protocol.is_qwp_udp() {
            return Ok(());
        }

        #[cfg(feature = "_sender-qwp-udp")]
        let supported = "TCP or QWP/UDP";
        #[cfg(not(feature = "_sender-qwp-udp"))]
        let supported = "TCP";

        Err(fmt!(
            ConfigError,
            "The {param_name:?} setting can only be used with the {supported} protocol."
        ))
    }
}

/// When parsing from config, we exclude certain characters.
/// Here we repeat the same validation logic for consistency.
#[cfg(feature = "_sync-sender")]
fn validate_value<T: AsRef<str>>(value: T) -> Result<T> {
    let str_ref = value.as_ref();
    for (p, c) in str_ref.chars().enumerate() {
        if matches!(c, '\u{0}'..='\u{1f}' | '\u{7f}'..='\u{9f}') {
            return Err(error::fmt!(
                ConfigError,
                "Invalid character {c:?} at position {p}"
            ));
        }
    }
    Ok(value)
}

#[cfg(feature = "_sync-sender")]
fn parse_conf_value<T>(param_name: &str, str_value: &str) -> Result<T>
where
    T: FromStr,
    T::Err: std::fmt::Debug,
{
    str_value.parse().map_err(|e| {
        fmt!(
            ConfigError,
            "Could not parse {param_name:?} to number: {e:?}"
        )
    })
}

/// `true` when the ingress parser recognizes `str_value` as an
/// `initial_connect_retry` mode that blocks or fails fast at startup.
/// The pool's `lazy_connect` conflict check derives from the parser so the
/// two cannot drift when a mode is added.
#[cfg(feature = "_sender-qwp-ws")]
pub(crate) fn initial_connect_retry_value_is_blocking(str_value: &str) -> bool {
    matches!(
        parse_initial_connect_retry_value(str_value),
        Ok(mode) if mode != conf::QwpWsInitialConnectMode::Async
    )
}

#[cfg(feature = "_sender-qwp-ws")]
fn parse_initial_connect_retry_value(str_value: &str) -> Result<conf::QwpWsInitialConnectMode> {
    if str_value.eq_ignore_ascii_case("on") || str_value.eq_ignore_ascii_case("true") {
        return Ok(conf::QwpWsInitialConnectMode::Sync);
    }
    if str_value.eq_ignore_ascii_case("sync") {
        return Ok(conf::QwpWsInitialConnectMode::Sync);
    }
    if str_value.eq_ignore_ascii_case("off") || str_value.eq_ignore_ascii_case("false") {
        return Ok(conf::QwpWsInitialConnectMode::Off);
    }
    if str_value.eq_ignore_ascii_case("async") {
        return Ok(conf::QwpWsInitialConnectMode::Async);
    }
    Err(error::fmt!(
        ConfigError,
        "invalid initial_connect_retry [value={str_value}, allowed-values=[on, off, true, false, sync, async]]"
    ))
}

#[cfg(feature = "_sender-qwp-ws")]
fn parse_size_conf_value(param_name: &str, str_value: &str) -> Result<u64> {
    let mut end = str_value.len();
    if end == 0 {
        return Err(error::fmt!(
            ConfigError,
            "invalid {param_name} [value={str_value}]"
        ));
    }

    let bytes = str_value.as_bytes();
    if matches!(bytes[end - 1], b'b' | b'B') {
        end -= 1;
    }

    let multiplier = if end > 0 {
        match bytes[end - 1] {
            b'k' | b'K' => {
                end -= 1;
                1024
            }
            b'm' | b'M' => {
                end -= 1;
                1024 * 1024
            }
            b'g' | b'G' => {
                end -= 1;
                1024 * 1024 * 1024
            }
            b't' | b'T' => {
                end -= 1;
                1024_u64 * 1024 * 1024 * 1024
            }
            _ => 1,
        }
    } else {
        1
    };

    if end == 0 {
        return Err(error::fmt!(
            ConfigError,
            "invalid {param_name} [value={str_value}]"
        ));
    }

    let digits = &str_value[..end];
    let value = digits
        .parse::<u64>()
        .map_err(|_| error::fmt!(ConfigError, "invalid {param_name} [value={str_value}]"))?;
    value.checked_mul(multiplier).ok_or_else(|| {
        error::fmt!(
            ConfigError,
            "{param_name} overflows u64 [value={str_value}]"
        )
    })
}

#[cfg(feature = "_sender-qwp-ws")]
fn parse_sf_durability_value(str_value: &str) -> Result<conf::SfDurability> {
    if str_value.eq_ignore_ascii_case("memory") {
        return Ok(conf::SfDurability::Memory);
    }
    if str_value.eq_ignore_ascii_case("periodic") {
        return Ok(conf::SfDurability::Periodic);
    }
    if str_value.eq_ignore_ascii_case("flush") {
        return Ok(conf::SfDurability::Flush);
    }
    if str_value.eq_ignore_ascii_case("append") {
        return Ok(conf::SfDurability::Append);
    }
    Err(error::fmt!(
        ConfigError,
        "invalid sf_durability [value={str_value}, allowed-values=[memory, periodic, flush, append]]"
    ))
}

#[cfg(feature = "_sender-qwp-ws")]
fn parse_qwp_ws_progress_value(str_value: &str) -> Result<QwpWsProgress> {
    if str_value.eq_ignore_ascii_case("background") {
        return Ok(QwpWsProgress::Background);
    }
    if str_value.eq_ignore_ascii_case("manual") {
        return Ok(QwpWsProgress::Manual);
    }
    Err(error::fmt!(
        ConfigError,
        "invalid qwp_ws_progress [value={str_value}, allowed-values=[background, manual]]"
    ))
}

#[cfg(feature = "_sender-qwp-ws")]
fn reject_unsupported_qwp_ws_sf_config(qwp_ws: &conf::QwpWsConfig) -> Result<()> {
    if matches!(
        *qwp_ws.sf_durability,
        conf::SfDurability::Flush | conf::SfDurability::Append
    ) {
        let durability = qwp_ws.sf_durability.as_conf_value();
        return Err(error::fmt!(
            ConfigError,
            "sf_durability={durability} is not yet supported (use sf_durability=memory or periodic)"
        ));
    }
    if *qwp_ws.sf_durability == conf::SfDurability::Periodic && qwp_ws.sf_dir.is_none() {
        return Err(error::fmt!(
            ConfigError,
            "sf_durability=periodic requires sf_dir"
        ));
    }
    if qwp_ws.sf_sync_interval.is_specified()
        && *qwp_ws.sf_durability != conf::SfDurability::Periodic
    {
        return Err(error::fmt!(
            ConfigError,
            "sf_sync_interval_millis requires sf_durability=periodic"
        ));
    }

    Ok(())
}

#[cfg(feature = "sync-sender-qwp-ws")]
fn qwp_ws_auth_header(auth: &Option<conf::AuthParams>) -> Result<Option<String>> {
    match auth {
        Some(conf::AuthParams::Basic(b)) => Ok(Some(b.to_header_string())),
        Some(conf::AuthParams::Token(t)) => Ok(Some(t.to_header_string()?)),
        #[cfg(feature = "_sender-tcp")]
        Some(conf::AuthParams::Ecdsa(_)) => Err(error::fmt!(
            AuthError,
            "ECDSA authentication is not supported for QWP/WebSocket. \
             Use basic or token authentication instead."
        )),
        None => Ok(None),
    }
}

#[cfg(feature = "_sender-tcp")]
fn b64_decode(descr: &'static str, buf: &str) -> Result<Vec<u8>> {
    use base64ct::{Base64UrlUnpadded, Encoding};
    Base64UrlUnpadded::decode_vec(buf).map_err(|b64_err| {
        fmt!(
            AuthError,
            "Misconfigured ILP authentication keys. Could not decode {}: {}. \
            Hint: Check the keys for a possible typo.",
            descr,
            b64_err
        )
    })
}

#[cfg(feature = "_sender-tcp")]
fn parse_public_key(pub_key_x: &str, pub_key_y: &str) -> Result<Vec<u8>> {
    let mut pub_key_x = b64_decode("public key x", pub_key_x)?;
    let mut pub_key_y = b64_decode("public key y", pub_key_y)?;

    // SEC 1 Uncompressed Octet-String-to-Elliptic-Curve-Point Encoding
    let mut encoded = Vec::new();
    encoded.push(4u8); // 0x04 magic byte that identifies this as uncompressed.
    let pub_key_x_ken = pub_key_x.len();
    if pub_key_x_ken > 32 {
        return Err(fmt!(
            AuthError,
            "Misconfigured ILP authentication keys. Public key x is too long. \
            Hint: Check the keys for a possible typo."
        ));
    }
    let pub_key_y_len = pub_key_y.len();
    if pub_key_y_len > 32 {
        return Err(fmt!(
            AuthError,
            "Misconfigured ILP authentication keys. Public key y is too long. \
            Hint: Check the keys for a possible typo."
        ));
    }
    encoded.resize((32 - pub_key_x_ken) + 1, 0u8);
    encoded.append(&mut pub_key_x);
    encoded.resize((32 - pub_key_y_len) + 1 + 32, 0u8);
    encoded.append(&mut pub_key_y);
    Ok(encoded)
}

#[cfg(feature = "_sender-tcp")]
fn parse_key_pair(auth: &conf::EcdsaAuthParams) -> Result<EcdsaKeyPair> {
    let private_key = b64_decode("private authentication key", auth.priv_key.as_str())?;
    let public_key = parse_public_key(auth.pub_key_x.as_str(), auth.pub_key_y.as_str())?;

    #[cfg(feature = "aws-lc-crypto")]
    let res = EcdsaKeyPair::from_private_key_and_public_key(
        &ECDSA_P256_SHA256_FIXED_SIGNING,
        &private_key[..],
        &public_key[..],
    );

    #[cfg(feature = "ring-crypto")]
    let res = {
        let system_random = SystemRandom::new();
        EcdsaKeyPair::from_private_key_and_public_key(
            &ECDSA_P256_SHA256_FIXED_SIGNING,
            &private_key[..],
            &public_key[..],
            &system_random,
        )
    };

    res.map_err(|key_rejected| {
        fmt!(
            AuthError,
            "Misconfigured ILP authentication keys: {}. Hint: Check the keys for a possible typo.",
            key_rejected
        )
    })
}

struct DebugBytes<'a>(pub &'a [u8]);

impl Debug for DebugBytes<'_> {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        write!(f, "b\"")?;

        for &byte in self.0 {
            match byte {
                // Printable ASCII characters (except backslash and quote)
                0x20..=0x21 | 0x23..=0x5B | 0x5D..=0x7E => {
                    write!(f, "{}", byte as char)?;
                }
                // Common escape sequences
                b'\n' => write!(f, "\\n")?,
                b'\r' => write!(f, "\\r")?,
                b'\t' => write!(f, "\\t")?,
                b'\\' => write!(f, "\\\\")?,
                b'"' => write!(f, "\\\"")?,
                b'\0' => write!(f, "\\0")?,
                // Non-printable bytes as hex escapes
                _ => write!(f, "\\x{byte:02x}")?,
            }
        }

        write!(f, "\"")
    }
}

#[cfg(all(test, feature = "_sync-sender"))]
mod tests;