fix(targets): probe webhook health by host port (#2854)

crates/e2e_test/src/common.rs

@@ -351,6 +351,7 @@ impl RustFSTestEnvironment {
     async fn start_rustfs_server_inner(
         &mut self,
         extra_args: Vec<&str>,
+        extra_env: &[(&str, &str)],
         cleanup_existing: bool,
     ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         if cleanup_existing {
@@ -362,10 +363,12 @@ impl RustFSTestEnvironment {
         info!("Starting RustFS server with args: {:?}", args);
 
         let binary_path = rustfs_binary_path();
-        let process = Command::new(&binary_path)
+        let mut command = Command::new(&binary_path);
-            .env("RUST_LOG", "rustfs=info,rustfs_notify=debug")
+        command.env("RUST_LOG", "rustfs=info,rustfs_notify=debug");
-            .args(&args)
+        for (key, value) in extra_env {
-            .spawn()?;
+            command.env(key, value);
+        }
+        let process = command.args(&args).spawn()?;
 
         self.process = Some(process);
 
@@ -377,7 +380,16 @@ impl RustFSTestEnvironment {
 
     /// Start RustFS server with basic configuration
     pub async fn start_rustfs_server(&mut self, extra_args: Vec<&str>) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-        self.start_rustfs_server_inner(extra_args, true).await
+        self.start_rustfs_server_inner(extra_args, &[], true).await
+    }
+
+    /// Start RustFS server with extra child-process environment variables.
+    pub async fn start_rustfs_server_with_env(
+        &mut self,
+        extra_args: Vec<&str>,
+        extra_env: &[(&str, &str)],
+    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        self.start_rustfs_server_inner(extra_args, extra_env, true).await
     }
 
     /// Start RustFS server without cleaning up other running RustFS processes.
@@ -388,7 +400,7 @@ impl RustFSTestEnvironment {
         &mut self,
         extra_args: Vec<&str>,
     ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-        self.start_rustfs_server_inner(extra_args, false).await
+        self.start_rustfs_server_inner(extra_args, &[], false).await
     }
 
     /// Wait for RustFS server to be ready.

crates/e2e_test/src/object_lambda_test.rs

@@ -25,7 +25,7 @@ use std::error::Error;
 use time::OffsetDateTime;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::TcpListener;
-use tokio::sync::oneshot;
+use tokio::sync::{mpsc, oneshot};
 use tokio::time::{Duration, timeout};
 
 #[derive(Debug)]
@@ -172,6 +172,28 @@ async fn spawn_object_lambda_webhook_server_with_response(
     Ok((webhook_url, request_rx, handle))
 }
 
+async fn read_request_path(stream: &mut tokio::net::TcpStream) -> Result<String, Box<dyn Error + Send + Sync>> {
+    let mut buffer = Vec::new();
+    let mut chunk = [0_u8; 4096];
+
+    let header_end = loop {
+        let read = stream.read(&mut chunk).await?;
+        if read == 0 {
+            return Err("request ended before headers were fully received".into());
+        }
+        buffer.extend_from_slice(&chunk[..read]);
+        if let Some(pos) = find_header_terminator(&buffer) {
+            break pos;
+        }
+    };
+
+    let header_text = std::str::from_utf8(&buffer[..header_end])?;
+    let request_line = header_text.lines().next().ok_or("missing request line")?;
+    let path = request_line.split_whitespace().nth(1).ok_or("missing request path")?;
+
+    Ok(path.to_string())
+}
+
 async fn presigned_get_request(
     url: &str,
     access_key: &str,
@@ -326,11 +348,15 @@ async fn delete_webhook_target(env: &RustFSTestEnvironment, target_name: &str) -
 }
 
 fn notification_target_is_listed(targets: &serde_json::Value, target_name: &str) -> bool {
+    notification_target_entry(targets, target_name).is_some()
+}
+
+fn notification_target_entry<'a>(targets: &'a serde_json::Value, target_name: &str) -> Option<&'a serde_json::Value> {
     targets["notification_endpoints"]
         .as_array()
         .into_iter()
         .flatten()
-        .any(|entry| {
+        .find(|entry| {
             entry["account_id"].as_str() == Some(target_name)
                 && entry["service"]
                     .as_str()
@@ -338,6 +364,10 @@ fn notification_target_is_listed(targets: &serde_json::Value, target_name: &str)
         })
 }
 
+fn notification_target_status<'a>(targets: &'a serde_json::Value, target_name: &str) -> Option<&'a str> {
+    notification_target_entry(targets, target_name).and_then(|entry| entry["status"].as_str())
+}
+
 async fn wait_for_target_visibility(
     env: &RustFSTestEnvironment,
     target_name: &str,
@@ -387,6 +417,34 @@ async fn restart_rustfs_server(env: &mut RustFSTestEnvironment) -> Result<(), Bo
     env.start_rustfs_server_without_cleanup(vec![]).await
 }
 
+async fn spawn_http_origin_probe_server() -> Result<
+    (
+        String,
+        mpsc::Receiver<String>,
+        tokio::task::JoinHandle<Result<(), Box<dyn Error + Send + Sync>>>,
+    ),
+    Box<dyn Error + Send + Sync>,
+> {
+    let listener = TcpListener::bind("127.0.0.1:0").await?;
+    let address = listener.local_addr()?;
+    let webhook_url = format!("http://{address}/hook");
+    let (path_tx, path_rx) = mpsc::channel(1);
+
+    let handle = tokio::spawn(async move {
+        loop {
+            let (mut stream, _) = listener.accept().await?;
+            let path = timeout(Duration::from_secs(2), read_request_path(&mut stream)).await??;
+            let _ = path_tx.try_send(path.clone());
+            if path == "/" {
+                let response = b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n";
+                stream.write_all(response).await?;
+            }
+        }
+    });
+
+    Ok((webhook_url, path_rx, handle))
+}
+
 async fn read_persisted_server_config(env: &RustFSTestEnvironment) -> String {
     let path = format!("{}/.rustfs.sys/config/config.json", env.temp_dir);
     match tokio::fs::read_to_string(&path).await {
@@ -515,6 +573,40 @@ async fn test_notification_target_persists_across_restart_and_delete() -> Result
     Ok(())
 }
 
+#[tokio::test]
+#[serial]
+async fn test_notification_target_with_path_is_online_via_transport_probe() -> Result<(), Box<dyn Error + Send + Sync>> {
+    init_logging();
+
+    let (webhook_url, mut probe_rx, probe_handle) = spawn_http_origin_probe_server().await?;
+
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server_with_env(vec![], &[("RUSTFS_NOTIFY_ENABLE", "true")])
+        .await?;
+
+    let target_name = "path-probe";
+    configure_webhook_target(&env, target_name, &webhook_url, "secret-token").await?;
+
+    let (visible_targets, visible_arns) = wait_for_target_visibility(&env, target_name).await?;
+    assert_eq!(notification_target_status(&visible_targets, target_name), Some("online"));
+    let observed_path = timeout(Duration::from_secs(10), probe_rx.recv())
+        .await
+        .map_err(|_| "probe server timed out waiting for a request")?
+        .ok_or("probe server did not observe a request")?;
+    assert_eq!(observed_path, "/");
+    assert!(
+        visible_arns
+            .iter()
+            .any(|arn| arn.ends_with(&format!(":{target_name}:webhook"))),
+        "target ARN missing for reachable path endpoint: {visible_arns:?}"
+    );
+
+    probe_handle.abort();
+    let _ = probe_handle.await;
+
+    Ok(())
+}
+
 #[tokio::test]
 #[serial]
 async fn test_get_object_lambda_accepts_presigned_requests() -> Result<(), Box<dyn Error + Send + Sync>> {

crates/targets/src/target/webhook.rs

@@ -106,6 +106,7 @@ where
 {
     id: TargetID,
     args: WebhookArgs,
+    health_check_url: Option<Url>,
     http_client: Arc<Client>,
     // Add Send + Sync constraints to ensure thread safety
     store: Option<Box<dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync>>,
@@ -124,6 +125,7 @@ where
         Box::new(WebhookTarget::<E> {
             id: self.id.clone(),
             args: self.args.clone(),
+            health_check_url: self.health_check_url.clone(),
             http_client: Arc::clone(&self.http_client),
             store: self.store.as_ref().map(|s| s.boxed_clone()),
             initialized: AtomicBool::new(self.initialized.load(Ordering::SeqCst)),
@@ -140,6 +142,11 @@ where
         args.validate()?;
         // Create a TargetID
         let target_id = TargetID::new(id, ChannelTargetType::Webhook.as_str().to_string());
+        let health_check_url = if args.enable {
+            Some(Self::health_check_url(&args.endpoint)?)
+        } else {
+            None
+        };
 
         // Build HTTP client using the helper function
         let http_client = Arc::new(Self::build_http_client(&args)?);
@@ -173,6 +180,7 @@ where
         Ok(WebhookTarget::<E> {
             id: target_id,
             args,
+            health_check_url,
             http_client,
             store: queue_store,
             initialized: AtomicBool::new(false),
@@ -222,38 +230,69 @@ where
             .map_err(|e| TargetError::Configuration(format!("Failed to build HTTP client: {e}")))
     }
 
+    fn health_check_url(endpoint: &Url) -> Result<Url, TargetError> {
+        endpoint
+            .host()
+            .ok_or_else(|| TargetError::Configuration(format!("Webhook endpoint '{}' is missing a host", endpoint)))?;
+        let mut health_check_url = endpoint.clone();
+        health_check_url.set_path("/");
+        health_check_url.set_query(None);
+        health_check_url.set_fragment(None);
+
+        Ok(health_check_url)
+    }
+
+    async fn probe_reachability(&self) -> Result<bool, TargetError> {
+        let Some(health_check_url) = self.health_check_url.as_ref() else {
+            return Ok(false);
+        };
+
+        match tokio::time::timeout(Duration::from_secs(5), self.http_client.head(health_check_url.as_str()).send()).await {
+            Ok(Ok(resp)) => {
+                debug!(
+                    target = %self.id,
+                    status = %resp.status(),
+                    health_check_url = %health_check_url,
+                    "Webhook health check request succeeded"
+                );
+                Ok(true)
+            }
+            Ok(Err(err)) if err.is_timeout() => Err(TargetError::Timeout(format!(
+                "Webhook health check request to {} timed out",
+                health_check_url
+            ))),
+            Ok(Err(err)) if err.is_connect() => Ok(false),
+            Ok(Err(err)) => Err(TargetError::Network(format!(
+                "Webhook health check request to {} failed: {}",
+                health_check_url, err
+            ))),
+            Err(_) => Err(TargetError::Timeout(format!(
+                "Webhook health check request to {} timed out",
+                health_check_url
+            ))),
+        }
+    }
+
     async fn init_inner(&self) -> Result<(), TargetError> {
         if self.initialized.load(Ordering::SeqCst) {
             return Ok(());
         }
 
-        // HTTP HEAD probe: verifies the full request path (proxy, TLS, firewall)
+        if !self.args.enable {
-        // unlike TCP connect which can't detect proxy issues.
+            return Ok(());
-        let probe_timeout = Duration::from_secs(5);
+        }
-        match tokio::time::timeout(probe_timeout, self.http_client.head(self.args.endpoint.as_str()).send()).await {
+
-            Ok(Ok(resp)) => {
+        // Use the configured reqwest client against the origin URL so proxy and TLS
-                let status = resp.status();
+        // behavior matches real delivery while avoiding path-specific false negatives.
-                if status.is_success() || status == StatusCode::NOT_FOUND {
+        match self.probe_reachability().await {
-                    // NOT_FOUND is acceptable for HEAD probes — the endpoint may not
+            Ok(true) => {
-                    // exist as a HEAD route, but the server is reachable.
+                debug!("Webhook target {} reachability probe succeeded via {:?}", self.id, self.health_check_url);
-                    debug!("Webhook target {} HEAD probe returned {}", self.id, status);
-                } else if status == StatusCode::METHOD_NOT_ALLOWED {
-                    // Server is reachable but doesn't support HEAD — still valid.
-                    debug!("Webhook target {} HEAD probe: METHOD_NOT_ALLOWED (reachable)", self.id);
-                } else {
-                    warn!("Webhook target {} HEAD probe returned {}", self.id, status);
-                }
             }
-            Ok(Err(e)) => {
+            Ok(false) => {
-                // Connection-level error (DNS, TLS, refused, timeout)
+                return Err(TargetError::NotConnected);
-                return Err(if e.is_timeout() || e.is_connect() {
-                    TargetError::NotConnected
-                } else {
-                    TargetError::Network(format!("Webhook HEAD probe failed: {e}"))
-                });
             }
-            Err(_) => {
+            Err(err) => {
-                return Err(TargetError::Timeout("Webhook HEAD probe timed out".to_string()));
+                return Err(err);
             }
         }
 
@@ -352,27 +391,11 @@ where
     }
 
     async fn is_active(&self) -> Result<bool, TargetError> {
-        match tokio::time::timeout(Duration::from_secs(5), self.http_client.head(self.args.endpoint.as_str()).send()).await {
+        if !self.args.enable {
-            Ok(Ok(resp)) => {
+            return Ok(false);
-                let status = resp.status();
-                if status.is_server_error() {
-                    debug!("Webhook {} server error: {}", self.id, status);
-                    Ok(false)
-                } else {
-                    debug!("Webhook {} is reachable (status: {})", self.id, status);
-                    Ok(true)
-                }
-            }
-            Ok(Err(e)) => {
-                debug!("Webhook {} request failed: {}", self.id, e);
-                if e.is_timeout() || e.is_connect() {
-                    Err(TargetError::NotConnected)
-                } else {
-                    Err(TargetError::Network(format!("Webhook health check failed: {e}")))
-                }
-            }
-            Err(_) => Err(TargetError::Timeout("Webhook health check timed out".to_string())),
         }
+
+        self.probe_reachability().await
     }
 
     async fn save(&self, event: Arc<EntityTarget<E>>) -> Result<(), TargetError> {
@@ -478,8 +501,10 @@ where
 
 #[cfg(test)]
 mod tests {
-    use super::WebhookArgs;
+    use super::{WebhookArgs, WebhookTarget};
-    use crate::target::{TargetType, decode_object_name};
+    use crate::target::{Target, TargetType, decode_object_name};
+    use tokio::net::TcpListener;
+    use tokio::sync::mpsc;
     use url::Url;
     use url::form_urlencoded;
 
@@ -573,4 +598,77 @@ mod tests {
         let decoded = decode_object_name(&form_encoded).unwrap();
         assert_eq!(decoded, object_name);
     }
+
+    #[test]
+    fn test_health_check_url_ignores_endpoint_path() {
+        let endpoint = Url::parse("https://example.com:9443/hook/path").unwrap();
+        let health_check_url = WebhookTarget::<serde_json::Value>::health_check_url(&endpoint).unwrap();
+
+        assert_eq!(health_check_url.as_str(), "https://example.com:9443/");
+    }
+
+    #[tokio::test]
+    async fn test_disabled_target_can_be_constructed_without_origin_probe() {
+        let args = WebhookArgs {
+            enable: false,
+            endpoint: Url::parse("about:blank").unwrap(),
+            ..base_args()
+        };
+        let target = WebhookTarget::<serde_json::Value>::new("disabled-target".to_string(), args).unwrap();
+
+        assert!(!target.is_active().await.unwrap());
+    }
+
+    #[tokio::test]
+    async fn test_is_active_uses_origin_reachability_for_path_endpoints() {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let address = listener.local_addr().unwrap();
+        let (path_tx, mut path_rx) = mpsc::channel(1);
+        let accept_task = tokio::spawn(async move {
+            loop {
+                let (mut stream, _) = listener.accept().await.unwrap();
+                let path_tx = path_tx.clone();
+                tokio::spawn(async move {
+                    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+                    let mut request = Vec::new();
+                    let mut buf = [0u8; 1024];
+                    loop {
+                        let read = stream.read(&mut buf).await.unwrap();
+                        if read == 0 {
+                            break;
+                        }
+                        request.extend_from_slice(&buf[..read]);
+                        if request.windows(4).any(|window| window == b"\r\n\r\n") {
+                            break;
+                        }
+                    }
+
+                    let request_line = request
+                        .split(|byte| *byte == b'\n')
+                        .next()
+                        .and_then(|line| std::str::from_utf8(line).ok())
+                        .unwrap_or_default()
+                        .trim();
+                    let path = request_line.split_whitespace().nth(1).unwrap_or_default().to_string();
+                    let _ = path_tx.send(path.clone()).await;
+
+                    if path == "/" {
+                        let response = b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n";
+                        let _ = stream.write_all(response).await;
+                    }
+                });
+            }
+        });
+
+        let args = WebhookArgs {
+            endpoint: Url::parse(&format!("http://{address}/hook")).unwrap(),
+            ..base_args()
+        };
+        let target = WebhookTarget::<serde_json::Value>::new("path-probe".to_string(), args).unwrap();
+
+        assert!(target.is_active().await.unwrap());
+        assert_eq!(path_rx.recv().await.unwrap(), "/");
+        accept_task.abort();
+    }
 }