# Streaming

Subscribe with observable errors, handshake-before-return, fail-closed overflow, and private Account ID requirements.

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.

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

```rust
use polyester::Error;

let mut sub = client
    .orders
    .subscribe(client.default_account_id.as_deref())
    .await?; // handshake done
sub.set_on_error(|error| eprintln!("realtime interruption: {error}"));
while let Some(order) = sub.recv_result().await? {
    println!("{} {}", order.status, order.order_id);
}
```

- Private channels need API key + Account ID.
- 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

| Stream                                | Method                                                     | Auth              | Consume         |
| ------------------------------------- | ---------------------------------------------------------- | ----------------- | --------------- |
| Public trades                         | `market_data.subscribe_trades`                             | public            | `recv_result()` |
| Candles                               | `market_data.subscribe_candles`                            | public            | `recv_result()` |
| Order book (managed)                  | `orderbook.create_subscription`                            | public            | `updates()`     |
| Market overview (managed)             | `market_overview.create_subscription`                      | public            | `updates()`     |
| Heatmap                               | `heatmap.subscribe_live`                                   | public            | `recv_result()` |
| Lifecycle                             | `lifecycle.subscribe_open_flows` / `subscribe_flow_detail` | public or private | `recv_result()` |
| Zipped-asset supply                   | `zipper.subscribe_zipped_asset_supply`                     | public            | `recv_result()` |
| Profile identity                      | `auth.profile.subscribe_identity`                          | public            | `recv_result()` |
| Orders / triggers / balances / trades | matching `subscribe` helpers                               | private           | `recv_result()` |
| Transfers                             | `transfers.subscribe`                                      | private           | `recv_result()` |
| API keys                              | `api_keys.subscribe`                                       | private           | `recv_result()` |
| API/subaccount policies               | `policies.subscribe_*_policies`                            | private           | `recv_result()` |
| Subaccounts / address book            | matching subscribe helpers                                 | private           | `recv_result()` |

> **Account-administration streams**
>
> Account-administration streams require the corresponding API-key read permission and an Account ID. An HTTP 403 `Error::PermissionDenied` is non-transient; inspect `code`, `context`, and `endpoint`, then update the key policy before reconnecting.

## Public trades

```rust
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;
}
```

## 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 (orders, 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.

## Overflow

> **Fail-closed overflow**
>
> Updates are never silently dropped when the queue fills. Resubscribe and rebuild authoritative state, a new stream alone does not recover missed entity events.

## Related

- [Realtime](https://testnet.polyester.com/docs/sdk/rust/reference/realtime)
- [WebSocket session model](https://testnet.polyester.com/docs/developer-docs/shared-concepts/websocket-session-model)