fix: add timeout to health checks

gateway/src/project.rs

@@ -15,7 +15,7 @@ use hyper::Client;
 use once_cell::sync::Lazy;
 use rand::distributions::{Alphanumeric, DistString};
 use serde::{Deserialize, Serialize};
-use tokio::time;
+use tokio::time::{self, timeout};
 use tracing::{debug, error};
 
 use crate::{
@@ -63,6 +63,7 @@ const MAX_RESTARTS: i64 = 3;
 
 // Client used for health checks
 static CLIENT: Lazy<Client<HttpConnector>> = Lazy::new(Client::new);
+static IS_HEALTHY_TIMEOUT: u64 = 10;  // Health check must succeed within 10 seconds
 
 #[async_trait]
 impl<Ctx> Refresh<Ctx> for ContainerInspectResponse
@@ -640,7 +641,8 @@ impl Service {
 
     pub async fn is_healthy(&mut self) -> bool {
         let uri = self.uri(format!("/projects/{}/status", self.name)).unwrap();
-        let is_healthy = matches!(CLIENT.get(uri).await, Ok(res) if res.status().is_success());
+        let resp = timeout(Duration::from_secs(IS_HEALTHY_TIMEOUT), CLIENT.get(uri)).await;
+        let is_healthy = matches!(resp, Ok(Ok(res)) if res.status().is_success());
         self.last_check = Some(HealthCheckRecord::new(is_healthy));
         is_healthy
     }

gateway/src/worker.rs

@@ -1,9 +1,13 @@
+use std::time::Duration;
+
 use tokio::sync::mpsc::{channel, Receiver, Sender};
-use tracing::{debug, info};
+use tokio::time;
+use tracing::{debug, info, warn};
 
 use crate::task::{BoxedTask, TaskResult};
 use crate::Error;
 
+const TASK_MAX_IDLE_TIMEOUT: u64 = 60;  // Maximum time before a task is considered degraded
 const WORKER_QUEUE_SIZE: usize = 2048;
 
 pub struct Worker<W = BoxedTask> {
@@ -57,7 +61,16 @@ impl Worker<BoxedTask> {
 
         while let Some(mut work) = self.recv.recv().await {
             loop {
-                match work.poll(()).await {
+                let timeout = time::sleep(Duration::from_secs(TASK_MAX_IDLE_TIMEOUT));
+                let mut poll = work.poll(());
+                let res = tokio::select! {
+                    res = &mut poll => res,
+                    _ = timeout => {
+                        warn!("a task has been running for a long time");
+                        poll.await
+                    }
+                };
+                match res {
                     TaskResult::Done(_) | TaskResult::Cancelled => break,
                     TaskResult::Pending(_) | TaskResult::TryAgain => continue,
                     TaskResult::Err(err) => {