jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-30 05:51:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(relay): measure Redis subscription recovery

Browse Source 
Expose active retry attempts, readiness duration, state age, and consecutive failures for each fixed Redis subscription path.

Co-authored-by: Ravneet Arora <rarora@squareup.com>
Signed-off-by: Ravneet Arora <rarora@squareup.com>
This commit is contained in:
Ravneet Arora
2026-09-09 18:29:33 -04:00
parent 3b1f1145da
commit 5a76bf6528
6 changed files with 202 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
crates/buzz-pubsub/Cargo.toml
View File

@@ -25,4 +25,4 @@ futures-util = { workspace = true }
metrics = { workspace = true }
[dev-dependencies]
tokio = { workspace = true }
tokio = { workspace = true, features = ["test-util"] }

3
crates/buzz-pubsub/src/cache_invalidation.rs
View File

@@ -125,9 +125,8 @@ pub async fn run_cache_invalidation_subscriber_until_cancelled(
health: Arc<SubscriptionHealth>,
) {
let mut backoff_secs = BACKOFF_INITIAL_SECS;
health.connecting(SubscriptionPath::Cache);
loop {
health.connecting(SubscriptionPath::Cache);
let attempt = connect_and_subscribe(&redis_url, &broadcast_tx, &health);
let result = tokio::select! {
() = cancel.cancelled() => return,

3
crates/buzz-pubsub/src/conn_control.rs
View File

@@ -115,9 +115,8 @@ pub async fn run_conn_control_subscriber_until_cancelled(
health: Arc<SubscriptionHealth>,
) {
let mut backoff_secs = BACKOFF_INITIAL_SECS;
health.connecting(SubscriptionPath::ConnectionControl);
loop {
health.connecting(SubscriptionPath::ConnectionControl);
let attempt = connect_and_subscribe(&redis_url, &broadcast_tx, &health);
let result = tokio::select! {
() = cancel.cancelled() => return,

184
crates/buzz-pubsub/src/health.rs
View File

@@ -5,6 +5,7 @@ use std::sync::Mutex;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::sync::Notify;
use tokio::time::Instant;
use uuid::Uuid;
/// Number of required cross-pod Redis subscription paths in every relay pod.
@@ -148,6 +149,14 @@ pub struct SubscriptionPathSnapshot {
pub consumer_attached: bool,
/// Unix timestamp of the latest completed end-to-end ready transition.
pub last_ready_timestamp_seconds: Option<u64>,
/// Time spent reaching the latest completed end-to-end ready transition.
pub last_readiness_duration: Option<Duration>,
/// Time spent continuously outside `ready`; zero while ready.
pub not_ready_duration: Duration,
/// Time spent in the current state.
pub state_duration: Duration,
/// Consecutive failed connection/subscription attempts since the latest ready state.
pub consecutive_failures: u64,
}
#[derive(Clone, Copy, Debug)]
@@ -158,25 +167,65 @@ struct PathStatus {
ever_ready: bool,
terminal: bool,
last_ready_timestamp_seconds: Option<u64>,
last_readiness_duration: Option<Duration>,
not_ready_since: Instant,
state_since: Instant,
consecutive_failures: u64,
}
impl PathStatus {
const INITIAL: Self = Self {
state: SubscriptionPathState::Connecting,
consumer_attached: false,
network_ready: false,
ever_ready: false,
terminal: false,
last_ready_timestamp_seconds: None,
};
fn initial(now: Instant) -> Self {
Self {
state: SubscriptionPathState::Connecting,
consumer_attached: false,
network_ready: false,
ever_ready: false,
terminal: false,
last_ready_timestamp_seconds: None,
last_readiness_duration: None,
not_ready_since: now,
state_since: now,
consecutive_failures: 0,
}
}
const fn snapshot(self) -> SubscriptionPathSnapshot {
fn snapshot(self, now: Instant) -> SubscriptionPathSnapshot {
SubscriptionPathSnapshot {
state: self.state,
consumer_attached: self.consumer_attached,
last_ready_timestamp_seconds: self.last_ready_timestamp_seconds,
last_readiness_duration: self.last_readiness_duration,
not_ready_duration: if self.state == SubscriptionPathState::Ready {
Duration::ZERO
} else {
now.saturating_duration_since(self.not_ready_since)
},
state_duration: now.saturating_duration_since(self.state_since),
consecutive_failures: self.consecutive_failures,
}
}
fn set_state(&mut self, state: SubscriptionPathState, now: Instant) {
if self.state != state {
self.state = state;
self.state_since = now;
}
}
fn mark_ready(&mut self, now: Instant) {
self.last_readiness_duration = Some(now.saturating_duration_since(self.not_ready_since));
self.set_state(SubscriptionPathState::Ready, now);
self.ever_ready = true;
self.last_ready_timestamp_seconds = Some(unix_timestamp_seconds());
self.consecutive_failures = 0;
}
fn mark_not_ready(&mut self, state: SubscriptionPathState, now: Instant) {
if self.state == SubscriptionPathState::Ready {
self.not_ready_since = now;
}
self.set_state(state, now);
}
}
/// Shared health snapshot and metric publisher for all Redis subscription paths.
@@ -197,10 +246,11 @@ impl Default for SubscriptionHealth {
impl SubscriptionHealth {
/// Create the per-process fixed path snapshot.
pub fn new() -> Self {
let now = Instant::now();
Self {
group_id: Uuid::new_v4(),
sequence: AtomicU64::new(1),
statuses: Mutex::new([PathStatus::INITIAL; REQUIRED_PATH_COUNT]),
statuses: Mutex::new([PathStatus::initial(now); REQUIRED_PATH_COUNT]),
publish_lock: Mutex::new(()),
changed: Notify::new(),
}
@@ -223,11 +273,9 @@ impl SubscriptionHealth {
} else {
SubscriptionTransition::Connected
});
status.state = SubscriptionPathState::Ready;
status.ever_ready = true;
status.last_ready_timestamp_seconds = Some(unix_timestamp_seconds());
status.mark_ready(Instant::now());
}
status.snapshot()
status.snapshot(Instant::now())
};
self.publish(
path,
@@ -237,13 +285,19 @@ impl SubscriptionHealth {
);
}
/// Record that the initial Redis connection attempt has started.
/// Record that a Redis connection/subscription attempt has started.
pub fn connecting(&self, path: SubscriptionPath) {
self.update_nonterminal(path, |status| {
if self.update_nonterminal(path, |status| {
status.network_ready = false;
status.state = SubscriptionPathState::Connecting;
status.mark_not_ready(SubscriptionPathState::Connecting, Instant::now());
(None, None)
});
}) {
metrics::counter!(
"buzz_redis_subscription_attempts_total",
"path" => path.as_str(),
)
.increment(1);
}
}
/// Record a completed Redis subscription handshake.
@@ -258,9 +312,7 @@ impl SubscriptionHealth {
} else {
SubscriptionTransition::Connected
};
status.state = SubscriptionPathState::Ready;
status.ever_ready = true;
status.last_ready_timestamp_seconds = Some(unix_timestamp_seconds());
status.mark_ready(Instant::now());
(
Some(transition),
Some(SubscriptionTransitionReason::Subscribe),
@@ -272,7 +324,8 @@ impl SubscriptionHealth {
pub fn reconnecting(&self, path: SubscriptionPath, reason: SubscriptionTransitionReason) {
self.update_nonterminal(path, |status| {
status.network_ready = false;
status.state = SubscriptionPathState::Reconnecting;
status.mark_not_ready(SubscriptionPathState::Reconnecting, Instant::now());
status.consecutive_failures = status.consecutive_failures.saturating_add(1);
let transition = if status.ever_ready {
SubscriptionTransition::Disconnected
} else {
@@ -286,7 +339,7 @@ impl SubscriptionHealth {
pub fn failed(&self, path: SubscriptionPath, reason: SubscriptionTransitionReason) {
self.update(path, |status| {
status.network_ready = false;
status.state = SubscriptionPathState::Failed;
status.mark_not_ready(SubscriptionPathState::Failed, Instant::now());
status.terminal = true;
(Some(SubscriptionTransition::Terminal), Some(reason))
});
@@ -297,7 +350,7 @@ impl SubscriptionHealth {
self.update_nonterminal(path, |status| {
status.network_ready = false;
status.consumer_attached = false;
status.state = SubscriptionPathState::Stopped;
status.mark_not_ready(SubscriptionPathState::Stopped, Instant::now());
status.terminal = true;
(
Some(SubscriptionTransition::Terminal),
@@ -308,16 +361,17 @@ impl SubscriptionHealth {
/// Return the current state for one path.
pub fn snapshot(&self, path: SubscriptionPath) -> SubscriptionPathSnapshot {
self.lock_statuses()[path.index()].snapshot()
self.lock_statuses()[path.index()].snapshot(Instant::now())
}
/// Return all path snapshots in [`SubscriptionPath::ALL`] order.
pub fn snapshots(&self) -> [SubscriptionPathSnapshot; REQUIRED_PATH_COUNT] {
let statuses = self.lock_statuses();
let now = Instant::now();
[
statuses[0].snapshot(),
statuses[1].snapshot(),
statuses[2].snapshot(),
statuses[0].snapshot(now),
statuses[1].snapshot(now),
statuses[2].snapshot(now),
]
}
@@ -373,8 +427,8 @@ impl SubscriptionHealth {
Option<SubscriptionTransition>,
Option<SubscriptionTransitionReason>,
),
) {
self.update(path, mutate);
) -> bool {
self.update(path, mutate)
}
fn update(
@@ -386,18 +440,19 @@ impl SubscriptionHealth {
Option<SubscriptionTransition>,
Option<SubscriptionTransitionReason>,
),
) {
) -> bool {
let _publish_guard = self.lock_publish();
let (snapshot, transition, reason) = {
let mut statuses = self.lock_statuses();
let status = &mut statuses[path.index()];
if status.terminal {
return;
return false;
}
let (transition, reason) = mutate(status);
(status.snapshot(), transition, reason)
(status.snapshot(Instant::now()), transition, reason)
};
self.publish(path, snapshot, transition, reason);
true
}
fn publish(
@@ -464,6 +519,28 @@ fn publish_path_gauges(path: SubscriptionPath, snapshot: SubscriptionPathSnapsho
)
.set(timestamp as f64);
}
if let Some(duration) = snapshot.last_readiness_duration {
metrics::gauge!(
"buzz_redis_subscription_last_readiness_duration_seconds",
"path" => path.as_str(),
)
.set(duration.as_secs_f64());
}
metrics::gauge!(
"buzz_redis_subscription_not_ready_duration_seconds",
"path" => path.as_str(),
)
.set(snapshot.not_ready_duration.as_secs_f64());
metrics::gauge!(
"buzz_redis_subscription_state_duration_seconds",
"path" => path.as_str(),
)
.set(snapshot.state_duration.as_secs_f64());
metrics::gauge!(
"buzz_redis_subscription_consecutive_failures",
"path" => path.as_str(),
)
.set(snapshot.consecutive_failures as f64);
}
fn publish_aggregate_gauges(snapshots: &[SubscriptionPathSnapshot; REQUIRED_PATH_COUNT]) {
@@ -533,6 +610,47 @@ mod tests {
);
}
#[tokio::test(start_paused = true)]
async fn readiness_duration_includes_retries_and_resets_after_recovery() {
let health = SubscriptionHealth::new();
health.consumer_attached(SubscriptionPath::Event);
health.connecting(SubscriptionPath::Event);
tokio::time::advance(Duration::from_secs(2)).await;
health.reconnecting(SubscriptionPath::Event, SubscriptionTransitionReason::Dial);
tokio::time::advance(Duration::from_secs(3)).await;
health.connecting(SubscriptionPath::Event);
tokio::time::advance(Duration::from_secs(4)).await;
health.network_ready(SubscriptionPath::Event);
let initial = health.snapshot(SubscriptionPath::Event);
assert_eq!(
initial.last_readiness_duration,
Some(Duration::from_secs(9))
);
assert_eq!(initial.not_ready_duration, Duration::ZERO);
assert_eq!(initial.consecutive_failures, 0);
health.reconnecting(
SubscriptionPath::Event,
SubscriptionTransitionReason::StreamClosed,
);
tokio::time::advance(Duration::from_secs(5)).await;
let recovering = health.snapshot(SubscriptionPath::Event);
assert_eq!(recovering.not_ready_duration, Duration::from_secs(5));
assert_eq!(recovering.state_duration, Duration::from_secs(5));
assert_eq!(recovering.consecutive_failures, 1);
health.connecting(SubscriptionPath::Event);
tokio::time::advance(Duration::from_secs(2)).await;
health.network_ready(SubscriptionPath::Event);
let recovered = health.snapshot(SubscriptionPath::Event);
assert_eq!(
recovered.last_readiness_duration,
Some(Duration::from_secs(7))
);
assert_eq!(recovered.consecutive_failures, 0);
}
#[tokio::test]
async fn wait_for_state_observes_transition_without_sleep() {
let health = Arc::new(SubscriptionHealth::new());

3
crates/buzz-pubsub/src/subscriber.rs
View File

@@ -45,9 +45,8 @@ pub(crate) async fn run_subscriber(
health: Arc<SubscriptionHealth>,
) {
let mut backoff_secs = BACKOFF_INITIAL_SECS;
health.connecting(SubscriptionPath::Event);
loop {
health.connecting(SubscriptionPath::Event);
let attempt = connect_and_subscribe(
&redis_url,
&broadcast_tx,

47
crates/buzz-relay/src/metrics.rs
View File

@@ -269,6 +269,29 @@ pub(crate) fn describe_redis_subscription_metrics() {
metrics::Unit::Seconds,
"Unix timestamp of the latest completed consumer-plus-Redis subscription handshake"
);
metrics::describe_gauge!(
"buzz_redis_subscription_last_readiness_duration_seconds",
metrics::Unit::Seconds,
"Seconds spent continuously not ready before the latest completed end-to-end subscription handshake"
);
metrics::describe_gauge!(
"buzz_redis_subscription_not_ready_duration_seconds",
metrics::Unit::Seconds,
"Seconds spent continuously outside ready, or zero while the subscription path is ready"
);
metrics::describe_gauge!(
"buzz_redis_subscription_state_duration_seconds",
metrics::Unit::Seconds,
"Seconds spent in the current bounded subscription path state"
);
metrics::describe_gauge!(
"buzz_redis_subscription_consecutive_failures",
"Consecutive connection or subscription failures since the path was last ready"
);
metrics::describe_counter!(
"buzz_redis_subscription_attempts_total",
"Redis connection and subscription attempts by required path"
);
metrics::describe_counter!(
"buzz_redis_subscription_transitions_total",
"Redis subscription path transitions by path, bounded transition, and bounded reason"
@@ -380,6 +403,7 @@ mod contract_tests {
let health = SubscriptionHealth::new();
for path in SubscriptionPath::ALL {
health.consumer_attached(path);
health.connecting(path);
health.network_ready(path);
}
health.refresh_metrics();
@@ -390,6 +414,13 @@ mod contract_tests {
assert!(
scrape.contains("# TYPE buzz_redis_subscription_last_ready_timestamp_seconds gauge")
);
assert!(
scrape.contains("# TYPE buzz_redis_subscription_last_readiness_duration_seconds gauge")
);
assert!(scrape.contains("# TYPE buzz_redis_subscription_not_ready_duration_seconds gauge"));
assert!(scrape.contains("# TYPE buzz_redis_subscription_state_duration_seconds gauge"));
assert!(scrape.contains("# TYPE buzz_redis_subscription_consecutive_failures gauge"));
assert!(scrape.contains("# TYPE buzz_redis_subscription_attempts_total counter"));
assert!(scrape.contains("# TYPE buzz_redis_subscription_transitions_total counter"));
assert!(scrape.contains("# TYPE buzz_redis_subscription_paths_ready gauge"));
assert!(scrape.contains("# TYPE buzz_redis_subscription_all_ready gauge"));
@@ -416,6 +447,22 @@ mod contract_tests {
}) {
assert_eq!(label_keys(line), BTreeSet::from(["path"]));
}
for prefix in [
"buzz_redis_subscription_last_readiness_duration_seconds{",
"buzz_redis_subscription_not_ready_duration_seconds{",
"buzz_redis_subscription_state_duration_seconds{",
"buzz_redis_subscription_consecutive_failures{",
"buzz_redis_subscription_attempts_total{",
] {
let lines = scrape
.lines()
.filter(|line| line.starts_with(prefix))
.collect::<Vec<_>>();
assert_eq!(lines.len(), 3, "unexpected series for {prefix}:\n{scrape}");
for line in lines {
assert_eq!(label_keys(line), BTreeSet::from(["path"]));
}
}
for line in scrape
.lines()
.filter(|line| line.starts_with("buzz_redis_subscription_transitions_total{"))