Broadcast subscriber: fix the inverted subscribe retry backoff
ci / fork-checks (pull_request) Successful in 56s
ci / build (pull_request) Successful in 5m14s

The broadcast subscriber waited 1 << retry_count.max(6) seconds between
failed subscribe attempts. max(6) turns the cap into a floor: the first
retry waited 64 s instead of 1 s, and each later one doubled without a
bound (and would overflow the shift after enough failures).

The delay now comes from subscribe_retry_delay(), 1 s, 2 s, 4 s ... capped
at 64 s, and the retry counter saturates. A unit test pins the schedule
and the top of the range.
This commit is contained in:
2026-09-24 07:01:56 -07:00
parent 499e4d7810
commit ca3abf40f0
+24 -3
View File
@@ -26,7 +26,7 @@ pub fn spawn_broadcast_subscriber(inner: Arc<Inner>, mut shutdown_rx: watch::Rec
}; };
tokio::spawn(async move { tokio::spawn(async move {
let mut retry_count = 0; let mut retry_count: u32 = 0;
trc::event!(Cluster(ClusterEvent::SubscriberStart)); trc::event!(Cluster(ClusterEvent::SubscriberStart));
@@ -53,7 +53,7 @@ pub fn spawn_broadcast_subscriber(inner: Arc<Inner>, mut shutdown_rx: watch::Rec
); );
match tokio::time::timeout( match tokio::time::timeout(
Duration::from_secs(1 << retry_count.max(6)), subscribe_retry_delay(retry_count),
shutdown_rx.changed(), shutdown_rx.changed(),
) )
.await .await
@@ -62,7 +62,7 @@ pub fn spawn_broadcast_subscriber(inner: Arc<Inner>, mut shutdown_rx: watch::Rec
break; break;
} }
Err(_) => { Err(_) => {
retry_count += 1; retry_count = retry_count.saturating_add(1);
continue; continue;
} }
} }
@@ -234,6 +234,11 @@ pub fn spawn_broadcast_subscriber(inner: Arc<Inner>, mut shutdown_rx: watch::Rec
}); });
} }
/// Delay before the next subscribe attempt: 1 s, 2 s, 4 s ... capped at 64 s.
fn subscribe_retry_delay(retry_count: u32) -> Duration {
Duration::from_secs(1u64 << retry_count.min(6))
}
fn log_event(event: &BroadcastEvent) -> trc::Value { fn log_event(event: &BroadcastEvent) -> trc::Value {
match event { match event {
BroadcastEvent::PushNotification(notification) => match notification { BroadcastEvent::PushNotification(notification) => match notification {
@@ -296,3 +301,19 @@ fn log_event(event: &BroadcastEvent) -> trc::Value {
BroadcastEvent::QueueRefresh => "QueueRefresh".into(), BroadcastEvent::QueueRefresh => "QueueRefresh".into(),
} }
} }
#[cfg(test)]
mod tests {
use super::subscribe_retry_delay;
use std::time::Duration;
#[test]
fn subscribe_retry_backoff_grows_then_caps() {
let schedule: Vec<u64> = (0..10)
.map(|n| subscribe_retry_delay(n).as_secs())
.collect();
assert_eq!(schedule, vec![1, 2, 4, 8, 16, 32, 64, 64, 64, 64]);
// No shift overflow at the top of the range.
assert_eq!(subscribe_retry_delay(u32::MAX), Duration::from_secs(64));
}
}