From 057f2bb8b359f8b942741a30ff54eeb8fbe3e0b1 Mon Sep 17 00:00:00 2001 From: Infiland <88491175+Infiland@users.noreply.github.com> Date: Sat, 18 Apr 2026 05:15:54 +0200 Subject: [PATCH] fix: add reconnection policy to Redis subscriber to prevent ghost state (#708) * fix: add reconnection policy to Redis subscriber to prevent ghost state - Add ReconnectPolicy::new_exponential(0, 100, 30_000, 2) to the subscriber builder, unlimited retries with exponential backoff (100ms min, 30s max) - Add on_reconnect handler that signals the listener loop to force a subscription reset, re-subscribing to all topics on the new connection - Add warn-level logging to on_error for all Redis subscriber errors (previously only Canceled was handled, others were silently ignored) Signed-off-by: Infiland * Update websocket.rs Signed-off-by: Infiland <88491175+Infiland@users.noreply.github.com> * Auto-manage subscriptions on reconnect Call subscriber.manage_subscriptions() so the subscriber will automatically re-subscribe tracked channels after a Redis reconnect. Remove the manual reconnect channel and on_reconnect handler along with the select branch that forced SubscriptionStateChange::Reset. Signed-off-by: Infiland <88491175+Infiland@users.noreply.github.com> --------- Signed-off-by: Infiland Signed-off-by: Infiland <88491175+Infiland@users.noreply.github.com> --- crates/bonfire/src/websocket.rs | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/crates/bonfire/src/websocket.rs b/crates/bonfire/src/websocket.rs index ec4202dc..b5611bd0 100644 --- a/crates/bonfire/src/websocket.rs +++ b/crates/bonfire/src/websocket.rs @@ -5,7 +5,7 @@ use authifier::AuthifierEvent; use fred::{ error::RedisErrorKind, interfaces::{ClientLike, EventInterface, PubsubInterface}, - types::RedisConfig, + types::{ReconnectPolicy, RedisConfig}, }; use futures::{ channel::oneshot, @@ -225,9 +225,9 @@ async fn listener( .unwrap_or(REDIS_URI.to_string()); let redis_config = RedisConfig::from_url(&url).unwrap(); - let subscriber = match report_internal_error!( - fred::types::Builder::from_config(redis_config).build_subscriber_client() - ) { + let mut builder = fred::types::Builder::from_config(redis_config); + builder.set_policy(ReconnectPolicy::new_exponential(8, 100, 30_000, 2)); + let subscriber = match report_internal_error!(builder.build_subscriber_client()) { Ok(subscriber) => subscriber, Err(_) => return, }; @@ -236,16 +236,21 @@ async fn listener( return; } + // Let Fred automatically re-subscribe to tracked channels on reconnect. + subscriber.manage_subscriptions(); + // Handle Redis connection dropping let (clean_up_s, clean_up_r) = async_channel::bounded(1); let clean_up_s = Arc::new(Mutex::new(clean_up_s)); subscriber.on_error(move |err| { + warn!("Redis subscriber error: {:?}", err); if let RedisErrorKind::Canceled = err.kind() { let clean_up_s = clean_up_s.clone(); spawn(async move { clean_up_s.lock().await.send(()).await.ok(); }); } + // Transient errors (IO, timeout) are handled by the reconnect policy. Ok(()) }); @@ -522,4 +527,4 @@ async fn worker( } } } -} +} \ No newline at end of file