Backpressure

Backpressure semantics differ by mode. A consumer registered with .on_event sees the two cases differently.

Live: drop-oldest, then the subscription ends

The live feed cannot be slowed without losing the connection's reconnect-replay window. If a callback consumer falls behind and the engine's per-subscription outbound buffer overflows, the engine drops the oldest staged batch to keep the connection alive. If the consumer keeps lagging and the buffer stays full, delivery stops: the subscription ends and its handle goes inactive. There is no per-poll drop signal to match — the documented path is the callback, so the signal you observe is sub.is_active() returning false:

use kairos::{Client, GreeksRow};

let client = Client::connect(("me@example.com", "secret"))?;

let sub = client
    .live()
    .greeks(["QQQ"])
    .on_event(|row: &GreeksRow| {
        // keep this fast; a slow callback is what makes the consumer lag
        tracing::trace!(delta = row.delta, gamma = row.gamma, "greeks");
    })?;

// ... later, on your own thread ...
if !sub.is_active() {
    tracing::warn!("subscription ended — consumer lagged past the buffer");
}

The callback runs on a shared delivery worker, so the single rule is: do not block inside it. Hand work off to your own channel or thread pool if a row needs more than a cheap update.

Historical / replay: producer blocks

The historical and replay paths block the producer when the consumer falls behind — the engine suspends the upstream pull, the query takes longer, but no rows are dropped. The callback fires once per row over the bounded set and the drop-oldest behavior above never applies.

Delivery model

One shared worker, not one thread per subscription

Every .on_event callback across every subscription is driven by a single shared delivery worker; there is no OS thread per stream to manage. Registering more subscriptions does not spawn more threads. Cloning the Client across your own threads is still cheap (an Arc bump of the engine link) when you want independent producers:

use kairos::{Client, GreeksRow};

let client = Client::connect(("me@example.com", "secret"))?;

let subs: Vec<_> = ["QQQ", "SPY"]
    .into_iter()
    .map(|sym| {
        client
            .live()
            .greeks([sym])
            .on_event(move |row: &GreeksRow| {
                tracing::trace!(sym, iv = row.implied_vol, "greeks");
            })
    })
    .collect::<Result<_, _>>()?;

// hold the handles; dropping them does not stop delivery, but
// `unsubscribe()` does.

Fan-out

The callback already delivers to your code; to fan the same rows out to multiple consumers, push from inside the callback onto a broadcast or in-process channel. There is no fan-out inside the engine; subscribe once per consumer that needs an independent view, or forward decoded rows over a channel:

use kairos::{Client, GreeksRow};
use std::sync::mpsc;

let (tx, _rx) = mpsc::sync_channel::<(f64, f64)>(2048);

let sub = client
    .live()
    .greeks(["QQQ"])
    .on_event(move |row: &GreeksRow| {
        // non-blocking send keeps the delivery worker free
        let _ = tx.try_send((row.delta, row.gamma));
    })?;