Realtime uses the binary Centrifugo Protobuf WebSocket protocol (always included; the Cargo realtime feature flag is a no-op stub). Helpers return TypedSubscription<T>. Prefer recv_result().await so a failed feed cannot look like a clean end of stream. Account orders are
the exception: orders.subscribe uses the private WebSocket and yields private-stream events.
The SDK negotiates the centrifuge-protobuf WebSocket subprotocol. Both Centrifugo control frames
and :proto channel publications are binary Protobuf. ConnectRPC's optional JSON wire mode is
unrelated and never changes realtime framing.
Managed order-book / market-overview subscriptions expose updates().recv().await plus set_on_error(...) and err().
Contract
use polyester::Error;
let mut sub = client
.balances
.subscribe(client.default_account_id.as_deref())
.await?; // handshake done
sub.set_on_error(|error| eprintln!("realtime interruption: {error}"));
while let Some(balance) = sub.recv_result().await? {
println!("{balance:?}");
}- Private channels need API key + Account ID.
- Public trades, user fills, ledger transfers, and trigger events arrive as batches. Their
subscriptions yield every event in order, one at a time; an event that fails to decode is reported
through
set_on_error, and the rest of its batch is still delivered. - Overflow is fail-closed (
Error::QueueOverflow). - Reconnect uses capped exponential backoff with per-subscription jitter.
- Hyphenated channel segments (e.g.
api-keys) must be signed with RFC 3986-preserving query encoding (built into the SDK).
What you can stream
Consume every stream below with recv_result(), except the managed order book and market overview,
which use updates().
- Public trades (public):
market_data.subscribe_trades - Candles (public):
market_data.subscribe_candles - Order book (managed) (public):
orderbook.create_subscription - Market overview (managed) (public):
market_overview.create_subscription - Heatmap (public):
heatmap.subscribe_live - Open lifecycle flows (either):
lifecycle.subscribe_open_flows - One lifecycle flow (public):
lifecycle.subscribe_flow_detail - Zipped-asset supply (public):
zipper.subscribe_zipped_asset_supply - Profile identity (public):
auth.profile.subscribe_identity - Orders (private):
orders.subscribe - Triggers, balances, trades (private): matching
subscribehelpers - Transfers (private):
transfers.subscribe - API keys (private):
api_keys.subscribe - API/subaccount policies (private):
policies.subscribe_*_policies - Subaccounts, address book (private): matching subscribe helpers
Error::PermissionDenied is non-transient; inspect code, context, and endpoint, then update the key policy before reconnecting.Public trades
client.wait_for_catalogs().await?;
let mut sub = client.market_data.subscribe_trades("BTC-USDT").await?;
while let Some(trade) = sub.recv_result().await? {
println!("{trade:?}");
break;
}Order stream
orders.subscribe streams the account's live orders over the private WebSocket
(GET /v1/ws/private). It yields PrivateOrdersEvent values, not bare orders. The call returns at
once, and the handshake result arrives as Connected or Error.
use polyester::models::{OrderUnitEvent, PrivateOrdersEvent};
let mut orders = client
.orders
.subscribe(client.default_account_id.as_deref())
.await?;
while let Some(event) = orders.recv_result().await? {
match event {
PrivateOrdersEvent::Snapshot(live) => println!("{} live orders", live.len()),
PrivateOrdersEvent::Ready => println!("order stream is current"),
PrivateOrdersEvent::Unit(unit) => {
for order in unit.orders {
println!("{} {} v{}", order.order_id, order.status, order.version);
}
for event in unit.events {
if let OrderUnitEvent::Order(event) = event {
println!("{} {:?} {:?}", event.kind, event.reject_reason, event.cancel_reason);
}
}
}
PrivateOrdersEvent::Error(error) => eprintln!("order stream: {error}"),
_ => {}
}
}Connected: the server confirmed the subscription; sent again after each reconnect.Snapshot(orders): the account's full live view. Replace local state with it.Unit(unit): one change, with changed orders (complete), removed order IDs, and events.Ready: the stream reached the account's read boundary. Wait for it before relying on the stream.Stale(stale): the view, or fororder_unavailableonly the orders inorder_ids, is not current.Disconnected: the stream dropped. It resumes by itself unless an error ended it.Error(error): a stream error. A terminal one is followed byDisconnectedand the end of the stream.
Keep the order copy with the higher version; equal versions are identical. To combine the stream
with an order read, read first, then subscribe, wait for Ready, and merge both by version. The
stream reconnects with backoff, resumes from its last applied position, and takes a fresh snapshot
when it cannot resume. A consumer that falls behind ends the stream with Error::QueueOverflow.
Snapshot-then-stream
Managed order-book subscriptions snapshot via REST, then stream deltas with gap recovery. Prefer these helpers over wiring raw channels yourself. The create call returns only after both the WebSocket handshake and initial snapshot succeed.
Market overview uses the same managed pattern. For entity streams (balances, trades, transfers, triggers), events are not a complete starting snapshot: subscribe first, fetch the corresponding list/snapshot, buffer events during the read, then reconcile by entity ID plus version/timestamp. After reconnect, re-read authoritative state before declaring the bot healthy.
Typed subscriptions do not recover missed publications across reconnect (no Centrifugo resume
cursor). Poll sub.take_resubscribed() (or watch sub.resubscribes()) after each receive and
treat a latch as a gap: rebuild from REST before trusting the stream again. A WebSocket read
timeout is treated as connection death so half-open sockets reconnect instead of freezing.
Never quote from raw order-book deltas. Use orderbook.create_subscription, which detects gaps
and refreshes its snapshot.