forked from jmug/stoatchat
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 <ljubica.citydesign@gmail.com> * 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 <ljubica.citydesign@gmail.com> Signed-off-by: Infiland <88491175+Infiland@users.noreply.github.com>
This commit is contained in:
@@ -5,7 +5,7 @@ use authifier::AuthifierEvent;
|
|||||||
use fred::{
|
use fred::{
|
||||||
error::RedisErrorKind,
|
error::RedisErrorKind,
|
||||||
interfaces::{ClientLike, EventInterface, PubsubInterface},
|
interfaces::{ClientLike, EventInterface, PubsubInterface},
|
||||||
types::RedisConfig,
|
types::{ReconnectPolicy, RedisConfig},
|
||||||
};
|
};
|
||||||
use futures::{
|
use futures::{
|
||||||
channel::oneshot,
|
channel::oneshot,
|
||||||
@@ -225,9 +225,9 @@ async fn listener(
|
|||||||
.unwrap_or(REDIS_URI.to_string());
|
.unwrap_or(REDIS_URI.to_string());
|
||||||
|
|
||||||
let redis_config = RedisConfig::from_url(&url).unwrap();
|
let redis_config = RedisConfig::from_url(&url).unwrap();
|
||||||
let subscriber = match report_internal_error!(
|
let mut builder = fred::types::Builder::from_config(redis_config);
|
||||||
fred::types::Builder::from_config(redis_config).build_subscriber_client()
|
builder.set_policy(ReconnectPolicy::new_exponential(8, 100, 30_000, 2));
|
||||||
) {
|
let subscriber = match report_internal_error!(builder.build_subscriber_client()) {
|
||||||
Ok(subscriber) => subscriber,
|
Ok(subscriber) => subscriber,
|
||||||
Err(_) => return,
|
Err(_) => return,
|
||||||
};
|
};
|
||||||
@@ -236,16 +236,21 @@ async fn listener(
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Let Fred automatically re-subscribe to tracked channels on reconnect.
|
||||||
|
subscriber.manage_subscriptions();
|
||||||
|
|
||||||
// Handle Redis connection dropping
|
// Handle Redis connection dropping
|
||||||
let (clean_up_s, clean_up_r) = async_channel::bounded(1);
|
let (clean_up_s, clean_up_r) = async_channel::bounded(1);
|
||||||
let clean_up_s = Arc::new(Mutex::new(clean_up_s));
|
let clean_up_s = Arc::new(Mutex::new(clean_up_s));
|
||||||
subscriber.on_error(move |err| {
|
subscriber.on_error(move |err| {
|
||||||
|
warn!("Redis subscriber error: {:?}", err);
|
||||||
if let RedisErrorKind::Canceled = err.kind() {
|
if let RedisErrorKind::Canceled = err.kind() {
|
||||||
let clean_up_s = clean_up_s.clone();
|
let clean_up_s = clean_up_s.clone();
|
||||||
spawn(async move {
|
spawn(async move {
|
||||||
clean_up_s.lock().await.send(()).await.ok();
|
clean_up_s.lock().await.send(()).await.ok();
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
// Transient errors (IO, timeout) are handled by the reconnect policy.
|
||||||
|
|
||||||
Ok(())
|
Ok(())
|
||||||
});
|
});
|
||||||
@@ -522,4 +527,4 @@ async fn worker(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
Reference in New Issue
Block a user