From 5dda371c2b75213bb3fa286cc7ba612980379613 Mon Sep 17 00:00:00 2001 From: Anna Khanova <32508607+khanova@users.noreply.github.com> Date: Wed, 24 Apr 2024 15:13:18 +0200 Subject: [PATCH] Fix a bug with retries (#7494) ## Problem ## Summary of changes By default, it's 5s retry. --- proxy/src/bin/proxy.rs | 7 +++++-- proxy/src/cache/endpoints.rs | 12 ++++++++++-- 2 files changed, 15 insertions(+), 4 deletions(-) diff --git a/proxy/src/bin/proxy.rs b/proxy/src/bin/proxy.rs index 7df320fd42..760ccf40d4 100644 --- a/proxy/src/bin/proxy.rs +++ b/proxy/src/bin/proxy.rs @@ -403,7 +403,7 @@ async fn main() -> anyhow::Result<()> { maintenance_tasks.spawn(usage_metrics::task_main(metrics_config)); client_tasks.spawn(usage_metrics::task_backup( &metrics_config.backup_metric_collection_config, - cancellation_token, + cancellation_token.clone(), )); } @@ -423,7 +423,10 @@ async fn main() -> anyhow::Result<()> { let cache = api.caches.endpoints_cache.clone(); let con = regional_redis_client; let span = tracing::info_span!("endpoints_cache"); - maintenance_tasks.spawn(async move { cache.do_read(con).await }.instrument(span)); + maintenance_tasks.spawn( + async move { cache.do_read(con, cancellation_token.clone()).await } + .instrument(span), + ); } } } diff --git a/proxy/src/cache/endpoints.rs b/proxy/src/cache/endpoints.rs index 2aa1986d5e..02511e6ff7 100644 --- a/proxy/src/cache/endpoints.rs +++ b/proxy/src/cache/endpoints.rs @@ -4,6 +4,7 @@ use std::{ atomic::{AtomicBool, Ordering}, Arc, }, + time::Duration, }; use dashmap::DashSet; @@ -13,6 +14,7 @@ use redis::{ }; use serde::Deserialize; use tokio::sync::Mutex; +use tokio_util::sync::CancellationToken; use tracing::info; use crate::{ @@ -111,16 +113,22 @@ impl EndpointsCache { pub async fn do_read( &self, mut con: ConnectionWithCredentialsProvider, + cancellation_token: CancellationToken, ) -> anyhow::Result { let mut last_id = "0-0".to_string(); loop { - self.ready.store(false, Ordering::Release); if let Err(e) = con.connect().await { tracing::error!("error connecting to redis: {:?}", e); - continue; + self.ready.store(false, Ordering::Release); } if let Err(e) = self.read_from_stream(&mut con, &mut last_id).await { tracing::error!("error reading from redis: {:?}", e); + self.ready.store(false, Ordering::Release); + } + if cancellation_token.is_cancelled() { + info!("cancellation token is cancelled, exiting"); + tokio::time::sleep(Duration::from_secs(60 * 60 * 24 * 7)).await; + // 1 week. } tokio::time::sleep(self.config.retry_interval).await; }