gsm-core 0.4.12

Core types and platform abstractions for the Greentic messaging runtime.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161

use std::time::{Duration, Instant};

use metrics::counter;
use tracing::{debug, info, warn};

/// Circuit breaker configuration shared across Direct Line polling loops.
#[derive(Clone, Debug)]
pub struct CircuitSettings {
    pub failure_threshold: u32,
    pub open_duration: Duration,
}

impl Default for CircuitSettings {
    fn default() -> Self {
        Self {
            failure_threshold: 5,
            open_duration: Duration::from_secs(30),
        }
    }
}

#[derive(Debug)]
enum CircuitState {
    Closed { consecutive_failures: u32 },
    HalfOpen,
    Open { reopen_at: Instant },
}

/// Lightweight circuit breaker used to guard the Direct Line poll loop.
pub struct CircuitBreaker {
    state: CircuitState,
    settings: CircuitSettings,
    labels: CircuitLabels,
}

impl CircuitBreaker {
    pub fn new(settings: CircuitSettings, labels: CircuitLabels) -> Self {
        Self {
            state: CircuitState::Closed {
                consecutive_failures: 0,
            },
            settings,
            labels,
        }
    }

    pub async fn before_request(&mut self) {
        if let CircuitState::Open { reopen_at } = self.state {
            let now = Instant::now();
            if reopen_at > now {
                let sleep = reopen_at - now;
                debug!(?sleep, "circuit breaker sleeping before half-open probe");
                tokio::time::sleep(sleep).await;
            }
            self.state = CircuitState::HalfOpen;
            info!(
                env = self.labels.env,
                tenant = self.labels.tenant,
                team = self.labels.team,
                conversation_id = self.labels.conversation_id,
                "circuit breaker half-open probe"
            );
        }
    }

    pub fn on_success(&mut self) {
        match self.state {
            CircuitState::Closed {
                ref mut consecutive_failures,
            } => {
                if *consecutive_failures > 0 {
                    debug!(
                        failures = *consecutive_failures,
                        "resetting failure counter"
                    );
                }
                *consecutive_failures = 0;
            }
            CircuitState::HalfOpen | CircuitState::Open { .. } => {
                info!(
                    env = self.labels.env,
                    tenant = self.labels.tenant,
                    team = self.labels.team,
                    conversation_id = self.labels.conversation_id,
                    "circuit breaker closed"
                );
                counter!(
                    "webchat_circuit_events_total",
                    "state" => "closed",
                    "env" => self.labels.env.clone(),
                    "tenant" => self.labels.tenant.clone(),
                    "team" => self.labels.team.clone(),
                    "conversation" => self.labels.conversation_id.clone(),
                )
                .increment(1);
                self.state = CircuitState::Closed {
                    consecutive_failures: 0,
                };
            }
        }
    }

    pub fn on_failure(&mut self) {
        match self.state {
            CircuitState::Closed {
                ref mut consecutive_failures,
            } => {
                *consecutive_failures += 1;
                if *consecutive_failures >= self.settings.failure_threshold {
                    self.open();
                }
            }
            CircuitState::HalfOpen => {
                self.open();
            }
            CircuitState::Open { .. } => {}
        }
    }

    fn open(&mut self) {
        let reopen_at = Instant::now() + self.settings.open_duration;
        self.state = CircuitState::Open { reopen_at };
        warn!(
            env = self.labels.env,
            tenant = self.labels.tenant,
            team = self.labels.team,
            conversation_id = self.labels.conversation_id,
            reopen_in = ?self.settings.open_duration,
            "circuit breaker opened"
        );
        counter!(
            "webchat_circuit_events_total",
            "state" => "open",
            "env" => self.labels.env.clone(),
            "tenant" => self.labels.tenant.clone(),
            "team" => self.labels.team.clone(),
            "conversation" => self.labels.conversation_id.clone(),
        )
        .increment(1);
    }
}

/// Metric label bundle carried by the circuit breaker.
#[derive(Clone, Debug)]
pub struct CircuitLabels {
    pub env: String,
    pub tenant: String,
    pub team: String,
    pub conversation_id: String,
}

impl CircuitLabels {
    pub fn new(env: String, tenant: String, team: String, conversation_id: String) -> Self {
        Self {
            env,
            tenant,
            team,
            conversation_id,
        }
    }
}