feat(integration): add iggy_harness proc macro and use it in mcp tests

Cargo.toml

@@ -42,6 +42,7 @@ members = [
     "core/connectors/sources/postgres_source",
     "core/connectors/sources/random_source",
     "core/consensus",
+    "core/harness_derive",
     "core/integration",
     "core/journal",
     "core/message_bus",
@@ -153,6 +154,7 @@ getrandom = { version = "0.3", features = ["wasm_js"] }
 git2 = { version = "0.20.3", default-features = false, features = ["vendored-libgit2"] }
 gloo = "0.11"
 governor = "0.10.4"
+harness_derive = { path = "core/harness_derive" }
 hash32 = "1.0.0"
 hostname = "0.4.2"
 human-repr = "1.1.0"

DEPENDENCIES.md

@@ -338,6 +338,7 @@ h2: 0.4.13, "MIT",
 half: 2.7.1, "Apache-2.0 OR MIT",
 halfbrown: 0.4.0, "Apache-2.0 OR MIT",
 handlebars: 6.4.0, "MIT",
+harness_derive: 0.1.0, "Apache-2.0",
 hash32: 0.2.1, "Apache-2.0 OR MIT",
 hash32: 1.0.0, "Apache-2.0 OR MIT",
 hashbrown: 0.12.3, "Apache-2.0 OR MIT",

core/ai/mcp/Cargo.toml

@@ -49,6 +49,7 @@ rmcp = { workspace = true, features = [
 ] }
 serde = { workspace = true }
 serde_json = { workspace = true }
+socket2 = "0.6"
 strum = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }

core/ai/mcp/src/api.rs

@@ -32,10 +32,58 @@ use rmcp::{
         StreamableHttpService, streamable_http_server::session::local::LocalSessionManager,
     },
 };
+use socket2::{Domain, Protocol, Socket, Type};
 use std::{net::SocketAddr, path::PathBuf, sync::Arc};
 use tokio::spawn;
 use tracing::{error, info};
 
+fn create_reusable_listener(address: &str) -> Result<std::net::TcpListener, McpRuntimeError> {
+    let addr: SocketAddr = address.parse().map_err(|_| {
+        error!("Invalid address: {address}");
+        McpRuntimeError::FailedToStartHttpServer
+    })?;
+
+    let domain = if addr.is_ipv6() {
+        Domain::IPV6
+    } else {
+        Domain::IPV4
+    };
+
+    let socket = Socket::new(domain, Type::STREAM, Some(Protocol::TCP)).map_err(|e| {
+        error!("Failed to create socket: {e}");
+        McpRuntimeError::FailedToStartHttpServer
+    })?;
+
+    socket.set_reuse_address(true).map_err(|e| {
+        error!("Failed to set SO_REUSEADDR: {e}");
+        McpRuntimeError::FailedToStartHttpServer
+    })?;
+
+    #[cfg(unix)]
+    socket.set_reuse_port(true).map_err(|e| {
+        error!("Failed to set SO_REUSEPORT: {e}");
+        McpRuntimeError::FailedToStartHttpServer
+    })?;
+
+    socket.bind(&addr.into()).map_err(|e| {
+        error!("Failed to bind to {address}: {e}");
+        McpRuntimeError::FailedToStartHttpServer
+    })?;
+
+    socket.listen(128).map_err(|e| {
+        error!("Failed to listen on {address}: {e}");
+        McpRuntimeError::FailedToStartHttpServer
+    })?;
+
+    let listener: std::net::TcpListener = socket.into();
+    listener.set_nonblocking(true).map_err(|e| {
+        error!("Failed to set non-blocking: {e}");
+        McpRuntimeError::FailedToStartHttpServer
+    })?;
+
+    Ok(listener)
+}
+
 pub async fn init(
     config: HttpConfig,
     iggy_client: Arc<IggyClient>,
@@ -77,15 +125,14 @@ pub async fn init(
     }
 
     if !config.tls.enabled {
-        let listener = tokio::net::TcpListener::bind(&config.address)
-            .await
-            .map_err(|error| {
-                error!("Failed to bind TCP listener: {:?}", error);
-                McpRuntimeError::FailedToStartHttpServer
-            })?;
-        let address = listener
+        let std_listener = create_reusable_listener(&config.address)?;
+        let address = std_listener
             .local_addr()
             .expect("Failed to get local address for HTTP server");
+        let listener = tokio::net::TcpListener::from_std(std_listener).map_err(|e| {
+            error!("Failed to convert to tokio listener: {e}");
+            McpRuntimeError::FailedToStartHttpServer
+        })?;
         info!(
             "HTTP API listening on: {address}, MCP path: {}",
             config.path
@@ -110,8 +157,7 @@ pub async fn init(
     .await
     .expect("Failed to load TLS certificate or key file");
 
-    let listener =
-        std::net::TcpListener::bind(&config.address).expect("Failed to bind TCP listener");
+    let listener = create_reusable_listener(&config.address)?;
     let address = listener
         .local_addr()
         .expect("Failed to get local address for HTTPS / TLS server");

core/configs/src/configs_impl/typed_env_provider.rs

@@ -54,6 +54,7 @@ const IGNORED_ENV_VARS: &[&str] = &[
     "IGGY_MCP_CONFIG_PATH",
     "IGGY_ROOT_PASSWORD",
     "IGGY_ROOT_USERNAME",
+    "IGGY_TEST_CLEANUP_DISABLED",
     "IGGY_TEST_VERBOSE",
 ];
 

core/harness_derive/Cargo.toml

@@ -0,0 +1,30 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[package]
+name = "harness_derive"
+version = "0.1.0"
+edition = "2024"
+license = "Apache-2.0"
+
+[lib]
+proc-macro = true
+
+[dependencies]
+proc-macro2 = { workspace = true }
+quote = { workspace = true }
+syn = { workspace = true }