fix: pod dying due to non-blocking tasks (#226)

picahq · Jan 27, 2025 · 1c8dd82 · 1c8dd82
1 parent 3b6ecd2
commit 1c8dd82
Show file tree

Hide file tree

Showing 3 changed files with 22 additions and 17 deletions.
diff --git a/watchdog/src/client.rs b/watchdog/src/client.rs
@@ -1,10 +1,10 @@
 use crate::config::WatchdogConfig;
 use cache::remote::RedisCache;
+use chrono::Utc;
 use entities::{cache::CacheConfig, database::DatabaseConfig, InternalError, PicaError};
 use redis::{AsyncCommands, RedisResult};
 use std::fmt::Display;
 use std::time::Duration;
-use tokio::task::JoinHandle;
 use tracing::{error, info};
 
 pub struct WatchdogClient {
@@ -35,8 +35,8 @@ impl WatchdogClient {
         }
     }
 
-    pub fn start(self) -> JoinHandle<Result<(), PicaError>> {
-        tokio::spawn(self.run())
+    pub async fn start(self) -> Result<(), PicaError> {
+        self.run().await
     }
 
     pub async fn run(self) -> Result<(), PicaError> {
@@ -59,13 +59,16 @@ impl WatchdogClient {
 
         let key = self.cache.api_throughput_key.clone();
         let mut redis_clone = cache.inner.clone();
-        tokio::spawn(async move {
-            loop {
-                let _: RedisResult<String> = async { redis_clone.del(key.clone()).await }.await;
-                tokio::time::sleep(Duration::from_secs(60)).await;
-            }
-        });
 
-        Ok(())
+        tracing::info!("Rate limiter enabled. Connecting to initialized cache");
+
+        loop {
+            let _: RedisResult<String> = async { redis_clone.del(key.clone()).await }.await;
+            tracing::info!("Rate limiter cleared for {key} at {}", Utc::now());
+            tokio::time::sleep(Duration::from_secs(
+                self.watchdog.rate_limiter_refresh_interval,
+            ))
+            .await;
+        }
     }
 }
diff --git a/watchdog/src/config.rs b/watchdog/src/config.rs
@@ -4,10 +4,8 @@ use std::fmt::{Display, Formatter};
 
 #[derive(Envconfig, Clone)] // Intentionally no Debug so secret is not printed
 pub struct WatchdogConfig {
-    #[envconfig(from = "EVENT_TIMEOUT", default = "300")] // 300 seconds/ 5 minutes
-    pub event_timeout: u64,
-    #[envconfig(from = "POLL_DURATION", default = "10")] // 10 seconds
-    pub poll_duration: u64,
+    #[envconfig(from = "RATE_LIMITER_REFRESH_INTERVAL", default = "60")]
+    pub rate_limiter_refresh_interval: u64,
     #[envconfig(nested = true)]
     pub redis: CacheConfig,
     #[envconfig(nested = true)]
@@ -16,8 +14,11 @@ pub struct WatchdogConfig {
 
 impl Display for WatchdogConfig {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        writeln!(f, "POLL_DURATION: {}", self.poll_duration)?;
-        writeln!(f, "EVENT_TIMEOUT: {}", self.event_timeout)?;
+        writeln!(
+            f,
+            "RATE_LIMITER_REFRESH_INTERVAL: {}",
+            self.rate_limiter_refresh_interval
+        )?;
         writeln!(f, "{}", self.redis)?;
         writeln!(f, "{}", self.db)
     }

diff --git a/watchdog/src/main.rs b/watchdog/src/main.rs
@@ -28,6 +28,7 @@ async fn main() -> Result<()> {
 
     let client = WatchdogClient::new(watchdog_config, cache_config, database_config);
 
-    client.start().await??;
+    client.start().await?;
+
     Ok(())
 }