From 82e340fdce9249e45f46d34150b8c0acea1c62c7 Mon Sep 17 00:00:00 2001 From: oyster.cafe Date: Wed, 06 May 2026 19:00:22 +0000 Subject: [PATCH] Binding multiple listen addresses I would like to bind multiple addresses! Linux may do funky stuff with v6es in and of themselves, but just making it specific that one can do a v4 and v6 separately if they want --- Cargo.lock | 1 + Cargo.toml | 1 + docs/configuration.md | 4 ++-- src/main.rs | 57 ++++++++++++++++++++++++++++++++++++++++++++++----------- tests/common.nu | 2 +- tests/throttling.nu | 2 +- tests/verify_crawler.nu | 2 +- src/api/mod.rs | 69 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++------------ src/control/mod.rs | 41 +++++++++++++++++++++++++++++++++++------ 9 file(s) changed, 145 insertion(s)(+), 34 deletion(s)(-) diff --git a/Cargo.lock b/Cargo.lock --- a/Cargo.lock +++ b/Cargo.lock @@ -1533,6 +1533,7 @@ "serde_urlencoded", "sha2", "smol_str", + "socket2", "tempfile", "thiserror 2.0.18", "tokio", diff --git a/Cargo.toml b/Cargo.toml --- a/Cargo.toml +++ b/Cargo.toml @@ -14,6 +14,7 @@ [dependencies] tokio = { version = "1.0", features = ["full"] } tokio-util = { version = "0.7", features = ["io"] } +socket2 = "0.6" tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } diff --git a/docs/configuration.md b/docs/configuration.md --- a/docs/configuration.md +++ b/docs/configuration.md @@ -10,9 +10,9 @@ | :--- | :--- | :--- | | `DATABASE_PATH` | `./hydrant.db` | path to the database folder | | `RUST_LOG` | `info` | log filter directives (e.g., `debug`, `hydrant=trace`). [tracing env-filter syntax](https://docs.rs/tracing-subscriber/latest/tracing_subscriber/filter/struct.EnvFilter.html) | -| `API_PORT` | `3000` | port for the API server | +| `API_BIND` | `0.0.0.0:3000,[::]:3000` | comma-separated list of `:` socket addresses to bind the API server to. literal IPs only (hostnames not resolved). when both an ipv4 and ipv6 entry share the same port, the v6 listener is set to v6-only to avoid bind collision; a lone `[::]:` listens dual-stack | | `ENABLE_DEBUG` | `false` | enable debug endpoints | -| `DEBUG_PORT` | `API_PORT + 1` | port for debug endpoints (if enabled) | +| `DEBUG_PORT` | first `API_BIND` port + 1 | port for debug endpoints (if enabled) | ## indexing mode diff --git a/src/main.rs b/src/main.rs --- a/src/main.rs +++ b/src/main.rs @@ -1,27 +1,62 @@ use futures::FutureExt; use hydrant::config::Config; -use hydrant::control::Hydrant; +use hydrant::control::{ApiBinds, Hydrant}; use mimalloc::MiMalloc; +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; + +const DEFAULT_API_PORT: u16 = 3000; +const DEFAULT_DEBUG_PORT: u16 = DEFAULT_API_PORT + 1; + +fn default_api_binds() -> ApiBinds { + ApiBinds::try_from_iter([ + SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), DEFAULT_API_PORT), + SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), DEFAULT_API_PORT), + ]) + .expect("two-element literal is non-empty") +} struct AppConfig { - api_port: u16, + api_binds: ApiBinds, enable_debug: bool, debug_port: u16, } impl AppConfig { - fn from_env() -> Self { + fn from_env() -> miette::Result { use hydrant::__cfg as cfg; - let api_port = cfg!("API_PORT", 3000u16); + let api_binds = parse_api_binds()?; let enable_debug = cfg!("ENABLE_DEBUG", false); - let debug_port: u16 = api_port + 1; - let debug_port = cfg!("DEBUG_PORT", debug_port); - Self { - api_port, + let debug_port_default = api_binds + .iter() + .next() + .expect("ApiBinds is non-empty by construction") + .port() + .checked_add(1) + .unwrap_or(DEFAULT_DEBUG_PORT); + let debug_port = cfg!("DEBUG_PORT", debug_port_default); + Ok(Self { + api_binds, enable_debug, debug_port, - } + }) } +} + +fn parse_api_binds() -> miette::Result { + let Ok(raw) = std::env::var("HYDRANT_API_BIND") else { + return Ok(default_api_binds()); + }; + let parsed = raw + .split(',') + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(|s| { + s.parse::() + .map_err(|e| miette::miette!("invalid HYDRANT_API_BIND entry `{s}`: {e}")) + }) + .collect::>>()?; + ApiBinds::try_from_iter(parsed) + .ok_or_else(|| miette::miette!("HYDRANT_API_BIND is set but contains no addresses")) } #[global_allocator] @@ -34,7 +69,7 @@ .ok(); let cfg = Config::from_env()?; - let app = AppConfig::from_env(); + let app = AppConfig::from_env()?; let env_filter = tracing_subscriber::EnvFilter::builder() .with_default_directive(tracing::Level::INFO.into()) @@ -50,7 +85,7 @@ tokio::select! { r = hydrant.run()? => r, - r = hydrant.serve(app.api_port) => r, + r = hydrant.serve(app.api_binds) => r, r = debug_fut => r, } } diff --git a/tests/common.nu b/tests/common.nu --- a/tests/common.nu +++ b/tests/common.nu @@ -141,7 +141,7 @@ let env_vars = { HYDRANT_DATABASE_PATH: ($db_path), HYDRANT_FULL_NETWORK: "false", - HYDRANT_API_PORT: ($port | into string), + HYDRANT_API_BIND: $"127.0.0.1:($port)", HYDRANT_ENABLE_DEBUG: "true", HYDRANT_DEBUG_PORT: (resolve-test-debug-port ($port + 1) | into string), HYDRANT_PLC_URL: "https://plc.klbr.net", diff --git a/tests/throttling.nu b/tests/throttling.nu --- a/tests/throttling.nu +++ b/tests/throttling.nu @@ -40,7 +40,7 @@ HYDRANT_RELAY_HOST: ($mock_url), HYDRANT_DISABLE_FIREHOSE: "true", HYDRANT_DISABLE_BACKFILL: "true", # disable backfill so pending count stays up - HYDRANT_API_PORT: ($port | into string), + HYDRANT_API_BIND: $"127.0.0.1:($port)", HYDRANT_LOG_LEVEL: "debug", RUST_LOG: "debug", HYDRANT_CRAWLER_MAX_PENDING_REPOS: "2", diff --git a/tests/verify_crawler.nu b/tests/verify_crawler.nu --- a/tests/verify_crawler.nu +++ b/tests/verify_crawler.nu @@ -45,7 +45,7 @@ HYDRANT_RELAY_HOST: ($mock_url), HYDRANT_DISABLE_FIREHOSE: "true", HYDRANT_DISABLE_BACKFILL: "true", - HYDRANT_API_PORT: ($port | into string), + HYDRANT_API_BIND: $"127.0.0.1:($port)", HYDRANT_ENABLE_DEBUG: "true", # for stats checking HYDRANT_DEBUG_PORT: (resolve-test-debug-port ($port + 1) | into string), HYDRANT_LOG_LEVEL: "debug", diff --git a/src/api/mod.rs b/src/api/mod.rs --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -1,9 +1,11 @@ -use crate::control::Hydrant; +use crate::control::{ApiBinds, Hydrant}; use crate::state::AppState; use axum::{Router, routing::get}; use std::{net::SocketAddr, sync::Arc}; use tower_http::cors::CorsLayer; use tower_http::trace::TraceLayer; + +const LISTEN_BACKLOG: i32 = 1024; #[cfg(feature = "indexer")] mod crawler; @@ -19,7 +21,7 @@ mod stream; mod xrpc; -pub async fn serve(hydrant: Hydrant, port: u16) -> miette::Result<()> { +pub async fn serve(hydrant: Hydrant, binds: ApiBinds) -> miette::Result<()> { let blocks_available = hydrant.state.is_block_storage_enabled(); let app = Router::new() .route( @@ -61,20 +63,63 @@ .layer(TraceLayer::new_for_http()) .layer(CorsLayer::permissive()); - let listener = tokio::net::TcpListener::bind(format!("0.0.0.0:{port}")) - .await - .map_err(|e| miette::miette!("failed to bind to port {port}: {e}"))?; + let bind_list: Vec = binds.iter().collect(); + let listeners: Vec = bind_list + .iter() + .map(|&addr| { + let v6only = v6only_for(addr, &bind_list); + let listener = bind_listener(addr, v6only) + .map_err(|e| miette::miette!("failed to bind to {addr}: {e}"))?; + tracing::info!("API server listening on {}", listener.local_addr().unwrap()); + Ok::<_, miette::Report>(listener) + }) + .collect::>()?; - tracing::info!("API server listening on {}", listener.local_addr().unwrap()); + let services = listeners.into_iter().map(|listener| { + let app = app.clone(); + async move { + axum::serve( + listener, + app.into_make_service_with_connect_info::(), + ) + .await + .map_err(|e| miette::miette!("axum server error: {e}")) + } + }); - axum::serve( - listener, - app.into_make_service_with_connect_info::(), - ) - .await - .map_err(|e| miette::miette!("axum server error: {e}"))?; + futures::future::try_join_all(services).await?; Ok(()) +} + +fn v6only_for(addr: SocketAddr, all: &[SocketAddr]) -> Option { + let SocketAddr::V6(v6) = addr else { + return None; + }; + let has_v4_sibling = all + .iter() + .any(|a| matches!(a, SocketAddr::V4(v4) if v4.port() == v6.port())); + Some(has_v4_sibling) +} + +fn bind_listener( + addr: SocketAddr, + v6only: Option, +) -> std::io::Result { + let domain = match addr { + SocketAddr::V4(_) => socket2::Domain::IPV4, + SocketAddr::V6(_) => socket2::Domain::IPV6, + }; + let socket = socket2::Socket::new(domain, socket2::Type::STREAM, Some(socket2::Protocol::TCP))?; + if let Some(only) = v6only { + socket.set_only_v6(only)?; + } + socket.set_reuse_address(true)?; + socket.set_nonblocking(true)?; + socket.bind(&addr.into())?; + socket.listen(LISTEN_BACKLOG)?; + let std_listener: std::net::TcpListener = socket.into(); + tokio::net::TcpListener::from_std(std_listener) } pub async fn serve_debug(state: Arc, port: u16) -> miette::Result<()> { diff --git a/src/control/mod.rs b/src/control/mod.rs --- a/src/control/mod.rs +++ b/src/control/mod.rs @@ -58,6 +58,34 @@ use url::Url; #[derive(Debug, Clone)] +pub struct ApiBinds { + head: std::net::SocketAddr, + tail: Vec, +} + +impl ApiBinds { + pub fn new(addr: std::net::SocketAddr) -> Self { + Self { + head: addr, + tail: Vec::new(), + } + } + + pub fn try_from_iter>(iter: I) -> Option { + let mut iter = iter.into_iter(); + let head = iter.next()?; + Some(Self { + head, + tail: iter.collect(), + }) + } + + pub fn iter(&self) -> impl Iterator + '_ { + std::iter::once(self.head).chain(self.tail.iter().copied()) + } +} + +#[derive(Debug, Clone)] /// infromation about a host hydrant is consuming from. pub struct Host { /// hostname of the host. @@ -90,15 +118,16 @@ /// # example /// /// ```rust,no_run -/// use hydrant::control::Hydrant; +/// use hydrant::control::{ApiBinds, Hydrant}; /// /// #[tokio::main] /// async fn main() -> miette::Result<()> { /// let hydrant = Hydrant::from_env().await?; +/// let binds = ApiBinds::new("0.0.0.0:3000".parse().unwrap()); /// /// tokio::select! { -/// r = hydrant.run()? => r, -/// r = hydrant.serve(3000) => r, +/// r = hydrant.run()? => r, +/// r = hydrant.serve(binds) => r, /// } /// } /// ``` @@ -806,7 +835,7 @@ Ok(StatsResponse { counts, sizes }) } - /// returns a future that runs the HTTP management API server on `0.0.0.0:{port}`. + /// returns a future that runs the HTTP management API server on the given bind addresses. /// /// the server exposes all management endpoints (`/filter`, `/repos`, `/ingestion`, /// `/stream`, `/stats`, `/db/*`, `/xrpc/*`). it runs indefinitely and resolves @@ -816,9 +845,9 @@ /// of `self` is deferred until the future is first polled. /// /// to disable the HTTP API entirely, simply don't call this method. - pub fn serve(&self, port: u16) -> impl Future> { + pub fn serve(&self, binds: ApiBinds) -> impl Future> { let hydrant = self.clone(); - async move { crate::api::serve(hydrant, port).await } + async move { crate::api::serve(hydrant, binds).await } } /// returns a future that runs the debug HTTP API server on `127.0.0.1:{port}`. -- tangled.sh