brokapi_runtime/
supervisor.rs

1//! Supervisor reconcile tick.
2
3use std::time::{Duration, Instant};
4
5use brokapi_sansio::SubscriptionSet;
6
7use crate::connection::ConnectionState;
8use crate::health::HealthClass;
9
10/// What the owner should do for one connection after a [`Supervisor::reconcile`] pass.
11///
12/// The supervisor only names the action. The owner performs it.
13#[derive(Debug, Clone, PartialEq, Eq)]
14pub enum ReconcileAction {
15    /// The connection is healthy. Make the wire match the desired set: send what
16    /// [`Supervisor::resubscribe_targets`] lists and cancel what was removed.
17    ApplyDesired,
18    /// The connection has been quiet for a while. Re-send the subscriptions that are missing,
19    /// without a reconnect.
20    ResubscribeMissing,
21    /// The connection is dead. Close it and open a new one, then apply the desired set.
22    ScheduleReconnect,
23}
24
25/// Connection supervisor without I/O: it compares desired and actual state and says what to do.
26///
27/// Call [`reconcile`](Self::reconcile) on a timer (every [`reconcile_interval`](Self::reconcile_interval)).
28/// It returns one [`ReconcileAction`] per connection and changes nothing on the wire. The decision
29/// is made by the connection's silence: see [`HealthClass`].
30///
31/// A new supervisor holds one connection with id `1`.
32///
33/// # Examples
34///
35/// ```
36/// use std::time::{Duration, Instant};
37/// use brokapi_core::{InstrumentId, InstrumentKey, NativeId, SubscriptionId};
38/// use brokapi_runtime::{ReconcileAction, Supervisor};
39/// use brokapi_sansio::{DesiredSubscription, SubscriptionKind, SubscriptionSet};
40///
41/// let sber = InstrumentId::new(
42///     InstrumentKey::new("SBER", "TQBR"),
43///     NativeId::TickerClass { ticker: "SBER".into(), class_code: "TQBR".into() },
44/// );
45/// let sub = DesiredSubscription {
46///     id: SubscriptionId::new(1),
47///     kind: SubscriptionKind::Quotes { instrument: sber },
48/// };
49///
50/// let t0 = Instant::now();
51/// let mut sup = Supervisor::new(t0);
52/// sup.update_desired(1, SubscriptionSet::with_subscriptions(vec![sub.clone()]));
53///
54/// // Fresh connection: apply the desired set. One subscription is missing on the wire.
55/// assert_eq!(sup.reconcile(t0), vec![(1, ReconcileAction::ApplyDesired)]);
56/// assert_eq!(sup.resubscribe_targets(1), vec![sub]);
57///
58/// // Silent for longer than `dead_after` (60 s by default): replace the connection.
59/// let later = t0 + Duration::from_secs(61);
60/// assert_eq!(sup.reconcile(later), vec![(1, ReconcileAction::ScheduleReconnect)]);
61/// ```
62#[derive(Debug)]
63pub struct Supervisor {
64    /// How often the owner should call [`reconcile`](Self::reconcile). Default 1 second. The
65    /// supervisor does not read it itself.
66    pub reconcile_interval: Duration,
67    /// Silence after which a connection is [`HealthClass::Dead`]. Default 60 seconds.
68    pub dead_after: Duration,
69    /// Silence after which a connection is [`HealthClass::Degraded`]. Default 15 seconds.
70    pub degraded_after: Duration,
71    /// The supervised connections.
72    pub connections: Vec<ConnectionState>,
73}
74
75impl Supervisor {
76    /// Creates a supervisor with one connection (id `1`) last heard from at `now`, and the default
77    /// timings listed on the fields.
78    pub fn new(now: Instant) -> Self {
79        Self {
80            reconcile_interval: Duration::from_secs(1),
81            dead_after: Duration::from_secs(60),
82            degraded_after: Duration::from_secs(15),
83            connections: vec![ConnectionState::new(1, now)],
84        }
85    }
86
87    /// The connection with this id, if there is one.
88    pub fn connection_mut(&mut self, id: u64) -> Option<&mut ConnectionState> {
89        self.connections.iter_mut().find(|c| c.id == id)
90    }
91
92    /// Decides what each connection needs at `now`. Pure: the connections are not changed.
93    ///
94    /// The silence of the connection picks the action: healthy gives
95    /// [`ReconcileAction::ApplyDesired`], degraded gives [`ReconcileAction::ResubscribeMissing`], dead
96    /// gives [`ReconcileAction::ScheduleReconnect`]. One entry per connection, in connection order.
97    pub fn reconcile(&mut self, now: Instant) -> Vec<(u64, ReconcileAction)> {
98        let mut actions = Vec::new();
99        for conn in &mut self.connections {
100            let health = conn.classify(now, self.dead_after, self.degraded_after);
101            let action = match health {
102                HealthClass::Healthy => ReconcileAction::ApplyDesired,
103                HealthClass::Degraded => ReconcileAction::ResubscribeMissing,
104                HealthClass::Dead => ReconcileAction::ScheduleReconnect,
105            };
106            actions.push((conn.id, action));
107        }
108        actions
109    }
110
111    /// Sets the desired subscriptions of a connection. An unknown `conn_id` is ignored.
112    pub fn update_desired(&mut self, conn_id: u64, desired: SubscriptionSet) {
113        if let Some(conn) = self.connection_mut(conn_id) {
114            conn.set_desired(desired);
115        }
116    }
117
118    /// The subscriptions of a connection that must be (re)sent, see
119    /// [`ConnectionState::needs_resubscribe`]. Empty for an unknown `conn_id`.
120    pub fn resubscribe_targets(&self, conn_id: u64) -> Vec<brokapi_sansio::DesiredSubscription> {
121        self.connections
122            .iter()
123            .find(|c| c.id == conn_id)
124            .map(ConnectionState::needs_resubscribe)
125            .unwrap_or_default()
126    }
127}
128
129#[cfg(test)]
130mod tests {
131    use super::*;
132    use brokapi_core::{InstrumentId, InstrumentKey, NativeId, SubscriptionId};
133    use brokapi_sansio::{DesiredSubscription, SubscriptionKind};
134
135    fn sber() -> InstrumentId {
136        InstrumentId::new(
137            InstrumentKey::new("SBER", "TQBR"),
138            NativeId::TickerClass {
139                ticker: "SBER".into(),
140                class_code: "TQBR".into(),
141            },
142        )
143    }
144
145    #[test]
146    fn failed_subscription_resubscribed_without_reconnect() {
147        let now = Instant::now();
148        let mut sup = Supervisor::new(now);
149        let sub = DesiredSubscription {
150            id: SubscriptionId::new(1),
151            kind: SubscriptionKind::Quotes { instrument: sber() },
152        };
153        sup.update_desired(1, SubscriptionSet::with_subscriptions(vec![sub.clone()]));
154        sup.connection_mut(1).unwrap().mark_failed(sub.id, "denied");
155        let targets = sup.resubscribe_targets(1);
156        assert_eq!(targets.len(), 1);
157        let actions = sup.reconcile(now);
158        assert!(
159            actions
160                .iter()
161                .any(|(_, a)| *a == ReconcileAction::ApplyDesired)
162        );
163    }
164}
brokapi_runtime/supervisor.rs

brokapi_runtime/
supervisor.rs
