bezant_server/events/mod.rs
1//! Event capture + REST surface for streaming CPAPI topics.
2//!
3//! `bezant-server` runs an internal [`bezant::WsClient`] that subscribes
4//! to order, PnL, and (lazily) market-data feeds. Decoded frames go into
5//! per-topic ring buffers behind cursor-based REST endpoints
6//! (`/events/{topic}?since=<cursor>`), so polling consumers can read every
7//! event the socket has seen since their last visit — no events lost
8//! between client polls, no need to keep a long-lived WS open from the
9//! consumer side.
10//!
11//! See the architecture sketch:
12//!
13//! ```text
14//! CPGateway WS ─► bezant_core::WsClient ─► Connector (tokio task)
15//! │
16//! ▼
17//! per-topic TopicRing
18//! │
19//! ▼
20//! GET /events/{topic}?since=<cursor>
21//! ```
22
23pub mod connector;
24pub mod persistence;
25pub mod ring;
26
27use std::collections::BTreeMap;
28
29use serde::Serialize;
30
31pub use connector::{spawn_connector, ConnectorCfg, EventsHandle, TestSink};
32pub use persistence::{EventLog, RetentionPolicy};
33pub use ring::{ReadResult, TopicRing};
34
35/// One captured event. Wire-shape returned by `/events/{topic}` endpoints.
36#[derive(Clone, Debug, Serialize)]
37pub struct ObservedEvent {
38 /// Server-assigned cursor, strictly increasing per topic for the life
39 /// of the process and seeded from the boot time so a restart does not
40 /// rewind it. Always below 2^53. Use it as the `since=` parameter on the
41 /// next poll.
42 pub cursor: u64,
43 /// Topic name — `"orders"`, `"pnl"`, `"marketdata:265598"`, `"gap"`.
44 pub topic: String,
45 /// RFC 3339 timestamp at which the connector pushed this into the ring.
46 pub received_at: String,
47 /// The epoch this event was captured under. Seeded from the boot time
48 /// (Unix ms) and incremented on every reconnect, so it changes on both a
49 /// reconnect and a restart. Clients use it to detect a gap.
50 pub reset_epoch: u64,
51 /// The decoded JSON frame. Shape depends on the topic — see the
52 /// generated `bezant-client` TS types for the typed views.
53 pub payload: serde_json::Value,
54}
55
56/// Snapshot of the connector's state, returned by `GET /events/_status`.
57#[derive(Clone, Debug, Default, Serialize)]
58pub struct EventsStatus {
59 /// `true` if the underlying WS is currently connected.
60 pub connected: bool,
61 /// RFC 3339 timestamp of the last frame the connector received, if any.
62 pub last_message_at: Option<String>,
63 /// How many times the connector has reconnected since the process
64 /// started. Bumps on every successful reconnect, not on each retry.
65 pub reconnect_count: u64,
66 /// Wall-clock seconds since the connector task spawned.
67 pub uptime_seconds: u64,
68 /// Current `reset_epoch`. Seeded from the boot time (Unix ms) and
69 /// bumped once per successful reconnect — never on a failed attempt.
70 pub reset_epoch: u64,
71 /// Topics currently subscribed at the upstream WS. Always includes
72 /// `"orders"` and `"pnl"`; market data topics appear when a client
73 /// has polled `/events/marketdata?conid=…` recently.
74 pub topics_subscribed: Vec<String>,
75 /// Per-topic ring buffer occupancy. Useful for "are we close to
76 /// wraparound?" capacity planning.
77 pub buffer_sizes: BTreeMap<String, usize>,
78 /// Whether CPAPI has actually honoured each standing subscription
79 /// (`orders`, `pnl`). `topics_subscribed` above records what we ASKED
80 /// for; this records what we GOT. The difference is the whole story of
81 /// a fill that never arrived: CPAPI answers `sor+{}` with
82 /// `{"error":"unable to subscribe"}` more often than not, and a socket
83 /// that outlives a Gateway re-login keeps heartbeating with every
84 /// subscription dead. A consumer confirming fills should treat anything
85 /// but `subscribed` on `orders` as "the stream will not tell you".
86 pub subscriptions: BTreeMap<String, SubscriptionState>,
87 /// How many subscribe refusals CPAPI has sent since the process started.
88 pub subscribe_refusals: u64,
89 /// How many times the connector tore its socket down because the
90 /// Gateway's session id changed underneath it (a re-login).
91 pub session_rollovers: u64,
92 /// How many times the connector task panicked and was restarted by its
93 /// supervisor since the process started.
94 pub connector_restarts: u64,
95 /// sqlite history appends that failed since the process started.
96 pub persist_failures: u64,
97}
98
99/// Where a standing subscription (`orders`, `pnl`) stands at the upstream WS.
100#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
101#[serde(rename_all = "snake_case")]
102pub enum SubscriptionState {
103 /// Subscribe sent; CPAPI has neither honoured nor refused it yet.
104 Pending,
105 /// CPAPI has sent at least one real frame on the topic this connection.
106 Subscribed,
107 /// CPAPI answered the subscribe with an error; the connector is retrying.
108 Refused,
109 /// Asked repeatedly (`BEZANT_EVENTS_QUIET_AFTER_ROUNDS`) with neither a
110 /// frame nor a refusal. CPAPI honours `sor+{}` in silence when there are
111 /// no live orders, so this is subscribed-as-far-as-anyone-can-tell: the
112 /// connector stops re-asking. A refusal puts it back to `refused`; the
113 /// first real frame makes it `subscribed`.
114 Quiet,
115}
116
117/// Reason a synthetic [`ObservedEvent`] of topic `"gap"` was injected.
118/// Surfaced to the client so they know they missed something.
119#[derive(Clone, Copy, Debug, Serialize)]
120#[serde(rename_all = "snake_case")]
121pub enum GapReason {
122 /// The connector reconnected — anything that happened while
123 /// disconnected is permanently gone from the upstream feed.
124 ReconnectedAfterDisconnect,
125 /// Process restarted — ring buffers were freshly initialised.
126 ProcessRestart,
127}