fix(connectors): Harden Elasticsearch sink test readiness

core/connectors/sinks/elasticsearch_sink/src/lib.rs

@@ -31,11 +31,14 @@ use secrecy::{ExposeSecret, SecretString};
 use serde::{Deserialize, Serialize};
 use serde_json::json;
 use simd_json::{OwnedValue, prelude::*};
+use std::time::Duration;
 use tokio::sync::Mutex;
 use tracing::{info, warn};
 
 sink_connector!(ElasticsearchSink);
 
+const DEFAULT_TIMEOUT_SECONDS: u64 = 30;
+
 #[derive(Debug)]
 struct State {
     invocations_count: usize,
@@ -83,7 +86,15 @@ impl ElasticsearchSink {
             .map_err(|error| Error::Connection(format!("Invalid Elasticsearch URL: {error}")))?;
 
         let conn_pool = elasticsearch::http::transport::SingleNodeConnectionPool::new(url);
-        let mut transport_builder = TransportBuilder::new(conn_pool);
+        // elasticsearch-rs defaults to no timeout; without this, a hung ES
+        // connection blocks FFI open() and the connectors HTTP health never binds.
+        let timeout_seconds = self
+            .config
+            .timeout_seconds
+            .unwrap_or(DEFAULT_TIMEOUT_SECONDS)
+            .max(1);
+        let mut transport_builder =
+            TransportBuilder::new(conn_pool).timeout(Duration::from_secs(timeout_seconds));
 
         if let (Some(username), Some(password)) = (&self.config.username, &self.config.password) {
             let credentials =

core/integration/src/harness/handle/connectors_runtime.rs

@@ -243,19 +243,33 @@ impl IggyServerDependent for ConnectorsRuntimeHandle {
     }
 
     async fn wait_ready(&mut self) -> Result<(), TestBinaryError> {
-        let http_address = self.http_url();
+        // Prefer /health over `/` so readiness means the connectors API is up,
+        // not some other listener that happened to bind the reserved port.
+        let health_url = format!("{}/health", self.http_url());
         let client = reqwest::Client::new();
 
         for retry in 0..common::DEFAULT_HEALTH_CHECK_RETRIES {
-            match client.get(&http_address).send().await {
-                Ok(_) => {
+            if let Some(pid) = self.pid()
+                && !common::is_process_alive(pid)
+            {
+                let (stdout, stderr) = self.collect_logs();
+                return Err(TestBinaryError::ProcessCrashed {
+                    binary: "iggy-connectors".to_string(),
+                    exit_code: None,
+                    stdout,
+                    stderr,
+                });
+            }
+
+            match client.get(&health_url).send().await {
+                Ok(response) if response.status().is_success() => {
                     return Ok(());
                 }
-                Err(_) => {
+                Ok(_) | Err(_) => {
                     if retry == common::DEFAULT_HEALTH_CHECK_RETRIES - 1 {
                         return Err(TestBinaryError::HealthCheckFailed {
                             binary: "iggy-connectors".to_string(),
-                            address: http_address,
+                            address: health_url,
                             retries: common::DEFAULT_HEALTH_CHECK_RETRIES,
                         });
                     }

core/integration/tests/connectors/fixtures/elasticsearch/container.rs

@@ -20,13 +20,16 @@ use reqwest_middleware::ClientWithMiddleware as HttpClient;
 use reqwest_retry::RetryTransientMiddleware;
 use reqwest_retry::policies::ExponentialBackoff;
 use serde::Deserialize;
+use std::io::Read;
+use std::process::{Command, Stdio};
+use std::time::{Duration, Instant};
 use testcontainers_modules::testcontainers::core::wait::HttpWaitStrategy;
 use testcontainers_modules::testcontainers::core::{IntoContainerPort, WaitFor};
 use testcontainers_modules::testcontainers::runners::AsyncRunner;
 use testcontainers_modules::testcontainers::{
     ContainerAsync, GenericImage, ImageExt, ReuseDirective,
 };
-use tracing::info;
+use tracing::{info, warn};
 
 const ELASTICSEARCH_IMAGE: &str = "docker.io/library/elasticsearch";
 const ELASTICSEARCH_TAG: &str = "9.3.0";
@@ -37,6 +40,17 @@ const ELASTICSEARCH_HEALTH_ENDPOINT: &str = "/_cluster/health";
 // name. Per-test isolation comes from a unique index per fixture, not a fresh
 // container.
 const ELASTICSEARCH_CONTAINER_NAME: &str = "iggy-test-elasticsearch";
+// Short probe timeouts: create_http_client() uses 30s + retries and must not
+// be used for readiness. One hung attempt there looks like a 60s+ test hang.
+const CLUSTER_READY_ATTEMPTS: usize = 40;
+const CLUSTER_READY_INTERVAL_MS: u64 = 250;
+const CLUSTER_READY_REQUEST_TIMEOUT_MS: u64 = 2_000;
+const DOCKER_RM_TIMEOUT_SECS: u64 = 15;
+// Indices from prior runs older than this are leftovers: a live concurrent
+// test's index is seconds old, so age-based sweeping never races other tests
+// sharing the reused container.
+const STALE_INDEX_MAX_AGE_MS: u128 = 30 * 60 * 1000;
+const STALE_INDEX_PATTERNS: &str = "iggy_messages_*,test_documents_*";
 
 pub const DEFAULT_TEST_STREAM: &str = "test_stream";
 pub const DEFAULT_TEST_TOPIC: &str = "test_topic";
@@ -101,6 +115,23 @@ pub struct ElasticsearchContainer {
 
 impl ElasticsearchContainer {
     pub async fn start() -> Result<Self, TestBinaryError> {
+        match Self::try_start().await {
+            Ok(started) => Ok(started),
+            Err(first_error) => {
+                // A reused container can be wedged (running but unhealthy:
+                // OOM-killed JVM, corrupted data dir, dead port mapping).
+                // Remove it and retry once with a fresh container instead of
+                // failing every test until someone cleans up manually.
+                warn!(
+                    "Elasticsearch container unusable, removing '{ELASTICSEARCH_CONTAINER_NAME}' and retrying once: {first_error}"
+                );
+                force_remove_container();
+                Self::try_start().await
+            }
+        }
+    }
+
+    async fn try_start() -> Result<Self, TestBinaryError> {
         let container = GenericImage::new(ELASTICSEARCH_IMAGE, ELASTICSEARCH_TAG)
             .with_exposed_port(ELASTICSEARCH_PORT.tcp())
             .with_wait_for(WaitFor::http(
@@ -137,14 +168,216 @@ impl ElasticsearchContainer {
                 message: "No mapping for Elasticsearch port".to_string(),
             })?;
 
-        let base_url = format!("http://localhost:{mapped_port}");
+        // Prefer IPv4 loopback: Docker publishes 0.0.0.0:HOST→9200. `localhost`
+        // can resolve to ::1 first on macOS and black-hole the elasticsearch-rs
+        // client while the fixture's reqwest client still looks healthy.
+        let base_url = format!("http://127.0.0.1:{mapped_port}");
         info!("Elasticsearch container available at {base_url}");
 
-        Ok(Self {
+        let started = Self {
             container,
             base_url,
+        };
+        // ReuseDirective::Always can attach to a days-old container without
+        // re-running HttpWaitStrategy; verify cluster health on every setup.
+        started.wait_until_ready().await?;
+        started.sweep_stale_indices().await;
+        Ok(started)
+    }
+
+    async fn wait_until_ready(&self) -> Result<(), TestBinaryError> {
+        // Dedicated probe client: short timeout, no retry middleware. The shared
+        // create_http_client() (30s + 3 retries) turns one black-holed request
+        // into a multi-minute hang that looks like the test is stuck.
+        let client = reqwest::Client::builder()
+            .timeout(std::time::Duration::from_millis(
+                CLUSTER_READY_REQUEST_TIMEOUT_MS,
+            ))
+            .build()
+            .map_err(|error| TestBinaryError::FixtureSetup {
+                fixture_type: "ElasticsearchContainer".to_string(),
+                message: format!("Failed to build readiness HTTP client: {error}"),
+            })?;
+        // timeout=1s keeps ES from holding the request when the cluster is slow.
+        let health_url = format!(
+            "{}{ELASTICSEARCH_HEALTH_ENDPOINT}?timeout=1s",
+            self.base_url
+        );
+        let mut last_error = String::from("no attempts made");
+
+        for attempt in 1..=CLUSTER_READY_ATTEMPTS {
+            match client.get(&health_url).send().await {
+                Ok(response) if response.status().is_success() => {
+                    let body = response.text().await.unwrap_or_default();
+                    if body.contains("\"timed_out\":true") {
+                        last_error = format!(
+                            "cluster health timed out on attempt {attempt}/{CLUSTER_READY_ATTEMPTS}: {body}"
+                        );
+                    } else {
+                        info!("Elasticsearch cluster ready at {}", self.base_url);
+                        return Ok(());
+                    }
+                }
+                Ok(response) => {
+                    last_error = format!(
+                        "cluster health status {} on attempt {attempt}/{CLUSTER_READY_ATTEMPTS}",
+                        response.status()
+                    );
+                }
+                Err(error) => {
+                    last_error = format!(
+                        "cluster health request failed on attempt {attempt}/{CLUSTER_READY_ATTEMPTS}: {error}"
+                    );
+                }
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(CLUSTER_READY_INTERVAL_MS)).await;
+        }
+
+        Err(TestBinaryError::FixtureSetup {
+            fixture_type: "ElasticsearchContainer".to_string(),
+            message: format!(
+                "Elasticsearch at {} not ready after {CLUSTER_READY_ATTEMPTS} attempts: {last_error}",
+                self.base_url
+            ),
         })
     }
+
+    /// Delete leftover test indices (empty or partially filled) from previous
+    /// runs so accumulated shards do not degrade the reused container. Only
+    /// indices older than [`STALE_INDEX_MAX_AGE_MS`] are removed, which keeps
+    /// the sweep safe against tests running concurrently in other processes.
+    /// Best-effort: failures are logged, never fail the fixture.
+    async fn sweep_stale_indices(&self) {
+        #[derive(Deserialize)]
+        struct CatIndexEntry {
+            index: String,
+            #[serde(rename = "creation.date")]
+            creation_date: Option<String>,
+        }
+
+        // Short timeout, no retries: sweep is best-effort and must not stall setup.
+        let Ok(client) = reqwest::Client::builder()
+            .timeout(std::time::Duration::from_secs(5))
+            .build()
+        else {
+            warn!("Skipping stale index sweep, failed to build HTTP client");
+            return;
+        };
+        let cat_url = format!(
+            "{}/_cat/indices/{STALE_INDEX_PATTERNS}?format=json&h=index,creation.date",
+            self.base_url
+        );
+
+        let entries = match client.get(&cat_url).send().await {
+            Ok(response) if response.status().is_success() => {
+                match response.json::<Vec<CatIndexEntry>>().await {
+                    Ok(entries) => entries,
+                    Err(error) => {
+                        warn!("Skipping stale index sweep, unparsable _cat response: {error}");
+                        return;
+                    }
+                }
+            }
+            Ok(response) => {
+                warn!(
+                    "Skipping stale index sweep, _cat/indices returned {}",
+                    response.status()
+                );
+                return;
+            }
+            Err(error) => {
+                warn!("Skipping stale index sweep, _cat/indices failed: {error}");
+                return;
+            }
+        };
+
+        let now_ms = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .map(|duration| duration.as_millis())
+            .unwrap_or(0);
+
+        let stale: Vec<String> = entries
+            .into_iter()
+            .filter_map(|entry| {
+                let created_ms = entry.creation_date.as_deref()?.parse::<u128>().ok()?;
+                (now_ms.saturating_sub(created_ms) > STALE_INDEX_MAX_AGE_MS).then_some(entry.index)
+            })
+            .collect();
+
+        if stale.is_empty() {
+            return;
+        }
+
+        // Chunked so the URL stays well under limits with many leftovers.
+        for chunk in stale.chunks(20) {
+            let delete_url = format!("{}/{}", self.base_url, chunk.join(","));
+            match client.delete(&delete_url).send().await {
+                Ok(response) if response.status().is_success() => {
+                    info!("Deleted {} stale Elasticsearch test indices", chunk.len());
+                }
+                Ok(response) => {
+                    warn!(
+                        "Failed to delete stale Elasticsearch indices, status {}",
+                        response.status()
+                    );
+                }
+                Err(error) => {
+                    warn!("Failed to delete stale Elasticsearch indices: {error}");
+                }
+            }
+        }
+    }
+}
+
+/// Remove the shared reuse container so the next start creates a fresh one.
+/// Uses the Docker CLI directly: testcontainers offers no "remove by name" API
+/// and the wedged container was created by an earlier process anyway.
+fn force_remove_container() {
+    let mut child = match Command::new("docker")
+        .args(["rm", "-f", ELASTICSEARCH_CONTAINER_NAME])
+        .stdout(Stdio::null())
+        .stderr(Stdio::piped())
+        .spawn()
+    {
+        Ok(child) => child,
+        Err(error) => {
+            warn!("Failed to invoke docker rm for '{ELASTICSEARCH_CONTAINER_NAME}': {error}");
+            return;
+        }
+    };
+
+    let deadline = Instant::now() + Duration::from_secs(DOCKER_RM_TIMEOUT_SECS);
+    loop {
+        match child.try_wait() {
+            Ok(Some(status)) if status.success() => {
+                info!("Removed wedged Elasticsearch container '{ELASTICSEARCH_CONTAINER_NAME}'");
+                return;
+            }
+            Ok(Some(status)) => {
+                let mut stderr = String::new();
+                if let Some(mut pipe) = child.stderr.take() {
+                    let _ = pipe.read_to_string(&mut stderr);
+                }
+                warn!(
+                    "docker rm -f {ELASTICSEARCH_CONTAINER_NAME} failed (exit {status}): {stderr}"
+                );
+                return;
+            }
+            Ok(None) if Instant::now() >= deadline => {
+                let _ = child.kill();
+                let _ = child.wait();
+                warn!(
+                    "docker rm -f {ELASTICSEARCH_CONTAINER_NAME} timed out after {DOCKER_RM_TIMEOUT_SECS}s"
+                );
+                return;
+            }
+            Ok(None) => std::thread::sleep(Duration::from_millis(100)),
+            Err(error) => {
+                warn!("Failed waiting for docker rm '{ELASTICSEARCH_CONTAINER_NAME}': {error}");
+                return;
+            }
+        }
+    }
 }
 
 pub fn create_http_client() -> HttpClient {