diff --git a/Cargo.lock b/Cargo.lock index 8c3389e..daac693 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1533,6 +1533,7 @@ dependencies = [ "serde_urlencoded", "sha2", "smol_str", + "socket2", "tempfile", "thiserror 2.0.18", "tokio", diff --git a/Cargo.toml b/Cargo.toml index ee92b98..755a9fb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -14,6 +14,7 @@ indexer_stream = ["indexer"] [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 index 9072233..0d5a717 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -10,9 +10,9 @@ hydrant is configured via environment variables, all prefixed with `HYDRANT_` (e | :--- | :--- | :--- | | `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/api/mod.rs b/src/api/mod.rs index 027ece6..2f1e7e4 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -1,10 +1,12 @@ -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; mod db; @@ -19,7 +21,7 @@ mod stats; 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,22 +63,65 @@ pub async fn serve(hydrant: Hydrant, port: u16) -> miette::Result<()> { .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<()> { let app = debug::router() .with_state(state) diff --git a/src/control/mod.rs b/src/control/mod.rs index 7615ed7..d43887e 100644 --- a/src/control/mod.rs +++ b/src/control/mod.rs @@ -57,6 +57,34 @@ use stream::event_stream_thread; use stream::relay_stream_thread; 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 { @@ -90,15 +118,16 @@ pub type Event = MarshallableEvt<'static>; /// # 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 @@ impl Hydrant { 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 @@ impl Hydrant { /// 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}`. diff --git a/src/main.rs b/src/main.rs index 5ba7311..4439381 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,29 +1,64 @@ 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] static GLOBAL: MiMalloc = MiMalloc; @@ -34,7 +69,7 @@ async fn main() -> miette::Result<()> { .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 @@ async fn main() -> miette::Result<()> { 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 index bd73b5f..f43e129 100644 --- a/tests/common.nu +++ b/tests/common.nu @@ -141,7 +141,7 @@ export def start-hydrant [binary: string, db_path: string, port: int] { 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 index 3a00dac..d9a7158 100644 --- a/tests/throttling.nu +++ b/tests/throttling.nu @@ -40,7 +40,7 @@ def main [] { 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 index db50135..da4e8ab 100644 --- a/tests/verify_crawler.nu +++ b/tests/verify_crawler.nu @@ -45,7 +45,7 @@ def main [] { 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",