WebSocket

WsConn is a TLS 1.3 WebSocket client that a Driver owns.

• Frames are parsed in place: WsHandler::on_message gets a slice of the buffer the kernel wrote into (one copy per byte).
• RxMeta carries the read’s kernel receive timestamp and, where the NIC has one, its hardware timestamp.
• Timers (Pings, timeouts, the close wait) run on the driver’s deadline check (every 250 µs), never on the read or send path.

A complete client

Connect, subscribe, print each message, reconnect and subscribe again after a drop, then close cleanly:

use ergon_io::reconnect::{Closed, LinkId, LinkState, Retry};
use ergon_io::{ConnId, Error, Handler, Reconnect, RxMeta, WsDial, WsHandler};

/// Sent on every connect and reconnect.
pub const SUBSCRIBE: &str = r#"{"method":"SUBSCRIBE","params":["btcusdt@bookTicker"],"id":1}"#;

/// The application. `Driver::poll` calls it.
pub struct App {
    /// Keeps the connection open: reopens it with backoff after a drop.
    pub links: Reconnect<WsDial>,
    /// The one link (a connection that `links` keeps open).
    pub link: LinkId,
    /// Connections opened during the last poll: subscribe on them.
    opened: Vec<ConnId>,
    /// Close after this many messages.
    pub limit: u64,
    /// Messages received.
    pub messages: u64,
    /// Connections opened; more than 1 means it reconnected.
    pub opens: u32,
}

impl App {
    /// `dial` builds a fresh connection set-up for each attempt.
    ///
    /// # Errors
    /// `retry` is out of range.
    pub fn new(
        retry: Retry,
        dial: impl FnMut() -> ergon_io::Result<WsDial> + 'static,
        limit: u64,
    ) -> ergon_io::Result<Self> {
        let mut links = Reconnect::new(retry)?;
        let link = links.add(dial);
        Ok(Self {
            links,
            link,
            opened: Vec::new(),
            limit,
            messages: 0,
            opens: 0,
        })
    }

    /// Closed cleanly, or the link gave up.
    #[must_use]
    pub fn finished(&self) -> bool {
        matches!(
            self.links.state(self.link),
            Some(LinkState::Stopped | LinkState::GaveUp)
        )
    }
}

impl Handler for App {
    fn on_open(&mut self, id: ConnId) {
        if let Some(opened) = self.links.on_open(id) {
            self.opens += 1;
            if opened.reconnect {
                println!("reconnected: subscribing again");
            }
            // Sends wait until `poll` returns: remember the connection.
            self.opened.push(id);
        }
    }

    fn on_closed(&mut self, id: ConnId, error: Error) {
        println!("closed: {error}");
        if let Some(Closed::Retry { delay_ms, .. }) = self.links.on_closed(id) {
            println!("reconnecting in {delay_ms} ms");
        }
    }
}

impl WsHandler for App {
    fn on_message(&mut self, _: ConnId, _: u8, payload: &[u8], _: &RxMeta, _: u16) {
        // `payload` points into the receive buffer: use it during this call only.
        println!("{}", std::str::from_utf8(payload).unwrap_or("<binary>"));
        self.messages += 1;
    }
}

/// The loop: runs until the clean close, or until the link gives up.
///
/// # Errors
/// The backend failed. A connection that fails goes to `on_closed` instead.
#[cfg(target_os = "linux")]
pub fn run(config: ergon_io::Config, app: &mut App) -> ergon_io::Result<()> {
    use ergon_io::{Driver, WsConn};

    let mut driver: Driver<WsConn> = Driver::new(config)?;
    let mut closing = false;
    while !app.finished() {
        driver.poll(app)?; // receive, decrypt, deliver, send
        app.links.poll(&mut driver); // start a reconnect when one is due
        for id in app.opened.drain(..) {
            // A connection the server is already closing takes no sends.
            if let Some(mut conn) = driver.get_mut(id)
                && !conn.is_closing()
            {
                conn.send_text(SUBSCRIBE)?;
            }
        }
        if app.messages >= app.limit && !closing {
            closing = true;
            app.links.stop(app.link); // no reconnect after this close
            if let Some(LinkState::Open(id)) = app.links.state(app.link)
                && let Some(mut conn) = driver.get_mut(id)
                && !conn.is_closing()
            {
                // `on_closed` follows once the server answers.
                conn.close(1000, "done")?;
            }
        }
    }
    Ok(())
}

#[cfg(all(target_os = "linux", feature = "webpki-roots"))]
fn main() -> Result<(), Box<dyn std::error::Error>> {
    use ergon_io::{Config, Target, WsOptions, tls};

    let mut args = std::env::args().skip(1);
    let limit: u64 = args.next().map_or(Ok(20), |a| a.parse())?;
    let config = Config::parse(&args.next().unwrap_or_else(|| "recv-busy".into()))?;
    // In production, pin this thread to an isolated core first.
    let tls = tls::client_config(tls::webpki_roots())?;
    let target = Target::new("stream.binance.com", 9443, "/ws");
    // Limits, pings and timeouts: WsOptions.
    let dial = move || WsDial::with_options(target.clone(), tls.clone(), WsOptions::default());
    let mut app = App::new(Retry::default(), dial, limit)?;
    run(config, &mut app)?;
    println!("{} messages, {} connection(s)", app.messages, app.opens);
    Ok(())
}

(From io/examples/websocket.rs. cargo test -p ergon-io builds it, and io/tests/example_websocket.rs runs it against a local server on every backend: a server close, the reconnect, the second subscribe and the clean close.)

Run it (Linux):

cargo run -p ergon-io --features webpki-roots --example websocket -- 20 recv-busy

• Arguments: messages to print before closing, then the backend configuration.
• The feed is Binance’s public spot stream; any public feed will do.

Options

WsOptions, on WsDial::with_options or WsConn::connect_with:

• Any byte received after a Ping counts as the answer, not only a Pong.
• Server Pings are answered at once.

Close

• Data that arrives after a server Close is dropped.
• A protocol error drops the connection (RFC 6455 allows it): no Close 1002.

Reconnect

Reconnect is optional. You own it (in your handler, as in the example) and feed it every open and close.

• on_open(id) returns Opened { link, reconnect }: subscribe either way; with reconnect, resync any state kept from the old connection.
• on_closed(id) returns the next step: Retry, GaveUp or Stopped.
• stop(link) vetoes reconnects; resume(link) starts again.
• Driver::close reports nothing: call on_closed(id) yourself after it.
• poll starts due attempts. It costs one branch while nothing waits.

Connect limits and the server’s wait

Venues limit connect attempts per IP (Binance: 300 per 5 minutes; OKX: 3 per second). One Budget bucket is the gate of every link on that IP:

    // Connect attempts per IP: 3 per second (OKX). One bucket, the gate of
    // every WebSocket link on that IP.
    let mut budget = Budget::new(vec![BucketSpec::rolling("ws-connects", 3, 1_000, 10)])?;
    let connects = budget.id("ws-connects").ok_or("no bucket")?;
    let mut links = Reconnect::new(Retry::default())?;

        // Once per loop pass; `now` is the budget's clock, read once.
        let started = links.poll_at_gated(&mut f, now, || {
            budget
                .try_charge(&[Charge::new(connects, 1)], Class::Normal, now)
                .map(drop) // passed: connect now
                .map_err(|l| l.wait_ns(now)) // refused: ask again after this
        });

(From io/tests/reconnect.rs, which cargo test -p ergon-io runs on every platform, with a stand-in connector and a fake clock.)

• poll_gated(driver, gate) starts an attempt only when the gate passes it, the first attempt after add included: 50 links added at once start at the gate’s pace, not in one burst.
• A refusal is not a failed attempt: the backoff does not grow. The link asks again after the gate’s wait, at most max_ms later.
• Each pass starts after the last link the gate passed, so links that keep failing cannot starve the others.
• defer(link, wait_ns): no attempt before then. It stays with the link across later closes and resume.

A refused handshake, in on_closed:

    // Binance's documented minimum for a 418: 2 minutes.
    let ban_min = NonZeroU64::new(120_000).ok_or("zero")?;
    // In `Handler::on_closed(id, error)`: a refused handshake holds the
    // connect bucket (429) or halts it (418), and the link waits it out.
    let on_closed = |rc: &mut Reconnect<u32>, budget: &mut Budget, id, error: &Error, now| -> R {
        let link = rc.link_of(id).ok_or("not a link")?;
        if let Some(Verdict::Throttled { until_ns, .. } | Verdict::Banned { until_ns, .. }) =
            outcome::refused_upgrade(
                error,
                connects,
                budget,
                ban_min,
                Now {
                    ns: now,
                    real_ns: now,
                },
            )?
        {
            rc.defer_at(link, u64::try_from(until_ns - now)?, now);
        }
        rc.on_closed_at(id, now);
        Ok(())
    };

• 429: the bucket is held for Retry-After (or its backoff without one).
• 418: the bucket is halted. No link on that IP connects again until a person calls Budget::clear_halt, even after the server’s wait.
• Give the HTTP pool a bucket of its own (HTTP client): venues count WebSocket connects.

Sending

• send_text, send_binary, ping, or send_frame(opcode, payload).
• A message larger than one TLS record goes out over several.
• On a driver-owned connection: driver.get_mut(id), then the same methods.
• Sends go out on the next poll, or at once with Driver::flush.
• Handlers cannot send (they have no driver): remember the id, send after poll returns (as the example does).
• stats().auto_frames counts the frames the connection sends by itself: Pongs, Pings on a quiet connection, the echo of a server Close. A venue that limits messages per connection may count them (Binance spot does): charge the change each pass with Budget::force.

Handshake

• Checks status 101, Upgrade, Connection and Sec-WebSocket-Accept.
• Any other status is Error::Upgrade { status, retry_after_ms, .. }: a 429 or 418 can be told apart, and retry_after_ms is the server’s Retry-After (seconds or a date). Wait at least that long before connecting again: outcome::refused_upgrade and Reconnect::defer (see Reconnect).
• Refuses a subprotocol it did not offer.
• Offers no extension and refuses a server that picks one.
• Extra headers (auth, say) go on the Target: Target::new(host, port, path).with_header(name, value).

Not supported

• Compression (permessage-deflate) and other extensions: decompressing breaks in-place parsing and costs CPU per message.
• Proxies.
• Sending fragmented messages (receiving them works).

Measured results

Order round trip: a 64-byte echo, one message in flight. Every figure, with its max:

• Fair run, Graviton: all six backends with 1, 4, 16 and 64 sockets on the loop.
• Round trips on x86: 1 connection.

Pick the backend by the total socket count on the loop, feeds included: choosing a backend.