use std::sync::atomic::{AtomicU64, AtomicU8, Ordering};
use std::sync::{Arc, OnceLock};
use std::time::{Duration, Instant};
use backon::{ExponentialBuilder, Retryable};
use parking_lot::Mutex;
use tap::TapOptional;
use crate::telemetry::GatewayMetrics;
#[derive(Debug, Clone, Copy)]
pub enum Profile {
Default,
Fast,
Aggressive,
}
#[derive(Debug, Clone)]
pub struct ResiliencePolicy {
pub max_attempts: u32,
pub initial_delay: Duration,
pub max_delay: Duration,
pub multiplier: f32,
pub jitter: bool,
pub breaker_threshold: u32,
pub breaker_open_for: Duration,
}
impl Profile {
pub fn policy(self) -> ResiliencePolicy {
match self {
Profile::Default => ResiliencePolicy {
max_attempts: 3,
initial_delay: Duration::from_millis(200),
max_delay: Duration::from_secs(5),
multiplier: 2.0,
jitter: true,
breaker_threshold: 5,
breaker_open_for: Duration::from_secs(30),
},
Profile::Fast => ResiliencePolicy {
max_attempts: 1,
initial_delay: Duration::from_millis(0),
max_delay: Duration::from_millis(0),
multiplier: 1.0,
jitter: false,
breaker_threshold: 10,
breaker_open_for: Duration::from_secs(15),
},
Profile::Aggressive => ResiliencePolicy {
max_attempts: 5,
initial_delay: Duration::from_millis(100),
max_delay: Duration::from_secs(10),
multiplier: 2.0,
jitter: true,
breaker_threshold: 10,
breaker_open_for: Duration::from_secs(60),
},
}
}
}
#[derive(Debug, thiserror::Error)]
pub enum ResilienceError<E = reqwest::Error> {
#[error("upstream call failed after retries: {0}")]
Exhausted(E),
#[error("circuit breaker {name} is open")]
CircuitOpen { name: String },
}
impl From<reqwest::Error> for ResilienceError {
fn from(e: reqwest::Error) -> Self {
ResilienceError::Exhausted(e)
}
}
pub async fn run<F, Fut, T>(
op: F,
profile: Profile,
breaker: &CircuitBreaker,
label: &'static str,
) -> Result<T, ResilienceError>
where
F: FnMut() -> Fut + Send,
Fut: std::future::Future<Output = Result<T, reqwest::Error>> + Send,
{
run_with_classifier(op, profile, breaker, label, is_retryable_reqwest).await
}
pub async fn run_with_classifier<T, E, F, Fut, Cls>(
op: F,
profile: Profile,
breaker: &CircuitBreaker,
label: &'static str,
is_retryable: Cls,
) -> Result<T, ResilienceError<E>>
where
F: FnMut() -> Fut + Send,
Fut: std::future::Future<Output = Result<T, E>> + Send,
E: std::fmt::Debug + Send,
Cls: Fn(&E) -> bool + Send + Sync,
{
let start = Instant::now();
if breaker.guard().is_err() {
record_call_metrics(breaker, label, "circuit_open", start.elapsed());
return Err(ResilienceError::CircuitOpen {
name: breaker.name.to_string(),
});
}
let policy = profile.policy();
let mut builder = ExponentialBuilder::default()
.with_max_times(policy.max_attempts.saturating_sub(1) as usize)
.with_min_delay(policy.initial_delay)
.with_max_delay(policy.max_delay)
.with_factor(policy.multiplier);
if policy.jitter {
builder = builder.with_jitter();
}
let mut attempt: u32 = 0;
let result = op
.retry(builder)
.when(&is_retryable)
.notify(|err, dur| {
attempt += 1;
breaker.emit(|m| m.retry_attempt(label));
tracing::warn!(
label,
attempt,
next_delay_ms = dur.as_millis() as u64,
error = ?err,
"retrying outbound call",
);
})
.await;
breaker.record(&result);
let outcome = if result.is_ok() {
"success"
} else {
"exhausted"
};
record_call_metrics(breaker, label, outcome, start.elapsed());
result.map_err(ResilienceError::Exhausted)
}
fn record_call_metrics(
breaker: &CircuitBreaker,
label: &'static str,
outcome: &'static str,
elapsed: Duration,
) {
breaker.emit(|m| m.resilience_call(label, outcome, elapsed.as_secs_f64()));
}
pub fn is_retryable_reqwest(e: &reqwest::Error) -> bool {
if e.is_timeout() || e.is_connect() {
return true;
}
if let Some(status) = e.status() {
return status.is_server_error()
|| status == reqwest::StatusCode::REQUEST_TIMEOUT
|| status == reqwest::StatusCode::TOO_MANY_REQUESTS;
}
false
}
pub struct CircuitBreaker {
pub(crate) name: &'static str,
pub(crate) threshold: u32,
pub(crate) open_for: Duration,
pub(crate) consecutive_failures: AtomicU64,
pub(crate) state: AtomicU8,
pub(crate) opened_at: Mutex<Option<Instant>>,
pub(crate) metrics: OnceLock<Arc<GatewayMetrics>>,
}
pub(crate) const STATE_CLOSED: u8 = 0;
pub(crate) const STATE_OPEN: u8 = 1;
pub(crate) const STATE_HALF_OPEN: u8 = 2;
impl CircuitBreaker {
pub fn new(name: &'static str, profile: Profile) -> Self {
let p = profile.policy();
Self {
name,
threshold: p.breaker_threshold,
open_for: p.breaker_open_for,
consecutive_failures: AtomicU64::new(0),
state: AtomicU8::new(STATE_CLOSED),
opened_at: Mutex::new(None),
metrics: OnceLock::new(),
}
}
pub fn attach_metrics(&self, metrics: Arc<GatewayMetrics>) {
let _ = self.metrics.set(metrics);
}
fn emit(&self, f: impl FnOnce(&GatewayMetrics)) {
self.metrics.get().tap_some(|m| f(m));
}
fn record_transition(&self, transition: &'static str, new: u8) {
self.emit(|m| m.breaker_transition(self.name, transition, new));
}
pub fn guard(&self) -> Result<(), ResilienceError> {
match self.state.load(Ordering::Acquire) {
STATE_CLOSED | STATE_HALF_OPEN => Ok(()),
STATE_OPEN => {
let opened_at = self.opened_at.lock();
let opened = opened_at.unwrap_or_else(Instant::now);
drop(opened_at);
if Instant::now().duration_since(opened) >= self.open_for {
self.state.swap(STATE_HALF_OPEN, Ordering::AcqRel);
tracing::info!(name = self.name, "circuit breaker half-open");
self.record_transition("half_open", STATE_HALF_OPEN);
Ok(())
} else {
Err(ResilienceError::CircuitOpen {
name: self.name.to_string(),
})
}
}
_ => Ok(()),
}
}
pub fn record<T, E>(&self, result: &Result<T, E>) {
match result {
Ok(_) => {
self.consecutive_failures.store(0, Ordering::Release);
let prev = self.state.swap(STATE_CLOSED, Ordering::AcqRel);
if prev == STATE_HALF_OPEN {
tracing::info!(name = self.name, "circuit breaker closed");
self.record_transition("closed", STATE_CLOSED);
}
}
Err(_) => {
let n = self.consecutive_failures.fetch_add(1, Ordering::AcqRel) + 1;
if n >= self.threshold as u64 {
let prev = self.state.swap(STATE_OPEN, Ordering::AcqRel);
*self.opened_at.lock() = Some(Instant::now());
if prev != STATE_OPEN {
tracing::warn!(
name = self.name,
consecutive_failures = n,
"circuit breaker opened",
);
self.record_transition("open", STATE_OPEN);
}
}
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::atomic::{AtomicU32, Ordering};
async fn fetch(url: &str) -> reqwest::Result<reqwest::Response> {
reqwest::Client::new()
.get(url)
.timeout(Duration::from_millis(50))
.send()
.await
}
#[tokio::test]
async fn is_retryable_timeout_is_true() {
let err = fetch("http://10.255.255.1:1/").await.unwrap_err();
assert!(
is_retryable_reqwest(&err),
"expected timeout/connect to be retryable, got {err:?}"
);
}
#[tokio::test]
async fn is_retryable_5xx_is_true() {
use wiremock::{matchers::method, Mock, MockServer, ResponseTemplate};
let server = MockServer::start().await;
Mock::given(method("GET"))
.respond_with(ResponseTemplate::new(503))
.mount(&server)
.await;
let resp = reqwest::Client::new()
.get(server.uri())
.send()
.await
.unwrap();
let err = resp.error_for_status().unwrap_err();
assert!(is_retryable_reqwest(&err));
}
#[tokio::test]
async fn run_retries_until_success() {
use wiremock::{matchers::method, Mock, MockServer, ResponseTemplate};
let server = MockServer::start().await;
Mock::given(method("GET"))
.respond_with(ResponseTemplate::new(503))
.up_to_n_times(2)
.mount(&server)
.await;
Mock::given(method("GET"))
.respond_with(ResponseTemplate::new(200))
.mount(&server)
.await;
let breaker = CircuitBreaker::new("test", Profile::Default);
let attempts = AtomicU32::new(0);
let result = run(
|| async {
attempts.fetch_add(1, Ordering::SeqCst);
reqwest::Client::new()
.get(server.uri())
.send()
.await?
.error_for_status()
},
Profile::Default,
&breaker,
"test",
)
.await;
assert!(result.is_ok(), "got {result:?}");
assert_eq!(attempts.load(Ordering::SeqCst), 3);
}
#[test]
fn breaker_opens_after_threshold() {
let b = CircuitBreaker::new("t", Profile::Default);
for _ in 0..5 {
let err: Result<(), reqwest::Error> = Err(make_fake_reqwest_error());
b.record(&err);
}
assert!(matches!(
b.guard(),
Err(ResilienceError::CircuitOpen { .. })
));
}
fn make_fake_reqwest_error() -> reqwest::Error {
reqwest::Client::new()
.get("http://invalid url")
.build()
.unwrap_err()
}
#[tokio::test]
async fn run_with_classifier_retries_on_classified_retryable() {
#[derive(Debug)]
struct MyErr(bool );
let breaker = CircuitBreaker::new("test-classifier", Profile::Fast);
let attempts = std::sync::atomic::AtomicU32::new(0);
let result: Result<u32, ResilienceError<MyErr>> = run_with_classifier(
|| {
let n = attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
async move {
if n < 2 {
Err(MyErr(true))
} else {
Ok(42u32)
}
}
},
Profile::Default,
&breaker,
"test-classifier",
|e: &MyErr| e.0,
)
.await;
assert_eq!(result.unwrap(), 42);
assert_eq!(attempts.load(std::sync::atomic::Ordering::SeqCst), 3);
}
#[tokio::test]
async fn run_with_classifier_fails_fast_on_non_retryable() {
#[derive(Debug)]
struct MyErr(bool);
let breaker = CircuitBreaker::new("test-fail-fast", Profile::Fast);
let attempts = std::sync::atomic::AtomicU32::new(0);
let result: Result<u32, ResilienceError<MyErr>> = run_with_classifier(
|| {
attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
async move { Err(MyErr(false)) }
},
Profile::Default,
&breaker,
"test-fail-fast",
|e: &MyErr| e.0,
)
.await;
assert!(matches!(
result,
Err(ResilienceError::Exhausted(MyErr(false)))
));
assert_eq!(attempts.load(std::sync::atomic::Ordering::SeqCst), 1);
}
#[test]
fn breaker_without_metrics_records_nothing_and_still_opens() {
let b = CircuitBreaker::new("quiet", Profile::Default);
(0..5).for_each(|_| b.record(&Err::<(), _>(make_fake_reqwest_error())));
assert!(matches!(
b.guard(),
Err(ResilienceError::CircuitOpen { .. })
));
}
#[cfg(feature = "server")]
#[tokio::test]
async fn attached_metrics_see_calls_retries_and_transitions() {
#[derive(Debug)]
struct MyErr;
let (m, exporter) = crate::telemetry::test_metrics();
let breaker = CircuitBreaker::new("metered", Profile::Default);
breaker.attach_metrics(m);
let attempts = AtomicU32::new(0);
let result: Result<u32, ResilienceError<MyErr>> = run_with_classifier(
|| {
let n = attempts.fetch_add(1, Ordering::SeqCst);
async move {
match n {
0 => Err(MyErr),
_ => Ok(7u32),
}
}
},
Profile::Default,
&breaker,
"metered",
|_: &MyErr| true,
)
.await;
assert_eq!(result.unwrap(), 7);
(0..5).for_each(|_| breaker.record(&Err::<(), _>(make_fake_reqwest_error())));
let text = crate::telemetry::scrape(&exporter);
for line in [
r#"synapse_resilience_retry_attempts_total{label="metered"} 1"#,
r#"synapse_resilience_calls_total{label="metered",outcome="success"} 1"#,
r#"synapse_resilience_breaker_transitions_total{name="metered",transition="open"} 1"#,
r#"synapse_resilience_breaker_state{name="metered"} 1"#,
] {
assert!(
text.lines().any(|l| l == line),
"missing `{line}` in:\n{text}"
);
}
}
}