mirror of
https://github.com/Such-Software/smirk-backend-core.git
synced 2026-10-17 07:21:32 +02:00
fix(electrum): tighter connect timeout + primary circuit breaker
The BTC/LTC balance path dials a configured primary (our Fulcrum) then randomized public fallbacks. When the primary was unreachable, every request paid the full 10s CONNECT_TIMEOUT before falling back, so a down/overwhelmed primary meant a ~10s user-facing stall on every balance read. - CONNECT_TIMEOUT 10s -> 4s (primary is a warm mesh node; fallbacks answer in ~1s) - Add a per-primary circuit breaker: after 2 consecutive primary failures, skip the primary for a 30s cooldown and go straight to fallbacks, so a down primary costs a stall at most once per cooldown instead of on every request. Success clears it immediately. Only skips the primary when a fallback exists. Protects the small such-nodes-one Fulcrum box: if it is overwhelmed or drops, the backend fails over to public electrum in ~1s instead of stalling users. 22/22 electrum unit tests pass (incl. a new breaker state-machine test).
This commit is contained in:
+116
-19
@@ -25,9 +25,9 @@
|
|||||||
//! `checked_add`. A hostile peer cannot drive a balance to `0` or `u64::MAX`.
|
//! `checked_add`. A hostile peer cannot drive a balance to `0` or `u64::MAX`.
|
||||||
//! * Each JSON-RPC response id is **correlated** with its request.
|
//! * Each JSON-RPC response id is **correlated** with its request.
|
||||||
|
|
||||||
use std::sync::atomic::{AtomicU64, Ordering};
|
use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
use std::time::Duration;
|
use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
||||||
|
|
||||||
use rand::seq::SliceRandom;
|
use rand::seq::SliceRandom;
|
||||||
use rustls_pki_types::ServerName;
|
use rustls_pki_types::ServerName;
|
||||||
@@ -42,8 +42,15 @@ use tracing::{instrument, warn};
|
|||||||
use crate::config::UtxoConfig;
|
use crate::config::UtxoConfig;
|
||||||
use crate::error::AppError;
|
use crate::error::AppError;
|
||||||
|
|
||||||
/// TCP + TLS connect deadline per server.
|
/// TCP + TLS connect deadline per server. Kept tight: the primary is a warm node
|
||||||
const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
|
/// on the private mesh (sub-second when healthy) and public fallbacks answer in
|
||||||
|
/// ~1s, so a longer deadline only turns a dead server into a user-facing stall.
|
||||||
|
const CONNECT_TIMEOUT: Duration = Duration::from_secs(4);
|
||||||
|
/// Consecutive primary failures before the primary is put on ice (circuit
|
||||||
|
/// breaker), so a down primary stops costing a connect timeout on every request.
|
||||||
|
const PRIMARY_FAIL_THRESHOLD: u32 = 2;
|
||||||
|
/// How long to skip a tripped primary before re-probing it once.
|
||||||
|
const PRIMARY_COOLDOWN: Duration = Duration::from_secs(30);
|
||||||
/// Deadline for the write-request / read-response exchange on a connected stream.
|
/// Deadline for the write-request / read-response exchange on a connected stream.
|
||||||
const IO_TIMEOUT: Duration = Duration::from_secs(30);
|
const IO_TIMEOUT: Duration = Duration::from_secs(30);
|
||||||
/// Hard cap on a single Electrum response line. Even a large verbose tx or a
|
/// Hard cap on a single Electrum response line. Even a large verbose tx or a
|
||||||
@@ -51,6 +58,15 @@ const IO_TIMEOUT: Duration = Duration::from_secs(30);
|
|||||||
/// unbounded stream into a clean error instead of unbounded memory growth.
|
/// unbounded stream into a clean error instead of unbounded memory growth.
|
||||||
const MAX_RESPONSE_BYTES: u64 = 16 * 1024 * 1024;
|
const MAX_RESPONSE_BYTES: u64 = 16 * 1024 * 1024;
|
||||||
|
|
||||||
|
/// Unix time in milliseconds. Used only as the primary circuit-breaker clock, so
|
||||||
|
/// a clock glitch at worst re-probes the primary a little early/late.
|
||||||
|
fn now_millis() -> u64 {
|
||||||
|
SystemTime::now()
|
||||||
|
.duration_since(UNIX_EPOCH)
|
||||||
|
.map(|d| d.as_millis() as u64)
|
||||||
|
.unwrap_or(0)
|
||||||
|
}
|
||||||
|
|
||||||
/// The coin + network an [`ElectrumClient`] is bound to. Used to validate that
|
/// The coin + network an [`ElectrumClient`] is bound to. Used to validate that
|
||||||
/// a queried address actually belongs to this chain (reject ltc-on-btc,
|
/// a queried address actually belongs to this chain (reject ltc-on-btc,
|
||||||
/// testnet-on-mainnet) before it is turned into a scripthash.
|
/// testnet-on-mainnet) before it is turned into a scripthash.
|
||||||
@@ -129,6 +145,11 @@ pub struct ElectrumClient {
|
|||||||
tls: TlsConnector,
|
tls: TlsConnector,
|
||||||
/// Monotonic JSON-RPC request id, shared across clones for correlation.
|
/// Monotonic JSON-RPC request id, shared across clones for correlation.
|
||||||
next_id: Arc<AtomicU64>,
|
next_id: Arc<AtomicU64>,
|
||||||
|
/// Primary circuit breaker: consecutive-failure count, and — once tripped —
|
||||||
|
/// the unix-millis instant until which the primary is skipped in favour of
|
||||||
|
/// fallbacks. Shared across clones so the whole process backs off together.
|
||||||
|
primary_fail_count: Arc<AtomicU32>,
|
||||||
|
primary_cooldown_until_ms: Arc<AtomicU64>,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Confirmed/unconfirmed balance in satoshis (integer fields straight off the
|
/// Confirmed/unconfirmed balance in satoshis (integer fields straight off the
|
||||||
@@ -361,6 +382,8 @@ impl ElectrumClient {
|
|||||||
fallbacks,
|
fallbacks,
|
||||||
tls: build_tls_connector(cfg.electrum_strict_tls),
|
tls: build_tls_connector(cfg.electrum_strict_tls),
|
||||||
next_id: Arc::new(AtomicU64::new(1)),
|
next_id: Arc::new(AtomicU64::new(1)),
|
||||||
|
primary_fail_count: Arc::new(AtomicU32::new(0)),
|
||||||
|
primary_cooldown_until_ms: Arc::new(AtomicU64::new(0)),
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -453,19 +476,49 @@ impl ElectrumClient {
|
|||||||
T: for<'de> Deserialize<'de>,
|
T: for<'de> Deserialize<'de>,
|
||||||
{
|
{
|
||||||
let id = self.next_id.fetch_add(1, Ordering::Relaxed);
|
let id = self.next_id.fetch_add(1, Ordering::Relaxed);
|
||||||
let servers = self.servers_in_order();
|
|
||||||
if servers.is_empty() {
|
if self.primary.is_none() && self.fallbacks.is_empty() {
|
||||||
return Err(AppError::NodeError("no Electrum server configured".into()));
|
return Err(AppError::NodeError("no Electrum server configured".into()));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
let now = now_millis();
|
||||||
|
// Skip a known-down primary (circuit breaker) ONLY when there is a
|
||||||
|
// fallback to use instead: a dead primary otherwise burns the full
|
||||||
|
// CONNECT_TIMEOUT on every request. With no fallback we still try the
|
||||||
|
// primary as a last resort rather than fail outright.
|
||||||
|
let use_primary = self
|
||||||
|
.primary
|
||||||
|
.as_ref()
|
||||||
|
.filter(|_| !(self.primary_in_cooldown(now) && !self.fallbacks.is_empty()));
|
||||||
|
|
||||||
let mut last_err = AppError::NodeError("all Electrum servers failed".into());
|
let mut last_err = AppError::NodeError("all Electrum servers failed".into());
|
||||||
for server in servers {
|
|
||||||
|
// Primary first, tracked by the breaker so repeated failures trip it and
|
||||||
|
// subsequent requests go straight to fallbacks instead of stalling.
|
||||||
|
if let Some(p) = use_primary {
|
||||||
|
match self.try_call::<T>(p, method, ¶ms, id).await {
|
||||||
|
Ok(value) => {
|
||||||
|
self.record_primary_success();
|
||||||
|
return Ok(value);
|
||||||
|
}
|
||||||
|
Err(e) => {
|
||||||
|
self.record_primary_failure(now);
|
||||||
|
// host/port are operator config (not attacker body); the
|
||||||
|
// detailed `inner` is logged privately by AppError.
|
||||||
|
warn!(host = %p.host, port = p.port, method, "Electrum primary query failed");
|
||||||
|
last_err = e;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Then fallbacks, in randomized order.
|
||||||
|
let mut fallbacks = self.fallbacks.clone();
|
||||||
|
fallbacks.shuffle(&mut rand::thread_rng());
|
||||||
|
for server in fallbacks {
|
||||||
match self.try_call::<T>(&server, method, ¶ms, id).await {
|
match self.try_call::<T>(&server, method, ¶ms, id).await {
|
||||||
Ok(value) => return Ok(value),
|
Ok(value) => return Ok(value),
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
// host/port are operator config (not attacker body); the
|
warn!(host = %server.host, port = server.port, method, "Electrum fallback query failed");
|
||||||
// detailed `inner` is logged privately by AppError.
|
|
||||||
warn!(host = %server.host, port = server.port, method, "Electrum server query failed");
|
|
||||||
last_err = e;
|
last_err = e;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -473,16 +526,25 @@ impl ElectrumClient {
|
|||||||
Err(last_err)
|
Err(last_err)
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Primary first, then fallbacks in randomized order.
|
/// True while the primary is inside its post-failure cooldown window.
|
||||||
fn servers_in_order(&self) -> Vec<ServerUrl> {
|
fn primary_in_cooldown(&self, now_ms: u64) -> bool {
|
||||||
let mut ordered = Vec::with_capacity(1 + self.fallbacks.len());
|
self.primary_cooldown_until_ms.load(Ordering::Relaxed) > now_ms
|
||||||
if let Some(p) = &self.primary {
|
}
|
||||||
ordered.push(p.clone());
|
|
||||||
|
/// Record a primary failure; trip the breaker once the threshold is reached.
|
||||||
|
fn record_primary_failure(&self, now_ms: u64) {
|
||||||
|
let count = self.primary_fail_count.fetch_add(1, Ordering::Relaxed) + 1;
|
||||||
|
if count >= PRIMARY_FAIL_THRESHOLD {
|
||||||
|
let until = now_ms.saturating_add(PRIMARY_COOLDOWN.as_millis() as u64);
|
||||||
|
self.primary_cooldown_until_ms.store(until, Ordering::Relaxed);
|
||||||
}
|
}
|
||||||
let mut fallbacks = self.fallbacks.clone();
|
}
|
||||||
fallbacks.shuffle(&mut rand::thread_rng());
|
|
||||||
ordered.extend(fallbacks);
|
/// Record a primary success; clear the failure count and any cooldown so the
|
||||||
ordered
|
/// primary is preferred again immediately.
|
||||||
|
fn record_primary_success(&self) {
|
||||||
|
self.primary_fail_count.store(0, Ordering::Relaxed);
|
||||||
|
self.primary_cooldown_until_ms.store(0, Ordering::Relaxed);
|
||||||
}
|
}
|
||||||
|
|
||||||
/// One attempt against one server: connect (TLS or TCP), exchange, and map
|
/// One attempt against one server: connect (TLS or TCP), exchange, and map
|
||||||
@@ -863,6 +925,41 @@ mod tests {
|
|||||||
ElectrumClient::new(UtxoNetwork::LitecoinMainnet, &cfg()).unwrap()
|
ElectrumClient::new(UtxoNetwork::LitecoinMainnet, &cfg()).unwrap()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn primary_circuit_breaker_trips_and_clears() {
|
||||||
|
let cfg = UtxoConfig {
|
||||||
|
network: "mainnet".into(),
|
||||||
|
electrum_primary: Some("tcp://primary.local:50002".into()),
|
||||||
|
electrum_fallbacks: vec!["ssl://fallback.example:50002".into()],
|
||||||
|
electrum_strict_tls: false,
|
||||||
|
};
|
||||||
|
let c = ElectrumClient::new(UtxoNetwork::LitecoinMainnet, &cfg).unwrap();
|
||||||
|
let t = 1_000_000u64;
|
||||||
|
let window = PRIMARY_COOLDOWN.as_millis() as u64;
|
||||||
|
|
||||||
|
assert!(!c.primary_in_cooldown(t), "starts closed");
|
||||||
|
c.record_primary_failure(t);
|
||||||
|
assert!(
|
||||||
|
!c.primary_in_cooldown(t),
|
||||||
|
"one failure is below threshold: still closed"
|
||||||
|
);
|
||||||
|
c.record_primary_failure(t);
|
||||||
|
assert!(
|
||||||
|
c.primary_in_cooldown(t),
|
||||||
|
"second consecutive failure trips the breaker"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
c.primary_in_cooldown(t + window - 1),
|
||||||
|
"still tripped inside the cooldown window"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
!c.primary_in_cooldown(t + window + 1),
|
||||||
|
"reopens after the cooldown elapses"
|
||||||
|
);
|
||||||
|
c.record_primary_success();
|
||||||
|
assert!(!c.primary_in_cooldown(t), "a success clears the breaker");
|
||||||
|
}
|
||||||
|
|
||||||
/// The lenient (default) TLS connector must complete the handshake + an
|
/// The lenient (default) TLS connector must complete the handshake + an
|
||||||
/// electrum request against PUBLIC ssl:// servers regardless of their cert
|
/// electrum request against PUBLIC ssl:// servers regardless of their cert
|
||||||
/// (CA-signed OR self-signed) — which is the whole point of the default: the
|
/// (CA-signed OR self-signed) — which is the whole point of the default: the
|
||||||
|
|||||||
Reference in New Issue
Block a user