brokapi_runtime/backpressure.rs
1//! Per-subscription backpressure buffers
2
3use std::collections::VecDeque;
4
5use brokapi_core::{Backpressure, Event, InstrumentId, SubscriptionId, UnixNanos};
6
7/// What happened to one event offered to a buffer. Returned by [`SubscriptionBuffer::try_push`] and
8/// [`BackpressureManager::ingest`].
9///
10/// The meaning depends on the subscription's [`Backpressure`] policy:
11///
12/// | Policy | Possible results |
13/// |---|---|
14/// | [`Backpressure::Conflate`] | always [`Conflated`](Self::Conflated) |
15/// | [`Backpressure::Lossless`] | [`Delivered`](Self::Delivered), then [`Overflow`](Self::Overflow) for ever |
16/// | [`Backpressure::DropNewest`] | [`Delivered`](Self::Delivered) or [`DroppedNewest`](Self::DroppedNewest) |
17#[derive(Debug, Clone, PartialEq, Eq)]
18pub enum PushResult {
19 /// The event is in the buffer and will be delivered in order.
20 Delivered,
21 /// The event is the buffer's "latest" value of its kind. It replaced the previous one of that kind, if any.
22 /// This is the only result of a `Conflate` subscription, even for its first event.
23 Conflated,
24 /// The buffer was full and the new event was thrown away. Older events are kept. The loss is
25 /// counted in [`SubscriptionBuffer::dropped`]. The connector does not become degraded.
26 DroppedNewest,
27 /// A `Lossless` buffer was full. The event was not stored. This is terminal until the buffer is
28 /// reset: every later event of this subscription is refused too, and the owner must tell the
29 /// consumer with [`Event::Overflow`] and mark the connector degraded (see [`BackpressureManager`]).
30 Overflow,
31}
32
33/// Egress buffer of one subscription. It applies the subscription's [`Backpressure`] policy.
34///
35/// The buffer never blocks and never waits. It is a plain in-memory queue (or single slot), not
36/// thread-safe by itself. [`BackpressureManager`] owns many of them, and the emitter wraps the manager in a
37/// lock.
38///
39/// # Examples
40///
41/// ```
42/// use brokapi_core::{
43/// Backpressure, Event, InstrumentId, InstrumentKey, NativeId, SubscriptionId,
44/// };
45/// use brokapi_runtime::{PushResult, SubscriptionBuffer};
46///
47/// let sber = InstrumentId::new(
48/// InstrumentKey::new("SBER", "TQBR"),
49/// NativeId::TickerClass { ticker: "SBER".into(), class_code: "TQBR".into() },
50/// );
51/// let mut buf = SubscriptionBuffer::new(SubscriptionId::new(1), sber, Backpressure::DropNewest { cap: 1 });
52///
53/// assert_eq!(buf.try_push(Event::Balances(vec![])), PushResult::Delivered);
54/// // The buffer holds one event. The next one is thrown away and counted.
55/// assert_eq!(buf.try_push(Event::Balances(vec![])), PushResult::DroppedNewest);
56/// assert_eq!(buf.dropped(), 1);
57/// assert_eq!(buf.drain().len(), 1);
58/// ```
59#[derive(Debug)]
60pub struct SubscriptionBuffer {
61 sub: SubscriptionId,
62 instrument: InstrumentId,
63 policy: Backpressure,
64 queue: VecDeque<Event>,
65 /// `Conflate`: the latest event of each kind, in the order the kinds first appeared.
66 latest: Vec<(std::mem::Discriminant<Event>, Event)>,
67 conflated: u64,
68 dropped: u64,
69 overflowed: bool,
70}
71
72impl SubscriptionBuffer {
73 /// Creates an empty buffer for a subscription with the given policy.
74 ///
75 /// A `cap` of `0` in `Lossless` or `DropNewest` makes a buffer that accepts nothing: the first
76 /// event already overflows (or is dropped).
77 pub fn new(sub: SubscriptionId, instrument: InstrumentId, policy: Backpressure) -> Self {
78 Self {
79 sub,
80 instrument,
81 policy,
82 queue: VecDeque::new(),
83 latest: Vec::new(),
84 conflated: 0,
85 dropped: 0,
86 overflowed: false,
87 }
88 }
89
90 /// The subscription this buffer serves.
91 pub fn sub_id(&self) -> SubscriptionId {
92 self.sub
93 }
94
95 /// The instrument of the subscription, as given to [`new`](Self::new).
96 pub fn instrument(&self) -> &InstrumentId {
97 &self.instrument
98 }
99
100 /// Whether a `Lossless` buffer has overflowed and not been reset yet.
101 pub fn is_overflowed(&self) -> bool {
102 self.overflowed
103 }
104
105 /// How many events a `Conflate` buffer replaced before they were taken.
106 pub fn conflated(&self) -> u64 {
107 self.conflated
108 }
109
110 /// How many events a `DropNewest` buffer threw away.
111 pub fn dropped(&self) -> u64 {
112 self.dropped
113 }
114
115 /// How many events wait in the buffer. For `Conflate` this is `0` or `1`.
116 pub fn len(&self) -> usize {
117 match self.policy {
118 Backpressure::Conflate => self.latest.len(),
119 _ => self.queue.len(),
120 }
121 }
122
123 /// Whether the buffer holds no event.
124 pub fn is_empty(&self) -> bool {
125 self.len() == 0
126 }
127
128 /// Offers an event to the buffer without blocking.
129 ///
130 /// - `Conflate`: the event becomes the latest value of its kind (an order book, a quote, a portfolio
131 /// and so on: one slot per kind). A value of the same kind that was still waiting is
132 /// replaced and counted in [`conflated`](Self::conflated). Result: [`PushResult::Conflated`].
133 /// - `Lossless { cap }`: stored while fewer than `cap` events wait. The first event that finds
134 /// the buffer full is not stored, the buffer is flagged overflowed and the result is
135 /// [`PushResult::Overflow`]. After that every event gets `Overflow` and is not stored.
136 /// - `DropNewest { cap }`: stored while fewer than `cap` events wait. Otherwise the new event is
137 /// thrown away and counted in [`dropped`](Self::dropped).
138 ///
139 /// Why `Lossless` stops for good: it is used for events that must not have a hole (orders,
140 /// fills). Once one is lost, later events would hide the gap, so the consumer must be told and
141 /// re-read the state.
142 ///
143 /// # Examples
144 ///
145 /// ```
146 /// use brokapi_core::{Backpressure, Event, InstrumentId, InstrumentKey, NativeId, SubscriptionId};
147 /// use brokapi_runtime::{PushResult, SubscriptionBuffer};
148 ///
149 /// let sber = InstrumentId::new(
150 /// InstrumentKey::new("SBER", "TQBR"),
151 /// NativeId::TickerClass { ticker: "SBER".into(), class_code: "TQBR".into() },
152 /// );
153 /// let mut buf = SubscriptionBuffer::new(SubscriptionId::new(1), sber, Backpressure::Lossless { cap: 1 });
154 ///
155 /// assert_eq!(buf.try_push(Event::Balances(vec![])), PushResult::Delivered);
156 /// assert_eq!(buf.try_push(Event::Balances(vec![])), PushResult::Overflow);
157 /// assert!(buf.is_overflowed());
158 /// // Even after the consumer takes the stored event, the buffer stays overflowed.
159 /// assert_eq!(buf.drain().len(), 1);
160 /// assert_eq!(buf.try_push(Event::Balances(vec![])), PushResult::Overflow);
161 /// ```
162 pub fn try_push(&mut self, event: Event) -> PushResult {
163 match self.policy {
164 Backpressure::Conflate => {
165 // One slot per kind of event: a subscription can carry several (T-Bank sends portfolio,
166 // positions and balances on one), and a newer balance must not replace a position.
167 let kind = std::mem::discriminant(&event);
168 match self.latest.iter_mut().find(|(k, _)| *k == kind) {
169 Some(slot) => {
170 slot.1 = event;
171 self.conflated += 1;
172 }
173 None => self.latest.push((kind, event)),
174 }
175 PushResult::Conflated
176 }
177 Backpressure::Lossless { cap } => {
178 if self.overflowed {
179 return PushResult::Overflow;
180 }
181 if self.queue.len() >= cap {
182 self.overflowed = true;
183 PushResult::Overflow
184 } else {
185 self.queue.push_back(event);
186 PushResult::Delivered
187 }
188 }
189 Backpressure::DropNewest { cap } if self.queue.len() >= cap => {
190 self.dropped += 1;
191 PushResult::DroppedNewest
192 }
193 // `DropNewest` with room, and a policy of a later version: keep everything, the safe reading.
194 _ => {
195 self.queue.push_back(event);
196 PushResult::Delivered
197 }
198 }
199 }
200
201 /// Builds the [`Event::Overflow`] notice for this subscription.
202 ///
203 /// `lost_from` is the wall-clock time from which events may be missing. The buffer does not
204 /// send the notice itself, the owner does.
205 ///
206 /// # Examples
207 ///
208 /// ```
209 /// use brokapi_core::{Backpressure, Event, InstrumentId, InstrumentKey, NativeId, SubscriptionId, UnixNanos};
210 /// use brokapi_runtime::SubscriptionBuffer;
211 ///
212 /// let sber = InstrumentId::new(
213 /// InstrumentKey::new("SBER", "TQBR"),
214 /// NativeId::TickerClass { ticker: "SBER".into(), class_code: "TQBR".into() },
215 /// );
216 /// let buf = SubscriptionBuffer::new(SubscriptionId::new(7), sber, Backpressure::Lossless { cap: 1 });
217 /// let notice = buf.overflow_event(UnixNanos::from_nanos(5));
218 /// assert!(matches!(notice, Event::Overflow { sub, .. } if sub == SubscriptionId::new(7)));
219 /// ```
220 pub fn overflow_event(&self, lost_from: UnixNanos) -> Event {
221 Event::Overflow {
222 sub: self.sub,
223 lost_from,
224 }
225 }
226
227 /// Takes everything out of the buffer: the latest conflated value first, then the queue in
228 /// arrival order.
229 ///
230 /// Draining does not clear the overflowed flag. Use [`BackpressureManager::clear_degraded`] for that.
231 pub fn drain(&mut self) -> Vec<Event> {
232 let mut out = Vec::new();
233 out.extend(self.latest.drain(..).map(|(_, ev)| ev));
234 out.extend(self.queue.drain(..));
235 out
236 }
237}
238
239/// The buffers of all active subscriptions of one connector, plus the connector's degraded flag.
240///
241/// Used from the parser task: [`ingest`](Self::ingest) puts an event into the buffer of its
242/// subscription without blocking, and [`pop_ready`](Self::pop_ready) takes the next one out for
243/// delivery. When any `Lossless` buffer overflows, the manager marks the whole connector degraded.
244/// A degraded connector refuses trading commands with [`brokapi_core::Error::Degraded`] (the
245/// caller applies that, for example through [`CommandGate`](crate::CommandGate)) until a successful
246/// reconcile calls [`clear_degraded`](Self::clear_degraded).
247///
248/// Most connectors use [`Emitter`](crate::Emitter), which owns a manager and also moves events
249/// into the output channel.
250///
251/// # Examples
252///
253/// ```
254/// use brokapi_core::{Backpressure, Event, InstrumentId, InstrumentKey, NativeId, SubscriptionId, UnixNanos};
255/// use brokapi_runtime::{BackpressureManager, PushResult};
256///
257/// let sber = InstrumentId::new(
258/// InstrumentKey::new("SBER", "TQBR"),
259/// NativeId::TickerClass { ticker: "SBER".into(), class_code: "TQBR".into() },
260/// );
261/// let sub = SubscriptionId::new(1);
262/// let mut mgr = BackpressureManager::new();
263/// mgr.register(sub, sber, Backpressure::Lossless { cap: 1 });
264///
265/// assert_eq!(mgr.ingest(sub, Event::Balances(vec![]), UnixNanos::ZERO), PushResult::Delivered);
266/// assert_eq!(mgr.ingest(sub, Event::Balances(vec![]), UnixNanos::ZERO), PushResult::Overflow);
267/// assert!(mgr.is_degraded());
268///
269/// // The reconcile that follows an overflow clears the flag.
270/// mgr.clear_degraded();
271/// assert!(!mgr.is_degraded());
272/// ```
273#[derive(Debug, Default)]
274pub struct BackpressureManager {
275 buffers: Vec<SubscriptionBuffer>,
276 connector_degraded: bool,
277}
278
279impl BackpressureManager {
280 /// Creates a manager with no subscriptions, not degraded.
281 pub fn new() -> Self {
282 Self::default()
283 }
284
285 /// Adds a buffer for a subscription.
286 ///
287 /// Registering the same `sub` twice creates two buffers. Only the first one is ever used, so
288 /// register each subscription id once.
289 pub fn register(
290 &mut self,
291 sub: SubscriptionId,
292 instrument: InstrumentId,
293 policy: Backpressure,
294 ) {
295 self.buffers
296 .push(SubscriptionBuffer::new(sub, instrument, policy));
297 }
298
299 /// Removes the buffer of a subscription. Events still waiting in it are discarded.
300 /// An unknown `sub` is ignored.
301 pub fn unregister(&mut self, sub: SubscriptionId) {
302 self.buffers.retain(|b| b.sub_id() != sub);
303 }
304
305 /// Whether some `Lossless` buffer overflowed since the last [`clear_degraded`](Self::clear_degraded).
306 pub fn is_degraded(&self) -> bool {
307 self.connector_degraded
308 }
309
310 /// Leaves the degraded state and reopens every overflowed buffer.
311 ///
312 /// Call it after a successful reconcile, when the consumer has re-read the state it lost.
313 /// Events that were refused during the overflow are not recovered.
314 pub fn clear_degraded(&mut self) {
315 self.connector_degraded = false;
316 for buf in &mut self.buffers {
317 buf.overflowed = false;
318 }
319 }
320
321 /// Offers an event to the buffer of `sub` without blocking.
322 ///
323 /// An [`PushResult::Overflow`] also sets the degraded flag of the manager. The `lost_from`
324 /// argument is accepted for the caller's convenience and is not used by the manager: the
325 /// caller builds the [`Event::Overflow`] notice itself (see [`SubscriptionBuffer::overflow_event`]).
326 ///
327 /// An event for a `sub` that is not registered is discarded and the result is still
328 /// [`PushResult::Delivered`]. Register the subscription before its first event can arrive.
329 pub fn ingest(
330 &mut self,
331 sub: SubscriptionId,
332 event: Event,
333 lost_from: UnixNanos,
334 ) -> PushResult {
335 let Some(buf) = self.buffers.iter_mut().find(|b| b.sub_id() == sub) else {
336 return PushResult::Delivered;
337 };
338 let result = buf.try_push(event);
339 if result == PushResult::Overflow {
340 self.connector_degraded = true;
341 let _ = lost_from;
342 }
343 result
344 }
345
346 /// Takes the next event of `sub` that is ready for delivery, if any.
347 ///
348 /// A `Conflate` subscription yields its latest value of each kind, one per call. Others yield their queue front, oldest
349 /// first. Returns `None` for an empty buffer and for an unknown `sub`.
350 ///
351 /// # Examples
352 ///
353 /// ```
354 /// use brokapi_core::{Backpressure, Event, InstrumentId, InstrumentKey, NativeId, SubscriptionId, UnixNanos};
355 /// use brokapi_runtime::BackpressureManager;
356 ///
357 /// let sber = InstrumentId::new(
358 /// InstrumentKey::new("SBER", "TQBR"),
359 /// NativeId::TickerClass { ticker: "SBER".into(), class_code: "TQBR".into() },
360 /// );
361 /// let sub = SubscriptionId::new(1);
362 /// let mut mgr = BackpressureManager::new();
363 /// mgr.register(sub, sber, Backpressure::Conflate);
364 ///
365 /// // Three snapshots arrive before the consumer looks. Only the last one is kept.
366 /// for _ in 0..3 {
367 /// mgr.ingest(sub, Event::Balances(vec![]), UnixNanos::ZERO);
368 /// }
369 /// assert!(mgr.pop_ready(sub).is_some());
370 /// assert!(mgr.pop_ready(sub).is_none());
371 /// ```
372 pub fn pop_ready(&mut self, sub: SubscriptionId) -> Option<Event> {
373 let buf = self.buffers.iter_mut().find(|b| b.sub_id() == sub)?;
374 if !buf.latest.is_empty() {
375 return Some(buf.latest.remove(0).1);
376 }
377 buf.queue.pop_front()
378 }
379}
380
381#[cfg(test)]
382mod tests {
383 use super::*;
384 use brokapi_core::{InstrumentKey, NativeId};
385
386 fn instrument() -> InstrumentId {
387 InstrumentId::new(
388 InstrumentKey::new("SBER", "TQBR"),
389 NativeId::TickerClass {
390 ticker: "SBER".into(),
391 class_code: "TQBR".into(),
392 },
393 )
394 }
395
396 #[test]
397 fn lossless_overflow_marks_degraded() {
398 let mut mgr = BackpressureManager::new();
399 let sub = SubscriptionId::new(1);
400 mgr.register(sub, instrument(), Backpressure::Lossless { cap: 2 });
401
402 assert_eq!(
403 mgr.ingest(sub, Event::Balances(vec![]), UnixNanos::ZERO),
404 PushResult::Delivered
405 );
406 assert_eq!(
407 mgr.ingest(sub, Event::Balances(vec![]), UnixNanos::ZERO),
408 PushResult::Delivered
409 );
410 assert_eq!(
411 mgr.ingest(sub, Event::Balances(vec![]), UnixNanos::from_nanos(1)),
412 PushResult::Overflow
413 );
414 assert!(mgr.is_degraded());
415 }
416}