brokapi_runtime/emitter.rs
1//! Egress: per-subscription buffers in front of the shared `Sender<Envelope>`
2//! Shared by the connectors that read their streams in a task of their own (T-Bank, BCS).
3//!
4//! The reader task never blocks: events go into the subscription's buffer (`BackpressureManager`),
5//! then as many as the channel accepts are moved out with `try_send`. A full channel leaves events in
6//! the buffer, so a slow consumer fills it and the policy applies: `Conflate` keeps the latest,
7//! `DropNewest` counts drops, `Lossless` overflows: `Event::Overflow` is emitted and the connector is
8//! `Degraded` (commands are refused) until `reconcile()`.
9
10use std::collections::{HashMap, HashSet, VecDeque};
11use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
12use std::sync::{Arc, Mutex};
13
14use brokapi_core::{
15 Backpressure, Envelope, Event, InstrumentId, InstrumentKey, NativeId, SubscriptionId,
16 Timestamps, UnixNanos,
17};
18use brokapi_sansio::CoreEvent;
19use tokio::sync::mpsc::{self, error::TrySendError};
20use tracing::warn;
21
22use crate::backpressure::{BackpressureManager, PushResult};
23
24fn wall_nanos() -> UnixNanos {
25 UnixNanos::from_nanos(
26 std::time::SystemTime::now()
27 .duration_since(std::time::UNIX_EPOCH)
28 .map(|d| d.as_nanos() as i64)
29 .unwrap_or(0),
30 )
31}
32
33struct State {
34 bp: BackpressureManager,
35 subs: Vec<SubscriptionId>,
36 /// Popped from the buffer but refused by a full channel; sent first next time.
37 pending: HashMap<SubscriptionId, Envelope>,
38 overflowed: HashSet<SubscriptionId>,
39 conflating: HashSet<SubscriptionId>,
40 notices: VecDeque<Envelope>,
41}
42
43/// Moves events from a connector's reader task into the consumer's channel, without ever blocking.
44///
45/// This is the egress half of a connector. Each subscription has a buffer with its
46/// [`Backpressure`] policy. [`emit`](Self::emit) puts events into the buffers and moves as many as
47/// the channel accepts into it with `try_send`. When the consumer is slow the channel fills and the
48/// buffers take the strain:
49///
50/// | Policy | A slow consumer causes |
51/// |---|---|
52/// | [`Backpressure::Conflate`] | only the latest value is kept, older ones are replaced |
53/// | [`Backpressure::DropNewest`] | new events are thrown away once the buffer is full |
54/// | [`Backpressure::Lossless`] | the buffer overflows: an [`Event::Overflow`] notice is queued once, the event is lost, the emitter becomes degraded |
55///
56/// Degraded means events were lost that must not be lost. The connector copies the flag
57/// ([`is_degraded`](Self::is_degraded)) into its command gate, so trading commands fail with
58/// [`brokapi_core::Error::Degraded`] until a successful reconcile calls
59/// [`clear_degraded`](Self::clear_degraded). Why: after a hole in the order or fill stream the
60/// client's picture of its orders is wrong, and trading on it could be harmful.
61///
62/// Things to know:
63///
64/// - Events with no subscription (`sub` is `None`) skip the buffers. If the channel is full or
65/// closed they are dropped and counted in [`dropped_unrouted`](Self::dropped_unrouted).
66/// - Resync notices ([`Event::Resync`], made by the protocol cores after a sequence gap or a
67/// reconnect) travel like any other event of their subscription. The emitter does not create
68/// them and does not treat them specially.
69/// - Each envelope is stamped with the time it is built, which is when it leaves the buffer, not
70/// when the event was parsed. For a slow consumer the stamp is later than the arrival.
71/// - The [`Event::Overflow`] notice has its own queue, which is sent before the buffers. So it can
72/// reach the consumer before older events of the same subscription that are still waiting. Treat
73/// it as "events of this subscription are missing, re-read the state", not as an ordered marker.
74/// - Events for a subscription that was not [`register`](Self::register)ed are discarded.
75/// - All methods take `&self` and are safe to call from several tasks. Share the emitter as an
76/// [`Arc`].
77///
78/// # Examples
79///
80/// ```
81/// use brokapi_core::{Backpressure, Event, SubscriptionId};
82/// use brokapi_runtime::Emitter;
83/// use brokapi_sansio::CoreEvent;
84/// use tokio::sync::mpsc;
85///
86/// // A channel that holds one envelope, and a subscription whose buffer holds one event.
87/// let (tx, mut rx) = mpsc::channel(1);
88/// let emitter = Emitter::new("demo".into(), tx);
89/// let sub = SubscriptionId::new(1);
90/// emitter.register(sub, None, Backpressure::Lossless { cap: 1 });
91///
92/// let event = || CoreEvent { sub: Some(sub), payload: Event::Balances(vec![]) };
93/// // The consumer does not read. The first event fills the channel, the second waits in a
94/// // slot, the third fills the buffer, the fourth finds no room.
95/// for _ in 0..4 {
96/// emitter.emit(vec![event()]);
97/// }
98/// assert!(emitter.is_degraded());
99/// assert!(emitter.has_backlog());
100///
101/// // The consumer wakes up. Flushing moves what is left into the channel.
102/// let mut overflow_notices = 0;
103/// let mut events = 0;
104/// for _ in 0..10 {
105/// while let Ok(envelope) = rx.try_recv() {
106/// assert_eq!(&*envelope.source, "demo");
107/// assert_eq!(envelope.sub, Some(sub));
108/// match envelope.payload {
109/// Event::Overflow { .. } => overflow_notices += 1,
110/// _ => events += 1,
111/// }
112/// }
113/// emitter.flush();
114/// }
115/// assert_eq!(overflow_notices, 1);
116/// assert_eq!(events, 3); // the fourth event was lost
117///
118/// // A successful reconcile ends the degraded state.
119/// emitter.clear_degraded();
120/// assert!(!emitter.is_degraded());
121/// ```
122pub struct Emitter {
123 source: Arc<str>,
124 out: mpsc::Sender<Envelope>,
125 state: Mutex<State>,
126 degraded: AtomicBool,
127 backlog: AtomicBool,
128 /// Envelopes with no subscription that the channel refused.
129 dropped_unrouted: AtomicU64,
130}
131
132impl Emitter {
133 /// Creates an emitter that writes into `out`.
134 ///
135 /// `source` is the connector label from its config (for example `"bcs-ud"`). It is put into
136 /// every [`Envelope`] so the consumer can tell connectors apart.
137 pub fn new(source: Arc<str>, out: mpsc::Sender<Envelope>) -> Arc<Self> {
138 Arc::new(Self {
139 source,
140 out,
141 state: Mutex::new(State {
142 bp: BackpressureManager::new(),
143 subs: Vec::new(),
144 pending: HashMap::new(),
145 overflowed: HashSet::new(),
146 conflating: HashSet::new(),
147 notices: VecDeque::new(),
148 }),
149 degraded: AtomicBool::new(false),
150 backlog: AtomicBool::new(false),
151 dropped_unrouted: AtomicU64::new(0),
152 })
153 }
154
155 /// The `source` label given to [`new`](Self::new).
156 pub fn source(&self) -> &Arc<str> {
157 &self.source
158 }
159
160 /// Starts buffering a subscription with the given policy.
161 ///
162 /// Call it before the first event of `sub` can arrive. `instrument` is the subscription's
163 /// instrument, or `None` for account-level streams such as order state or portfolio.
164 pub fn register(
165 &self,
166 sub: SubscriptionId,
167 instrument: Option<InstrumentId>,
168 policy: Backpressure,
169 ) {
170 let instrument = instrument.unwrap_or_else(|| {
171 InstrumentId::new(
172 InstrumentKey::new("", ""),
173 NativeId::TickerClass {
174 ticker: String::new(),
175 class_code: String::new(),
176 },
177 )
178 });
179 let mut st = self.state.lock().expect("emitter poisoned");
180 st.bp.register(sub, instrument, policy);
181 st.subs.push(sub);
182 if policy == Backpressure::Conflate {
183 st.conflating.insert(sub);
184 }
185 }
186
187 /// Stops buffering a subscription. Events still waiting for it are discarded.
188 ///
189 /// An [`Event::Overflow`] notice already queued for the subscription is still sent.
190 pub fn unregister(&self, sub: SubscriptionId) {
191 let mut st = self.state.lock().expect("emitter poisoned");
192 st.bp.unregister(sub);
193 st.subs.retain(|s| *s != sub);
194 st.pending.remove(&sub);
195 st.overflowed.remove(&sub);
196 st.conflating.remove(&sub);
197 }
198
199 /// Whether the connector lost events and no reconcile has cleared it yet: a `Lossless` subscription
200 /// overflowed, or a stream of orders or fills was cut and reopened ([`Event::Reconnected`]).
201 pub fn is_degraded(&self) -> bool {
202 self.degraded.load(Ordering::Acquire)
203 }
204
205 /// Enters the degraded state by hand: the owner knows that events were lost for a reason the emitter
206 /// cannot see (a message that could not be processed). The trading gate closes until the next
207 /// successful reconcile, like after an overflow. Send an [`Event::Overflow`] for the subscription
208 /// to tell the consumer.
209 pub fn mark_degraded(&self) {
210 self.degraded.store(true, Ordering::Release);
211 }
212
213 /// Leaves the degraded state. Call it from a successful reconcile.
214 ///
215 /// Overflowed `Lossless` buffers accept events again and a new overflow sends a new
216 /// [`Event::Overflow`]. Events lost during the overflow are not recovered, the reconcile is
217 /// what makes up for them. Events already waiting in the buffers are still delivered.
218 pub fn clear_degraded(&self) {
219 let mut st = self.state.lock().expect("emitter poisoned");
220 st.bp.clear_degraded();
221 st.overflowed.clear();
222 self.degraded.store(false, Ordering::Release);
223 }
224
225 /// Whether some events or notices are waiting because the channel was full.
226 ///
227 /// While it is `true`, call [`flush`](Self::flush) again later. [`emit`](Self::emit) also
228 /// retries on every call.
229 pub fn has_backlog(&self) -> bool {
230 self.backlog.load(Ordering::Acquire)
231 }
232
233 /// How many events with no subscription were dropped because the channel was full or closed.
234 pub fn dropped_unrouted(&self) -> u64 {
235 self.dropped_unrouted.load(Ordering::Relaxed)
236 }
237
238 fn envelope(&self, sub: Option<SubscriptionId>, payload: Event) -> Envelope {
239 Envelope::new(
240 Arc::clone(&self.source),
241 Timestamps::new(wall_nanos(), crate::now()),
242 payload,
243 )
244 .with_sub_opt(sub)
245 }
246
247 /// Buffers events and moves what the channel accepts. Never blocks.
248 ///
249 /// Events with a subscription go through that subscription's buffer and policy. A `Lossless`
250 /// overflow queues one [`Event::Overflow`] (with `lost_from` set to the current wall clock)
251 /// per subscription and marks the emitter degraded. Events with no subscription are sent
252 /// straight to the channel or dropped, see the type docs. Nothing is returned: the outcome is
253 /// visible through [`is_degraded`](Self::is_degraded), [`has_backlog`](Self::has_backlog) and
254 /// [`dropped_unrouted`](Self::dropped_unrouted).
255 pub fn emit(&self, events: Vec<CoreEvent>) {
256 if events.is_empty() {
257 return;
258 }
259 let mut st = self.state.lock().expect("emitter poisoned");
260 for CoreEvent { sub, payload } in events {
261 // Events were lost while a stream of orders or fills was down: close the trading gate before
262 // the notice can reach the consumer.
263 if matches!(payload, Event::Reconnected { .. }) {
264 self.degraded.store(true, Ordering::Release);
265 }
266 match sub {
267 None => {
268 let env = self.envelope(None, payload);
269 if self.out.try_send(env).is_err() {
270 self.dropped_unrouted.fetch_add(1, Ordering::Relaxed);
271 warn!(
272 "egress channel full or closed: dropping an event with no subscription"
273 );
274 }
275 }
276 Some(sub) => {
277 let now = wall_nanos();
278 if st.conflating.contains(&sub) {
279 // A newer snapshot replaces one the channel has not taken yet.
280 st.pending.remove(&sub);
281 }
282 if st.bp.ingest(sub, payload, now) == PushResult::Overflow {
283 self.degraded.store(true, Ordering::Release);
284 if st.overflowed.insert(sub) {
285 let ev = Event::Overflow {
286 sub,
287 lost_from: now,
288 };
289 let env = self.envelope(Some(sub), ev);
290 st.notices.push_back(env);
291 }
292 }
293 }
294 }
295 }
296 self.flush_locked(&mut st);
297 }
298
299 /// Retries delivery of buffered events and notices. Never blocks.
300 ///
301 /// Call it periodically while [`has_backlog`](Self::has_backlog) is `true`, for example when
302 /// the consumer has read from the channel.
303 pub fn flush(&self) {
304 let mut st = self.state.lock().expect("emitter poisoned");
305 self.flush_locked(&mut st);
306 }
307
308 fn flush_locked(&self, st: &mut State) {
309 while let Some(env) = st.notices.pop_front() {
310 if let Err(e) = self.out.try_send(env) {
311 if let TrySendError::Full(env) = e {
312 st.notices.push_front(env);
313 }
314 break;
315 }
316 }
317 let subs = st.subs.clone();
318 let mut backlog = !st.notices.is_empty();
319 for sub in subs {
320 loop {
321 let env = match st.pending.remove(&sub) {
322 Some(env) => env,
323 None => match st.bp.pop_ready(sub) {
324 Some(ev) => self.envelope(Some(sub), ev),
325 None => break,
326 },
327 };
328 match self.out.try_send(env) {
329 Ok(()) => {}
330 Err(TrySendError::Full(env)) => {
331 st.pending.insert(sub, env);
332 backlog = true;
333 break;
334 }
335 Err(TrySendError::Closed(_)) => break,
336 }
337 }
338 }
339 self.backlog.store(backlog, Ordering::Release);
340 }
341}
342
343#[cfg(test)]
344mod tests {
345 use super::*;
346 use brokapi_core::{Balance, Money};
347 use std::time::Instant;
348
349 fn ev(n: u32) -> CoreEvent {
350 CoreEvent {
351 sub: Some(SubscriptionId::new(1)),
352 payload: Event::Balances(vec![Balance {
353 currency: "RUB".into(),
354 available: Money::from(n),
355 blocked: Money::ZERO,
356 ts: Timestamps::new(wall_nanos(), Instant::now()),
357 }]),
358 }
359 }
360
361 fn avail(env: &Envelope) -> Money {
362 match &env.payload {
363 Event::Balances(b) => b[0].available,
364 other => panic!("{other:?}"),
365 }
366 }
367
368 #[test]
369 fn delivers_with_sub_and_source() {
370 let (tx, mut rx) = mpsc::channel(8);
371 let e = Emitter::new("src".into(), tx);
372 e.register(
373 SubscriptionId::new(1),
374 None,
375 Backpressure::Lossless { cap: 4 },
376 );
377 e.emit(vec![ev(1)]);
378 let env = rx.try_recv().unwrap();
379 assert_eq!(env.sub, Some(SubscriptionId::new(1)));
380 assert_eq!(env.source.as_ref(), "src");
381 }
382
383 #[test]
384 fn lossless_overflow_is_explicit_and_degrades() {
385 let (tx, mut rx) = mpsc::channel(1);
386 let e = Emitter::new("src".into(), tx);
387 e.register(
388 SubscriptionId::new(1),
389 None,
390 Backpressure::Lossless { cap: 2 },
391 );
392 // channel holds 1; buffer holds 2 (+1 popped waiting); then overflow
393 for n in 1..=6 {
394 e.emit(vec![ev(n)]);
395 }
396 assert!(e.is_degraded());
397 assert!(e.has_backlog());
398 let mut seen = Vec::new();
399 let mut overflow = false;
400 for _ in 0..20 {
401 while let Ok(env) = rx.try_recv() {
402 match env.payload {
403 Event::Overflow { sub, .. } => {
404 assert_eq!(sub, SubscriptionId::new(1));
405 overflow = true;
406 }
407 _ => seen.push(avail(&env)),
408 }
409 }
410 e.flush();
411 }
412 assert!(overflow, "Event::Overflow must be delivered");
413 // what was accepted before the overflow arrives in order, nothing after the hole
414 assert_eq!(seen, [1, 2, 3, 4].map(Money::from).to_vec());
415 e.clear_degraded();
416 assert!(!e.is_degraded());
417 }
418
419 #[test]
420 fn a_reconnected_notice_degrades_and_comes_in_order_with_the_events() {
421 let (tx, mut rx) = mpsc::channel(8);
422 let e = Emitter::new("src".into(), tx);
423 let sub = SubscriptionId::new(1);
424 e.register(sub, None, Backpressure::Lossless { cap: 4 });
425 assert!(!e.is_degraded());
426 e.emit(vec![
427 ev(1),
428 CoreEvent {
429 sub: Some(sub),
430 payload: Event::Reconnected {
431 sub,
432 gap_from: wall_nanos(),
433 },
434 },
435 ev(2),
436 ]);
437 // The gate is shut before the consumer can read the notice, and nothing was lost or reordered.
438 assert!(e.is_degraded());
439 assert_eq!(avail(&rx.try_recv().unwrap()), Money::from(1));
440 assert!(matches!(
441 rx.try_recv().unwrap().payload,
442 Event::Reconnected { sub: s, .. } if s == sub
443 ));
444 assert_eq!(avail(&rx.try_recv().unwrap()), Money::from(2));
445 e.clear_degraded();
446 assert!(!e.is_degraded());
447 }
448
449 #[test]
450 fn the_owner_can_mark_the_emitter_degraded() {
451 let (tx, _rx) = mpsc::channel(1);
452 let e = Emitter::new("src".into(), tx);
453 e.mark_degraded();
454 assert!(e.is_degraded());
455 e.clear_degraded();
456 assert!(!e.is_degraded());
457 }
458
459 #[test]
460 fn conflate_keeps_latest_for_slow_consumer() {
461 let (tx, mut rx) = mpsc::channel(1);
462 let e = Emitter::new("src".into(), tx);
463 e.register(SubscriptionId::new(1), None, Backpressure::Conflate);
464 for n in 1..=5 {
465 e.emit(vec![ev(n)]);
466 }
467 assert_eq!(avail(&rx.try_recv().unwrap()), Money::from(1));
468 e.flush();
469 assert_eq!(avail(&rx.try_recv().unwrap()), Money::from(5));
470 assert!(rx.try_recv().is_err());
471 assert!(!e.is_degraded());
472 }
473
474 #[test]
475 fn drop_newest_counts_not_degrades() {
476 let (tx, mut rx) = mpsc::channel(1);
477 let e = Emitter::new("src".into(), tx);
478 e.register(
479 SubscriptionId::new(1),
480 None,
481 Backpressure::DropNewest { cap: 1 },
482 );
483 for n in 1..=5 {
484 e.emit(vec![ev(n)]);
485 }
486 assert!(!e.is_degraded());
487 let mut got = vec![avail(&rx.try_recv().unwrap())];
488 e.flush();
489 while let Ok(env) = rx.try_recv() {
490 got.push(avail(&env));
491 }
492 assert_eq!(got[0], Money::from(1));
493 assert!(got.len() <= 3);
494 }
495}