diff --git a/crates/e2e_test/src/common.rs b/crates/e2e_test/src/common.rs index 7fb344bb4..b34f65c55 100644 --- a/crates/e2e_test/src/common.rs +++ b/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> { 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) - .env("RUST_LOG", "rustfs=info,rustfs_notify=debug") - .args(&args) - .spawn()?; + let mut command = Command::new(&binary_path); + command.env("RUST_LOG", "rustfs=info,rustfs_notify=debug"); + for (key, value) in extra_env { + 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> { - 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> { + 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> { - 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. diff --git a/crates/e2e_test/src/object_lambda_test.rs b/crates/e2e_test/src/object_lambda_test.rs index 658b799de..0e35f4e5d 100644 --- a/crates/e2e_test/src/object_lambda_test.rs +++ b/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> { + 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, + tokio::task::JoinHandle>>, + ), + Box, +> { + 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> { + 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> { diff --git a/crates/targets/src/target/webhook.rs b/crates/targets/src/target/webhook.rs index e0437359f..de632b7e5 100644 --- a/crates/targets/src/target/webhook.rs +++ b/crates/targets/src/target/webhook.rs @@ -106,6 +106,7 @@ where { id: TargetID, args: WebhookArgs, + health_check_url: Option, http_client: Arc, // Add Send + Sync constraints to ensure thread safety store: Option + Send + Sync>>, @@ -124,6 +125,7 @@ where Box::new(WebhookTarget:: { 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:: { 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 { + 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 { + 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) - // unlike TCP connect which can't detect proxy issues. - 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)) => { - let status = resp.status(); - if status.is_success() || status == StatusCode::NOT_FOUND { - // NOT_FOUND is acceptable for HEAD probes — the endpoint may not - // exist as a HEAD route, but the server is reachable. - 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); - } + if !self.args.enable { + return Ok(()); + } + + // Use the configured reqwest client against the origin URL so proxy and TLS + // behavior matches real delivery while avoiding path-specific false negatives. + match self.probe_reachability().await { + Ok(true) => { + debug!("Webhook target {} reachability probe succeeded via {:?}", self.id, self.health_check_url); } - Ok(Err(e)) => { - // Connection-level error (DNS, TLS, refused, timeout) - return Err(if e.is_timeout() || e.is_connect() { - TargetError::NotConnected - } else { - TargetError::Network(format!("Webhook HEAD probe failed: {e}")) - }); + Ok(false) => { + return Err(TargetError::NotConnected); } - Err(_) => { - return Err(TargetError::Timeout("Webhook HEAD probe timed out".to_string())); + Err(err) => { + return Err(err); } } @@ -352,27 +391,11 @@ where } async fn is_active(&self) -> Result { - match tokio::time::timeout(Duration::from_secs(5), self.http_client.head(self.args.endpoint.as_str()).send()).await { - Ok(Ok(resp)) => { - 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())), + if !self.args.enable { + return Ok(false); } + + self.probe_reachability().await } async fn save(&self, event: Arc>) -> Result<(), TargetError> { @@ -478,8 +501,10 @@ where #[cfg(test)] mod tests { - use super::WebhookArgs; - use crate::target::{TargetType, decode_object_name}; + use super::{WebhookArgs, WebhookTarget}; + 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::::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::::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::::new("path-probe".to_string(), args).unwrap(); + + assert!(target.is_active().await.unwrap()); + assert_eq!(path_rx.recv().await.unwrap(), "/"); + accept_task.abort(); + } }