This is an automated email from the ASF dual-hosted git repository. mmodzelewski pushed a commit to branch advertised-addresses in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 8e13d5f5c8ebe13cf7886b268db2a0d1342605d1 Author: Maciej Modzelewski <[email protected]> AuthorDate: Tue Aug 4 15:14:25 2026 +0200 feat(server-ng): add per-client-network advertised address selectors A node has one advertised_address, so a split-network cluster must pick a side: advertise the public address and in-VPC clients route out through the public edge, or advertise the private one and external clients cannot reach the cluster at all. Each roster node now takes advertised_addresses selectors (client_cidr + address). The transport-level peer IP picks the address by longest-prefix match, first declared wins a tie, with fallback to the catch-all advertised_address and then the roster ip. The selection feeds binary and HTTP cluster metadata and the 307 leader-redirect Location, so a redirected client stays on its own network. HTTP handlers see the peer address through axum connect-info on the plain listener; the TLS serve loop bypasses the make-service, so it stamps the same ConnectInfo extension per connection. Boot validation parses every selector strictly, rejects duplicate CIDRs within a node, and checks advertised endpoint conflicts per client network, since reusing one host:port across different networks is the point. --- Cargo.lock | 1 + Cargo.toml | 1 + core/configs/Cargo.toml | 1 + core/configs/src/server_ng_config/cluster.rs | 654 ++++++++++++++++++++- core/configs/src/server_ng_config/defaults.rs | 1 + core/configs/src/server_ng_config/validators.rs | 1 + .../tests/server/cluster_metadata_vsr.rs | 192 +++++- core/integration/tests/server/http_client.rs | 19 +- core/server-ng/config.toml | 23 + core/server-ng/src/bootstrap.rs | 9 +- core/server-ng/src/cluster_meta.rs | 112 +++- core/server-ng/src/dispatch.rs | 24 +- core/server-ng/src/http.rs | 30 +- core/server-ng/src/http/error.rs | 100 +++- core/server-ng/src/http/extractor.rs | 15 +- core/server-ng/src/http/handlers.rs | 7 +- core/server-ng/src/http/reads.rs | 3 +- core/server-ng/src/http/state.rs | 22 +- core/server-ng/src/http/tls.rs | 18 +- core/server-ng/src/responses.rs | 12 +- core/server-ng/src/session_manager.rs | 11 + 21 files changed, 1157 insertions(+), 99 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 3cd7dccbc..c8e9a5a7a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3189,6 +3189,7 @@ dependencies = [ "err_trail", "figment", "iggy_common", + "ipnet", "jsonwebtoken", "serde", "serde_json", diff --git a/Cargo.toml b/Cargo.toml index 0bd07364f..4b327d8a3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -207,6 +207,7 @@ iggy_common = { path = "core/common", version = "0.10.3-edge.3" } iggy_connector_sdk = { path = "core/connectors/sdk", version = "0.3.1-edge.1" } indexmap = "2.14.0" integration = { path = "core/integration" } +ipnet = "2.12.0" journal = { path = "core/journal" } js-sys = "0.3" jsonwebtoken = { version = "10.4.0", features = ["rust_crypto"] } diff --git a/core/configs/Cargo.toml b/core/configs/Cargo.toml index 6e9c85d2b..858ea2b17 100644 --- a/core/configs/Cargo.toml +++ b/core/configs/Cargo.toml @@ -28,6 +28,7 @@ cpu_allocation = { workspace = true } err_trail = { workspace = true } figment = { workspace = true } iggy_common = { workspace = true } +ipnet = { workspace = true } jsonwebtoken = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } diff --git a/core/configs/src/server_ng_config/cluster.rs b/core/configs/src/server_ng_config/cluster.rs index a10a5384f..f262fc64b 100644 --- a/core/configs/src/server_ng_config/cluster.rs +++ b/core/configs/src/server_ng_config/cluster.rs @@ -27,6 +27,7 @@ use crate::ConfigurationError; use crate::http::HttpJwtConfig; use configs::ConfigEnv; use iggy_common::{IggyDuration, Validatable}; +use ipnet::{IpNet, Ipv4Net}; use serde::{Deserialize, Serialize}; use serde_with::{DisplayFromStr, serde_as}; use std::fmt; @@ -367,6 +368,12 @@ pub struct ClusterNodeConfig { /// to use [`Self::ip`]. #[serde(default)] pub advertised_address: Option<String>, + /// Client-network-scoped overrides of [`Self::advertised_address`], + /// resolved by longest-prefix match over the client's IP (see + /// [`AdvertisedAddressSelector`]). Empty by default, so existing configs + /// keep the single catch-all address. + #[serde(default)] + pub advertised_addresses: Vec<AdvertisedAddressSelector>, /// Numeric replica ID for VSR consensus (0-based). /// /// Must be unique across [`ClusterConfig::nodes`] and strictly less than @@ -375,6 +382,136 @@ pub struct ClusterNodeConfig { pub ports: TransportPorts, } +/// One client-network-scoped advertised address: clients whose IP falls +/// inside `client_cidr` are told `address` instead of the node's catch-all +/// [`ClusterNodeConfig::advertised_address`]. +/// +/// Typical split-network case: the roster `ip` is VPC-private and +/// `advertised_address` is public; a selector with the VPC CIDR keeps +/// in-VPC clients on the private address while everyone else stays on the +/// public one. Selection is longest-prefix match across a node's selectors; +/// a tie on prefix length resolves to the first declared selector. Selection +/// sees the transport-level peer address, so clients arriving through a +/// proxy or load balancer match the proxy's network, not their own. +#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)] +#[serde(deny_unknown_fields)] +pub struct AdvertisedAddressSelector { + /// Client network this selector matches, in CIDR notation + /// (`10.0.0.0/16`, `2001:db8::/32`). Must parse at boot; duplicate + /// networks within one node are rejected. A v4-mapped v6 network + /// (`::ffff:10.0.0.0/104`) canonicalizes to its v4 form (`10.0.0.0/8`), + /// matching how client IPs canonicalize before matching. + pub client_cidr: String, + /// Address advertised to matching clients: a literal IP or a DNS + /// hostname, validated as [`AdvertisedAddress`] at boot (no port; ports + /// come from [`ClusterNodeConfig::ports`]). + pub address: String, +} + +/// A roster node with its advertised-address selectors parsed once, built +/// wherever a roster is assembled for serving clients (listener/shard start). +/// Per-request resolution must not re-parse selector strings: validation +/// guarantees they parse only while the cluster is enabled, so on an +/// unvalidated config a per-request parse would skip bad selectors silently +/// on every call. Here they are dropped once, loudly, at build time. +#[derive(Debug, Clone)] +pub struct ResolvedClusterNode { + pub config: ClusterNodeConfig, + /// Truncated, canonicalized selector networks with their parsed + /// addresses, in declaration order (the longest-prefix tiebreak). + selectors: Vec<(IpNet, AdvertisedAddress)>, +} + +impl From<ClusterNodeConfig> for ResolvedClusterNode { + fn from(config: ClusterNodeConfig) -> Self { + let selectors = config + .advertised_addresses + .iter() + .filter_map(|selector| { + // Parse failures only occur on an unvalidated config (the + // cluster validator rejects them while the cluster is + // enabled); drop the selector once here rather than skipping + // it silently on every request. + let network = match selector.client_cidr.parse::<IpNet>() { + Ok(network) => canonical_ip_net(network.trunc()), + Err(error) => { + eprintln!( + "Ignoring advertised_addresses selector for node '{}': client_cidr '{}': {error}", + config.name, selector.client_cidr + ); + return None; + } + }; + match selector.address.parse::<AdvertisedAddress>() { + Ok(address) => Some((network, address)), + Err(error) => { + eprintln!( + "Ignoring advertised_addresses selector for node '{}': address '{}': {error}", + config.name, selector.address + ); + None + } + } + }) + .collect(); + Self { config, selectors } + } +} + +impl ResolvedClusterNode { + /// The client-facing address for a client connecting from `client_ip`: + /// longest-prefix match over the selector networks, then the catch-all + /// [`ClusterNodeConfig::advertised_address`], then the roster + /// [`ClusterNodeConfig::ip`]. A configured catch-all that does not parse + /// fails closed (`None`) instead of leaking the roster ip - unreachable + /// after validation, but validation is skipped while the cluster is + /// disabled. + #[must_use] + pub fn advertised_for(&self, client_ip: Option<IpAddr>) -> Option<AdvertisedAddress> { + client_ip + .and_then(|client_ip| self.selector_address(client_ip)) + .or_else(|| match self.config.advertised_address.as_deref() { + Some(advertised_address) => advertised_address.parse().ok(), + None => self.config.ip.parse().ok(), + }) + } + + /// Longest-prefix match over the boot-parsed selector networks; a tie on + /// prefix length resolves to the first declared selector. The client IP + /// is canonicalized first so a v4-mapped v6 peer (`::ffff:10.0.0.7`, the + /// shape a dual-stack listener reports) matches v4 networks. + fn selector_address(&self, client_ip: IpAddr) -> Option<AdvertisedAddress> { + let client_ip = client_ip.to_canonical(); + let mut best: Option<(u8, &AdvertisedAddress)> = None; + for (network, address) in &self.selectors { + if !network.contains(&client_ip) { + continue; + } + if best.is_none_or(|(prefix_len, _)| network.prefix_len() > prefix_len) { + best = Some((network.prefix_len(), address)); + } + } + best.map(|(_, address)| address.clone()) + } +} + +/// Network-side mirror of the `IpAddr::to_canonical` applied to client IPs +/// before matching: a selector network written in v4-mapped v6 form +/// (`::ffff:10.0.0.0/104`) becomes its v4 equivalent (`10.0.0.0/8`), since a +/// canonicalized client could never match the v6 spelling. Prefixes shorter +/// than 96 bits cannot drop the `::ffff:` mapping and stay v6 (they match +/// native v6 clients only). +fn canonical_ip_net(network: IpNet) -> IpNet { + if let IpNet::V6(v6_network) = network + && v6_network.prefix_len() >= 96 + && let IpAddr::V4(v4_address) = v6_network.addr().to_canonical() + && let Ok(v4_network) = Ipv4Net::new(v4_address, v6_network.prefix_len() - 96) + { + return IpNet::V4(v4_network); + } + network +} + /// Per-node listener ports advertised in the cluster roster. In cluster mode /// the roster is the single source of ports: every enabled transport needs /// an explicit per-node port (validated at startup, no fallback to the @@ -746,8 +883,7 @@ impl Validatable<ConfigurationError> for ClusterConfig { let mut seen_ids = std::collections::HashSet::new(); let mut seen_names = std::collections::HashSet::new(); let mut used_endpoints = std::collections::HashSet::new(); - let mut used_advertised_endpoints = std::collections::HashSet::new(); - let mut used_raw_advertised_endpoints = std::collections::HashSet::new(); + let mut advertised_endpoints: Vec<AdvertisedEndpoint> = Vec::new(); for node in &self.nodes { if node.name.trim().is_empty() { @@ -838,26 +974,82 @@ impl Validatable<ConfigurationError> for ClusterConfig { None => node.ip.parse::<AdvertisedAddress>().ok(), }; + // Catch-all endpoints enter the same conflict pool as the + // selector endpoints below: a catch-all is an implicit + // match-every-client selector, so it can collide with another + // node's selector just as well as with another catch-all. for (name, port) in &client_ports { if let Some(port) = port { - let (endpoint, inserted) = match &client_address { - Some(address) => ( - address.authority(*port), - used_advertised_endpoints.insert((address.clone(), *port)), - ), - None => { - let endpoint = format!("{}:{port}", node.ip); - let inserted = used_raw_advertised_endpoints.insert(endpoint.clone()); - (endpoint, inserted) - } + let host = match &client_address { + Some(address) => AdvertisedHost::Parsed(address.clone()), + None => AdvertisedHost::Raw(&node.ip), }; - if !inserted { + insert_advertised_endpoint( + &mut advertised_endpoints, + AdvertisedEndpoint { + node_name: &node.name, + transport: name, + network: None, + host, + port: *port, + }, + )?; + } + } + + // Selector CIDRs and addresses feed clients the same way the + // catch-all advertised address does, so they get the same strict + // parse. Endpoint conflicts are checked across every node's + // selectors and catch-alls: two nodes may reuse one host:port for + // DISJOINT client networks (that is the feature), but wherever + // two networks overlap - equal, nested, or a catch-all, which + // matches everything - a client in the overlap would resolve one + // endpoint to two nodes. Networks are compared truncated + // (`10.0.1.0/16` == `10.0.0.0/16`) and canonicalized + // (`::ffff:10.0.0.0/104` == `10.0.0.0/8`) since matching + // truncates and canonicalizes too. + let mut seen_selector_cidrs = std::collections::HashSet::new(); + for selector in &node.advertised_addresses { + let client_cidr = match selector.client_cidr.parse::<IpNet>() { + Ok(client_cidr) => canonical_ip_net(client_cidr.trunc()), + Err(error) => { eprintln!( - "Invalid cluster configuration: advertised client endpoint conflict - {endpoint} is already used (node '{}', transport {name})", - node.name + "Invalid cluster configuration: advertised_addresses client_cidr '{}' for node '{}': {error}", + selector.client_cidr, node.name + ); + return Err(ConfigurationError::InvalidConfigurationValue); + } + }; + if !seen_selector_cidrs.insert(client_cidr) { + eprintln!( + "Invalid cluster configuration: duplicate advertised_addresses client_cidr '{}' for node '{}'", + selector.client_cidr, node.name + ); + return Err(ConfigurationError::InvalidConfigurationValue); + } + let address = match selector.address.parse::<AdvertisedAddress>() { + Ok(address) => address, + Err(error) => { + eprintln!( + "Invalid cluster configuration: advertised_addresses address '{}' for node '{}': {error}", + selector.address, node.name ); return Err(ConfigurationError::InvalidConfigurationValue); } + }; + for (name, port) in &client_ports { + if let Some(port) = port { + insert_advertised_endpoint( + &mut advertised_endpoints, + AdvertisedEndpoint { + node_name: &node.name, + transport: name, + network: Some(client_cidr), + host: AdvertisedHost::Parsed(address.clone()), + port: *port, + }, + )?; + } } } } @@ -933,6 +1125,84 @@ impl Validatable<ConfigurationError> for ClusterConfig { } } +/// One advertised client endpoint and the client network it serves, pooled by +/// [`ClusterConfig::validate`] so selectors and catch-alls conflict-check +/// against each other. `network: None` is the catch-all (`advertised_address`, +/// or the roster `ip` as fallback): it matches every client, so it overlaps +/// every network. +struct AdvertisedEndpoint<'roster> { + node_name: &'roster str, + transport: &'static str, + network: Option<IpNet>, + host: AdvertisedHost<'roster>, + port: u16, +} + +/// Advertised host in comparable form. `Raw` carries a roster `ip` that +/// failed the strict [`AdvertisedAddress`] parse, so it can never equal a +/// `Parsed` host. +#[derive(PartialEq, Eq)] +enum AdvertisedHost<'roster> { + Parsed(AdvertisedAddress), + Raw(&'roster str), +} + +impl AdvertisedEndpoint<'_> { + /// True when some client would resolve both entries to one host:port on + /// two different nodes. Entries of one node never conflict: longest-prefix + /// match picks exactly one selector per transport, and the bind-endpoint + /// check already forces distinct ports across a node's transports. + fn conflicts_with(&self, other: &Self) -> bool { + if self.node_name == other.node_name || self.port != other.port || self.host != other.host { + return false; + } + match (self.network, other.network) { + (Some(network), Some(other_network)) => { + network.contains(&other_network) || other_network.contains(&network) + } + _ => true, + } + } + + fn authority(&self) -> String { + match &self.host { + AdvertisedHost::Parsed(address) => address.authority(self.port), + AdvertisedHost::Raw(raw_ip) => format!("{raw_ip}:{}", self.port), + } + } + + fn network_description(&self) -> String { + match self.network { + Some(network) => format!("client_cidr {network}"), + None => "every client network (catch-all)".to_owned(), + } + } +} + +fn insert_advertised_endpoint<'roster>( + advertised_endpoints: &mut Vec<AdvertisedEndpoint<'roster>>, + endpoint: AdvertisedEndpoint<'roster>, +) -> Result<(), ConfigurationError> { + if let Some(existing) = advertised_endpoints + .iter() + .find(|existing| existing.conflicts_with(&endpoint)) + { + eprintln!( + "Invalid cluster configuration: advertised client endpoint conflict - {} is advertised for {} (node '{}', transport {}) and for {} (node '{}', transport {}); the networks overlap, so a client in the overlap would resolve both nodes to one endpoint", + endpoint.authority(), + endpoint.network_description(), + endpoint.node_name, + endpoint.transport, + existing.network_description(), + existing.node_name, + existing.transport, + ); + return Err(ConfigurationError::InvalidConfigurationValue); + } + advertised_endpoints.push(endpoint); + Ok(()) +} + #[cfg(test)] mod tests { use super::*; @@ -1075,6 +1345,191 @@ mod advertised_address_tests { } } +#[cfg(test)] +mod advertised_for_tests { + use super::*; + + fn node_with_selectors(selectors: Vec<AdvertisedAddressSelector>) -> ClusterNodeConfig { + ClusterNodeConfig { + name: "node-0".to_owned(), + ip: "10.0.1.5".to_owned(), + advertised_address: Some("203.0.113.10".to_owned()), + advertised_addresses: selectors, + replica_id: 0, + ports: TransportPorts::default(), + } + } + + fn selector(client_cidr: &str, address: &str) -> AdvertisedAddressSelector { + AdvertisedAddressSelector { + client_cidr: client_cidr.to_owned(), + address: address.to_owned(), + } + } + + fn resolved(node: ClusterNodeConfig) -> ResolvedClusterNode { + node.into() + } + + fn ip(address: &str) -> IpAddr { + address.parse().unwrap() + } + + #[test] + fn falls_back_to_advertised_address_without_selectors() { + let node = node_with_selectors(Vec::new()); + assert_eq!( + resolved(node).advertised_for(Some(ip("10.0.0.7"))), + Some(AdvertisedAddress::Ip(ip("203.0.113.10"))) + ); + } + + #[test] + fn falls_back_to_roster_ip_without_advertised_address() { + let mut node = node_with_selectors(Vec::new()); + node.advertised_address = None; + assert_eq!( + resolved(node).advertised_for(Some(ip("10.0.0.7"))), + Some(AdvertisedAddress::Ip(ip("10.0.1.5"))) + ); + } + + #[test] + fn is_none_when_no_fallback_parses() { + let mut node = node_with_selectors(Vec::new()); + node.advertised_address = None; + node.ip = "iggy_node".to_owned(); + assert_eq!(resolved(node).advertised_for(Some(ip("10.0.0.7"))), None); + } + + #[test] + fn unparsable_advertised_address_fails_closed() { + // Only reachable when validation was skipped (cluster disabled). The + // roster ip may be a private address, so leaking it as the catch-all + // would misroute clients; resolution fails closed instead. + let mut node = node_with_selectors(Vec::new()); + node.advertised_address = Some("203.0.113.10:8090".to_owned()); + assert_eq!(resolved(node).advertised_for(Some(ip("10.0.0.7"))), None); + } + + #[test] + fn matching_selector_beats_advertised_address() { + let node = node_with_selectors(vec![selector("10.0.0.0/16", "10.0.1.5")]); + assert_eq!( + resolved(node).advertised_for(Some(ip("10.0.200.7"))), + Some(AdvertisedAddress::Ip(ip("10.0.1.5"))) + ); + } + + #[test] + fn unmatched_client_falls_back_to_advertised_address() { + let node = node_with_selectors(vec![selector("10.0.0.0/16", "10.0.1.5")]); + assert_eq!( + resolved(node).advertised_for(Some(ip("192.168.0.7"))), + Some(AdvertisedAddress::Ip(ip("203.0.113.10"))) + ); + } + + #[test] + fn unknown_client_ip_falls_back_to_advertised_address() { + let node = node_with_selectors(vec![selector("10.0.0.0/16", "10.0.1.5")]); + assert_eq!( + resolved(node).advertised_for(None), + Some(AdvertisedAddress::Ip(ip("203.0.113.10"))) + ); + } + + #[test] + fn longest_prefix_wins_regardless_of_declaration_order() { + let node = resolved(node_with_selectors(vec![ + selector("10.0.0.0/8", "10.255.255.1"), + selector("10.0.0.0/16", "10.0.1.5"), + ])); + assert_eq!( + node.advertised_for(Some(ip("10.0.200.7"))), + Some(AdvertisedAddress::Ip(ip("10.0.1.5"))), + "the /16 must win over the /8 even though it is declared second" + ); + assert_eq!( + node.advertised_for(Some(ip("10.9.0.7"))), + Some(AdvertisedAddress::Ip(ip("10.255.255.1"))), + "a client outside the /16 but inside the /8 must match the /8" + ); + } + + #[test] + fn prefix_length_tie_resolves_to_first_declared() { + // Overlapping same-length networks (host bits differ, same masked + // network) both match; the first declared must win. + let node = node_with_selectors(vec![ + selector("10.0.1.0/16", "10.0.1.5"), + selector("10.0.2.0/16", "10.0.2.5"), + ]); + assert_eq!( + resolved(node).advertised_for(Some(ip("10.0.200.7"))), + Some(AdvertisedAddress::Ip(ip("10.0.1.5"))) + ); + } + + #[test] + fn v4_mapped_v6_client_matches_v4_cidr() { + // A dual-stack listener reports v4 peers as `::ffff:a.b.c.d`. + let node = node_with_selectors(vec![selector("10.0.0.0/16", "10.0.1.5")]); + assert_eq!( + resolved(node).advertised_for(Some(ip("::ffff:10.0.0.7"))), + Some(AdvertisedAddress::Ip(ip("10.0.1.5"))) + ); + } + + #[test] + fn v4_mapped_v6_selector_cidr_matches_v4_client() { + // The mirror case: the CIDR side canonicalizes at build, so + // `::ffff:10.0.0.0/104` matches like `10.0.0.0/8` instead of being + // a silently dead selector. + let node = node_with_selectors(vec![selector("::ffff:10.0.0.0/104", "10.0.1.5")]); + assert_eq!( + resolved(node).advertised_for(Some(ip("10.0.0.7"))), + Some(AdvertisedAddress::Ip(ip("10.0.1.5"))) + ); + } + + #[test] + fn v6_selector_matches_v6_client() { + let node = node_with_selectors(vec![selector("2001:db8::/32", "2001:db8::1")]); + assert_eq!( + resolved(node).advertised_for(Some(ip("2001:db8::7"))), + Some(AdvertisedAddress::Ip(ip("2001:db8::1"))) + ); + } + + #[test] + fn selector_address_may_be_a_hostname() { + let node = node_with_selectors(vec![selector("10.0.0.0/16", "Broker.Internal.Example")]); + assert_eq!( + resolved(node).advertised_for(Some(ip("10.0.0.7"))), + Some(AdvertisedAddress::Hostname( + "broker.internal.example".to_owned() + )) + ); + } + + #[test] + fn unparsable_selectors_are_dropped_at_build_not_per_request() { + // Only reachable when validation was skipped (cluster disabled). The + // bad entries fall out once at build, so LPM runs over the parsable + // rest: the broken /16 no longer shadows the /8. + let node = node_with_selectors(vec![ + selector("not-a-cidr", "10.0.1.5"), + selector("10.0.0.0/16", "iggy_node"), + selector("10.0.0.0/8", "10.255.255.1"), + ]); + assert_eq!( + resolved(node).advertised_for(Some(ip("10.0.0.7"))), + Some(AdvertisedAddress::Ip(ip("10.255.255.1"))) + ); + } +} + #[cfg(test)] mod cluster_validate_tests { use super::*; @@ -1084,11 +1539,19 @@ mod cluster_validate_tests { name: name.to_string(), ip: "127.0.0.1".to_string(), advertised_address: None, + advertised_addresses: Vec::new(), replica_id: id, ports: TransportPorts::default(), } } + fn selector(client_cidr: &str, address: &str) -> AdvertisedAddressSelector { + AdvertisedAddressSelector { + client_cidr: client_cidr.to_owned(), + address: address.to_owned(), + } + } + fn cfg(nodes: Vec<ClusterNodeConfig>) -> ClusterConfig { ClusterConfig { enabled: true, @@ -1506,6 +1969,167 @@ mod cluster_validate_tests { assert!(cfg(vec![n1, n2]).validate().is_ok()); } + #[test] + fn validate_accepts_selectors_with_distinct_cidrs() { + let mut n1 = node("n1", 0); + n1.ip = "10.0.0.1".to_owned(); + n1.advertised_address = Some("203.0.113.1".to_owned()); + n1.advertised_addresses = vec![ + selector("10.0.0.0/16", "10.0.0.1"), + selector("10.0.0.0/8", "broker-1.internal.example"), + ]; + n1.ports.tcp = Some(8090); + + assert!(cfg(vec![n1, node("n2", 1)]).validate().is_ok()); + } + + #[test] + fn validate_rejects_malformed_selector_cidr() { + for client_cidr in ["10.0.0.0", "10.0.0.0/33", "not-a-cidr", ""] { + let mut n1 = node("n1", 0); + n1.advertised_addresses = vec![selector(client_cidr, "10.0.0.1")]; + + assert!( + cfg(vec![n1, node("n2", 1)]).validate().is_err(), + "client_cidr '{client_cidr}' must be rejected" + ); + } + } + + #[test] + fn validate_rejects_malformed_selector_address() { + for address in ["", "10.0.0.1:8090", "10.0.0.256", "iggy_node"] { + let mut n1 = node("n1", 0); + n1.advertised_addresses = vec![selector("10.0.0.0/16", address)]; + + assert!( + cfg(vec![n1, node("n2", 1)]).validate().is_err(), + "selector address '{address}' must be rejected" + ); + } + } + + #[test] + fn validate_rejects_duplicate_selector_cidr_within_a_node() { + // `10.0.1.0/16` truncates to `10.0.0.0/16`: the two selectors match + // the identical client set, so the second could never win LPM. + let mut n1 = node("n1", 0); + n1.advertised_addresses = vec![ + selector("10.0.0.0/16", "10.0.0.1"), + selector("10.0.1.0/16", "10.0.0.2"), + ]; + + assert!(cfg(vec![n1, node("n2", 1)]).validate().is_err()); + } + + #[test] + fn validate_rejects_selector_endpoint_conflict_within_one_cidr() { + let mut n1 = node("n1", 0); + n1.ip = "10.0.0.1".to_owned(); + n1.advertised_addresses = vec![selector("10.0.0.0/16", "10.0.7.7")]; + n1.ports.tcp = Some(8090); + let mut n2 = node("n2", 1); + n2.ip = "10.0.0.2".to_owned(); + n2.advertised_addresses = vec![selector("10.0.0.0/16", "10.0.7.7")]; + n2.ports.tcp = Some(8090); + + assert!(cfg(vec![n1, n2]).validate().is_err()); + } + + #[test] + fn validate_accepts_identical_selector_endpoint_across_different_cidrs() { + // Reusing one host:port across DIFFERENT client networks is the + // feature (e.g. each network NATs the address to its local node). + let mut n1 = node("n1", 0); + n1.ip = "10.0.0.1".to_owned(); + n1.advertised_addresses = vec![selector("10.1.0.0/16", "192.0.2.10")]; + n1.ports.tcp = Some(8090); + let mut n2 = node("n2", 1); + n2.ip = "10.0.0.2".to_owned(); + n2.advertised_addresses = vec![selector("10.2.0.0/16", "192.0.2.10")]; + n2.ports.tcp = Some(8090); + + assert!(cfg(vec![n1, n2]).validate().is_ok()); + } + + #[test] + fn validate_rejects_v4_mapped_v6_selector_cidr_duplicating_its_v4_form() { + // `::ffff:10.0.0.0/104` canonicalizes to `10.0.0.0/8` (matching how + // client IPs canonicalize before LPM), so these two selectors match + // the identical client set. + let mut n1 = node("n1", 0); + n1.advertised_addresses = vec![ + selector("10.0.0.0/8", "10.0.0.1"), + selector("::ffff:10.0.0.0/104", "10.0.0.2"), + ]; + + assert!(cfg(vec![n1, node("n2", 1)]).validate().is_err()); + } + + #[test] + fn validate_rejects_selector_endpoint_clashing_with_another_nodes_catch_all() { + // The catch-all matches every client, so a 10.0.0.0/16 client would + // resolve both nodes to 192.0.2.10:8090. + let mut n1 = node("n1", 0); + n1.ip = "10.0.0.1".to_owned(); + n1.advertised_addresses = vec![selector("10.0.0.0/16", "192.0.2.10")]; + n1.ports.tcp = Some(8090); + let mut n2 = node("n2", 1); + n2.ip = "10.0.0.2".to_owned(); + n2.advertised_address = Some("192.0.2.10".to_owned()); + n2.ports.tcp = Some(8090); + + assert!(cfg(vec![n1, n2]).validate().is_err()); + } + + #[test] + fn validate_rejects_selector_endpoint_clashing_with_another_nodes_roster_ip() { + // Without an advertised_address the roster ip backs the catch-all, + // so the same cross-set conflict applies to it. + let mut n1 = node("n1", 0); + n1.ip = "10.0.0.1".to_owned(); + n1.advertised_addresses = vec![selector("10.0.0.0/16", "10.0.0.2")]; + n1.ports.tcp = Some(8090); + let mut n2 = node("n2", 1); + n2.ip = "10.0.0.2".to_owned(); + n2.ports.tcp = Some(8090); + + assert!(cfg(vec![n1, n2]).validate().is_err()); + } + + #[test] + fn validate_rejects_identical_selector_endpoint_across_nested_cidrs() { + // LPM runs per node, not cluster-wide: a 10.0.0.0/16 client matches + // n1 through /8 and n2 through /16, resolving both to 192.0.2.10:8090. + // Nested networks conflict just like equal ones. + let mut n1 = node("n1", 0); + n1.ip = "10.0.0.1".to_owned(); + n1.advertised_addresses = vec![selector("10.0.0.0/8", "192.0.2.10")]; + n1.ports.tcp = Some(8090); + let mut n2 = node("n2", 1); + n2.ip = "10.0.0.2".to_owned(); + n2.advertised_addresses = vec![selector("10.0.0.0/16", "192.0.2.10")]; + n2.ports.tcp = Some(8090); + + assert!(cfg(vec![n1, n2]).validate().is_err()); + } + + #[test] + fn validate_accepts_selector_reusing_its_own_nodes_catch_all_address() { + // Redundant but harmless: within one node the selector and the + // catch-all cannot resolve a client to two different nodes. + let mut n1 = node("n1", 0); + n1.ip = "10.0.0.1".to_owned(); + n1.advertised_address = Some("192.0.2.10".to_owned()); + n1.advertised_addresses = vec![selector("10.0.0.0/16", "192.0.2.10")]; + n1.ports.tcp = Some(8090); + let mut n2 = node("n2", 1); + n2.ip = "10.0.0.2".to_owned(); + n2.ports.tcp = Some(8091); + + assert!(cfg(vec![n1, n2]).validate().is_ok()); + } + #[test] fn validate_rejects_zero_tcp_replica_port() { let ports = TransportPorts { diff --git a/core/configs/src/server_ng_config/defaults.rs b/core/configs/src/server_ng_config/defaults.rs index a3894aeef..e68f0889c 100644 --- a/core/configs/src/server_ng_config/defaults.rs +++ b/core/configs/src/server_ng_config/defaults.rs @@ -119,6 +119,7 @@ impl Default for ClusterConfig { name: node.name.parse().unwrap(), ip: node.ip.parse().unwrap(), advertised_address: None, + advertised_addresses: Vec::new(), replica_id: u8::try_from(node.replica_id).expect( "static_toml replica_id must fit in u8 (0..=255); \ fix core/server-ng/config.toml", diff --git a/core/configs/src/server_ng_config/validators.rs b/core/configs/src/server_ng_config/validators.rs index c00d55df8..8c6941f5e 100644 --- a/core/configs/src/server_ng_config/validators.rs +++ b/core/configs/src/server_ng_config/validators.rs @@ -745,6 +745,7 @@ mod tests { name: format!("node-{replica_id}"), ip: "127.0.0.1".to_string(), advertised_address: None, + advertised_addresses: Vec::new(), replica_id, ports: TransportPorts { tcp: Some(8090 + u16::from(replica_id)), diff --git a/core/integration/tests/server/cluster_metadata_vsr.rs b/core/integration/tests/server/cluster_metadata_vsr.rs index f15ad4416..ca50564a0 100644 --- a/core/integration/tests/server/cluster_metadata_vsr.rs +++ b/core/integration/tests/server/cluster_metadata_vsr.rs @@ -17,10 +17,21 @@ //! Binary `GetClusterMetadata` over a real VSR cluster. The reply must carry //! the full configured roster with real client ports and live leader/follower -//! roles - not a single synthesized self node with zeroed ports. +//! roles - not a single synthesized self node with zeroed ports. The +//! selector suite below additionally proves per-client-network advertised +//! addresses (`cluster.nodes.advertised_addresses`) end to end: a loopback +//! client must see the selector address in binary metadata, HTTP metadata, +//! and the 307 leader-redirect `Location`. + +use std::time::Instant; use iggy::prelude::*; +use integration::harness::TestHarness; use integration::iggy_harness; +use reqwest::StatusCode; +use tokio::time::sleep; + +use crate::server::http_client::{HttpClient, LOGIN_RETRY_INTERVAL, LOGIN_TIMEOUT}; /// One shard per node so the client connection is always served by shard 0, /// where the metadata consensus that marks the leader lives. A request @@ -74,3 +85,182 @@ async fn given_two_node_cluster_when_getting_cluster_metadata_should_return_full "every other node must follow, got {metadata}" ); } + +const SELECTOR_CLUSTER_NODES: usize = 2; + +/// Every test client in this file connects over loopback, so this CIDR is the +/// one the server must match its peer address against. +const LOOPBACK_CIDR: &str = "127.0.0.0/8"; + +/// Selector marker: distinct from the harness roster ip (`127.0.0.1`), so it +/// appears in metadata or a redirect only when the selector path ran - yet it +/// still resolves, which matters because the leader-aware SDK client redials +/// whatever address metadata advertises. +const SELECTOR_ADDRESS: &str = "localhost"; + +/// A 2-node cluster whose roster gives every node a loopback-CIDR selector. +/// The roster (selectors included) must be identical on every node, so each +/// server process gets the env vars for all nodes. +fn selector_cluster() -> TestHarness { + let mut harness = TestHarness::builder() + .default_server() + .cluster_nodes(SELECTOR_CLUSTER_NODES) + .build() + .expect("build selector cluster harness"); + for node in 0..SELECTOR_CLUSTER_NODES { + for roster_entry in 0..SELECTOR_CLUSTER_NODES { + harness.node_mut(node).add_env( + format!("IGGY_CLUSTER_NODES_{roster_entry}_ADVERTISED_ADDRESSES_0_CLIENT_CIDR"), + LOOPBACK_CIDR, + ); + harness.node_mut(node).add_env( + format!("IGGY_CLUSTER_NODES_{roster_entry}_ADVERTISED_ADDRESSES_0_ADDRESS"), + SELECTOR_ADDRESS, + ); + } + } + harness +} + +/// Assert every roster node reports the selector address to this loopback +/// client; without the selector the same roster publishes `127.0.0.1`. +fn assert_selector_addresses(nodes: impl IntoIterator<Item = (String, String)>) { + let mut seen = 0; + for (name, ip) in nodes { + assert_eq!( + ip, SELECTOR_ADDRESS, + "node '{name}' must advertise its selector address to a loopback client" + ); + seen += 1; + } + assert_eq!(seen, SELECTOR_CLUSTER_NODES, "full roster must be reported"); +} + +#[tokio::test] +#[serial_test::parallel] +async fn given_matching_client_cidr_when_getting_binary_cluster_metadata_should_return_selector_addresses() + { + let mut harness = selector_cluster(); + harness.start().await.expect("start selector cluster"); + + let client = harness + .node(0) + .tcp_client() + .expect("tcp client") + .with_root_login() + .connect() + .await + .expect("connect to node 0"); + let metadata = client + .get_cluster_metadata() + .await + .expect("get cluster metadata"); + + assert_selector_addresses( + metadata + .nodes + .iter() + .map(|node| (node.name.clone(), node.ip.clone())), + ); +} + +#[tokio::test] +#[serial_test::parallel] +async fn given_matching_client_cidr_when_getting_http_cluster_metadata_should_return_selector_addresses() + { + let mut harness = selector_cluster(); + harness.start().await.expect("start selector cluster"); + + let http = HttpClient::login_root(&harness).await; + let response = http.get("/cluster/metadata").await; + assert_eq!(response.status(), StatusCode::OK); + let metadata: serde_json::Value = response.json().await.expect("decode cluster metadata"); + + let nodes = metadata["nodes"] + .as_array() + .expect("metadata must carry a nodes array") + .iter() + .map(|node| { + ( + node["name"].as_str().expect("node name").to_owned(), + node["ip"].as_str().expect("node ip").to_owned(), + ) + }) + .collect::<Vec<_>>(); + assert_selector_addresses(nodes); +} + +/// A linearizable read on the follower must 307 to the primary, and the +/// `Location` host must be the primary's SELECTOR address: the redirected +/// client is the same loopback peer, so pointing it at the catch-all (or the +/// roster ip) would route it off its network. +#[tokio::test] +#[serial_test::parallel] +async fn given_matching_client_cidr_when_redirected_to_primary_should_target_selector_address() { + const READ_PATH: &str = "/streams?consistency=linearizable"; + + let mut harness = selector_cluster(); + harness.start().await.expect("start selector cluster"); + + // Redirects must surface, not be followed: the `Location` itself is what + // is under test, and the selector hostname resolves by design (see + // SELECTOR_ADDRESS), so a redirect-following client would chase it to the + // primary, get a 200, and hide a wrong Location host. Bearers are + // node-local in this keyless cluster, so each node gets its own login. + let deadline = Instant::now() + LOGIN_TIMEOUT; + let mut verdicts: Vec<(usize, StatusCode, Option<String>)>; + loop { + verdicts = Vec::new(); + for node in 0..SELECTOR_CLUSTER_NODES { + let addr = harness.node(node).http_addr().expect("node http address"); + let session = HttpClient::login_root_no_redirect(format!("http://{addr}";)).await; + let response = session.get(READ_PATH).await; + let location = response + .headers() + .get(reqwest::header::LOCATION) + .map(|value| value.to_str().expect("Location header").to_owned()); + verdicts.push((node, response.status(), location)); + } + // Settled view: exactly one primary serves the read locally and every + // other node redirects. Anything else (a 503 follower that cannot + // resolve the primary yet, a double-307 round mid-election) means the + // view is still settling; retry within the shared warmup budget. + let primary_count = verdicts + .iter() + .filter(|(_, status, _)| *status == StatusCode::OK) + .count(); + let redirect_count = verdicts + .iter() + .filter(|(_, status, _)| *status == StatusCode::TEMPORARY_REDIRECT) + .count(); + if primary_count == 1 && redirect_count == SELECTOR_CLUSTER_NODES - 1 { + break; + } + assert!( + Instant::now() < deadline, + "cluster did not settle on one primary and redirecting followers within {LOGIN_TIMEOUT:?}: {verdicts:?}" + ); + sleep(LOGIN_RETRY_INTERVAL).await; + } + + // The loop only breaks on one 200 with every other node at 307, so the + // non-redirected partition is exactly the primary. + let (redirected, primaries): (Vec<_>, Vec<_>) = verdicts + .iter() + .partition(|(_, status, _)| *status == StatusCode::TEMPORARY_REDIRECT); + let (primary_index, _, _) = primaries[0]; + + let primary_http_port = harness + .node(*primary_index) + .http_addr() + .expect("primary http address") + .port(); + let expected_location = format!("http://{SELECTOR_ADDRESS}:{primary_http_port}{READ_PATH}";); + for (follower, _, location) in redirected { + assert_eq!( + location.as_deref(), + Some(expected_location.as_str()), + "follower node {follower} must redirect this loopback client to the primary's selector address" + ); + } +} diff --git a/core/integration/tests/server/http_client.rs b/core/integration/tests/server/http_client.rs index 7d9558494..e254fbfd7 100644 --- a/core/integration/tests/server/http_client.rs +++ b/core/integration/tests/server/http_client.rs @@ -53,12 +53,29 @@ impl HttpClient { .server() .http_addr() .expect("HTTP transport not configured on test server"); - let base_url = format!("http://{addr}";); let client = reqwest::Client::builder() .timeout(REQUEST_TIMEOUT) .build() .expect("build reqwest client"); + Self::login_root_with(client, format!("http://{addr}";)).await + } + + /// Log in as root against an explicit listener base URL, with redirects + /// surfaced instead of followed - for suites that assert on `Location` + /// (reqwest follows a 307 transparently by default). The explicit URL also + /// reaches a follower node, which [`Self::login_root`] (pinned to node 0) + /// cannot. + pub async fn login_root_no_redirect(base_url: String) -> Self { + let client = reqwest::Client::builder() + .timeout(REQUEST_TIMEOUT) + .redirect(reqwest::redirect::Policy::none()) + .build() + .expect("build reqwest client"); + Self::login_root_with(client, base_url).await + } + /// Shared root-login retry loop over an arbitrary client + base URL. + async fn login_root_with(client: reqwest::Client, base_url: String) -> Self { let body = json!({ "username": DEFAULT_ROOT_USERNAME, "password": DEFAULT_ROOT_PASSWORD, diff --git a/core/server-ng/config.toml b/core/server-ng/config.toml index af6d3c9ce..29d716576 100644 --- a/core/server-ng/config.toml +++ b/core/server-ng/config.toml @@ -739,6 +739,29 @@ ca_file = "" # (RFC 1123: ASCII letters, digits, '-' and '.'; no port, no trailing dot). # When unset, clients receive 'ip'. # +# When different client networks need different addresses (a public +# 'advertised_address' would route in-VPC clients out through the public +# side), add per-network 'advertised_addresses' selectors: clients whose +# peer IP falls inside 'client_cidr' are handed 'address' instead of the +# catch-all. The longest matching prefix wins; a tie on prefix length goes +# to the first declared selector; clients matching no selector fall back to +# 'advertised_address', then 'ip'. Matching sees the transport-level peer +# address, so clients behind a proxy or load balancer match the proxy's +# network, not their own. Note for rolling upgrades: older server binaries +# reject configs containing 'advertised_addresses' - upgrade every binary +# first, then add selectors. +# +# [[cluster.nodes]] +# name = "iggy-node-1" +# ip = "10.0.1.5" # replica plane + last-resort fallback +# advertised_address = "203.0.113.10" # catch-all for unmatched clients +# replica_id = 0 +# ports = { tcp = 8090, http = 3000, tcp_replica = 9090 } +# +# [[cluster.nodes.advertised_addresses]] +# client_cidr = "10.0.0.0/16" # in-VPC clients stay private +# address = "10.0.1.5" +# # In cluster mode, 'ports' is the single source of listener ports: every # enabled transport needs an explicit per-node port, otherwise the server # refuses to start. diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs index c1e196a1b..9fddb3886 100644 --- a/core/server-ng/src/bootstrap.rs +++ b/core/server-ng/src/bootstrap.rs @@ -1596,7 +1596,13 @@ fn build_cluster_roster( ClusterRoster { enabled: config.cluster.enabled, name: config.cluster.name.clone(), - nodes: config.cluster.nodes.clone(), + nodes: config + .cluster + .nodes + .iter() + .cloned() + .map(Into::into) + .collect(), self_ip: topology.client_listen_addr.ip().to_string(), self_ports: configs::ng_cluster::TransportPorts { tcp: Some(topology.client_listen_addr.port()), @@ -3976,6 +3982,7 @@ mod tests { name: "node".to_owned(), ip: ip.to_owned(), advertised_address: None, + advertised_addresses: Vec::new(), replica_id: 0, ports: configs::ng_cluster::TransportPorts { tcp, diff --git a/core/server-ng/src/cluster_meta.rs b/core/server-ng/src/cluster_meta.rs index 8a1bff535..aac36afe2 100644 --- a/core/server-ng/src/cluster_meta.rs +++ b/core/server-ng/src/cluster_meta.rs @@ -28,10 +28,11 @@ //! leader, but the full roster is still returned). The self-synthesized single //! node is the cluster-disabled fallback, shared by both callers. -use configs::ng_cluster::{AdvertisedAddress, ClusterNodeConfig, TransportPorts}; +use configs::ng_cluster::{ResolvedClusterNode, TransportPorts}; use iggy_common::{ ClusterMetadata, ClusterNode, ClusterNodeRole, ClusterNodeStatus, TransportEndpoints, }; +use std::net::IpAddr; use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; @@ -50,7 +51,9 @@ const SINGLE_NODE_CLUSTER_NAME: &str = "single-node"; pub struct ClusterRoster { pub enabled: bool, pub name: String, - pub nodes: Vec<ClusterNodeConfig>, + /// Roster nodes with selectors parsed once at roster build, so the + /// per-request address resolution never re-parses config strings. + pub nodes: Vec<ResolvedClusterNode>, /// This node's own address, reported for the synthesized self node. pub self_ip: String, /// This node's own client ports for the same self node (`None` = transport @@ -98,17 +101,24 @@ impl ClusterRoster { /// emit one node per entry, marking the node at `primary_index` the leader /// and the rest followers; `None` (no on-shard consensus) leaves every node /// a follower. Otherwise synthesize the single self node as the sole - /// leader. - pub fn cluster_metadata(&self, primary_index: Option<u8>) -> ClusterMetadata { + /// leader. `client_ip` is the requesting client's transport-level peer + /// address, used to pick each node's advertised address from its + /// per-client-network selectors; `None` (unknown peer) serves the + /// catch-all address. + pub fn cluster_metadata( + &self, + primary_index: Option<u8>, + client_ip: Option<IpAddr>, + ) -> ClusterMetadata { if self.enabled && !self.nodes.is_empty() { let nodes = self .nodes .iter() .map(|node| ClusterNode { - name: node.name.clone(), - ip: client_host(node), - endpoints: ports_to_endpoints(&node.ports), - role: role_for(primary_index, node.replica_id), + name: node.config.name.clone(), + ip: client_host(node, client_ip), + endpoints: ports_to_endpoints(&node.config.ports), + role: role_for(primary_index, node.config.replica_id), status: ClusterNodeStatus::Healthy, }) .collect(); @@ -137,14 +147,23 @@ impl ClusterRoster { /// Client-facing host in normalized form (lowercase hostname, canonical IP), /// matching what boot validation compared and what redirect URLs render, so -/// textual config variants of one address publish identical metadata. A -/// roster `ip` that parses as neither (boot only requires it non-empty) -/// passes through verbatim; a configured `advertised_address` always parses, -/// validation rejects it otherwise. -fn client_host(node: &ClusterNodeConfig) -> String { - let host = node.advertised_address.as_deref().unwrap_or(&node.ip); - host.parse::<AdvertisedAddress>() - .map_or_else(|_| host.to_owned(), |address| address.to_string()) +/// textual config variants of one address publish identical metadata. The +/// per-client-network selectors, the catch-all `advertised_address`, and the +/// roster `ip` are consulted in that order ([`ResolvedClusterNode::advertised_for`]); +/// a host that parses as neither IP nor hostname (only reachable when boot +/// validation was skipped, or for the roster `ip`, which boot only requires +/// non-empty) passes through verbatim. +fn client_host(node: &ResolvedClusterNode, client_ip: Option<IpAddr>) -> String { + node.advertised_for(client_ip).map_or_else( + || { + node.config + .advertised_address + .as_deref() + .unwrap_or(&node.config.ip) + .to_owned() + }, + |address| address.to_string(), + ) } const fn role_for(primary_index: Option<u8>, replica_id: u8) -> ClusterNodeRole { @@ -167,40 +186,52 @@ fn ports_to_endpoints(ports: &TransportPorts) -> TransportEndpoints { mod tests { use super::*; - fn roster(advertised_address: Option<String>) -> ClusterRoster { + use configs::ng_cluster::{AdvertisedAddressSelector, ClusterNodeConfig}; + + fn node_config(advertised_address: Option<String>) -> ClusterNodeConfig { + ClusterNodeConfig { + name: "node-0".to_owned(), + ip: "10.0.0.1".to_owned(), + advertised_address, + advertised_addresses: Vec::new(), + replica_id: 0, + ports: TransportPorts::default(), + } + } + + fn roster_of(node: ClusterNodeConfig) -> ClusterRoster { ClusterRoster { enabled: true, name: "test-cluster".to_owned(), - nodes: vec![ClusterNodeConfig { - name: "node-0".to_owned(), - ip: "10.0.0.1".to_owned(), - advertised_address, - replica_id: 0, - ports: TransportPorts::default(), - }], + nodes: vec![node.into()], self_ip: "127.0.0.1".to_owned(), self_ports: TransportPorts::default(), metadata_view: Arc::new(AtomicU64::new(METADATA_VIEW_UNKNOWN)), } } + fn roster(advertised_address: Option<String>) -> ClusterRoster { + roster_of(node_config(advertised_address)) + } + #[test] fn cluster_metadata_uses_advertised_address_when_configured() { - let metadata = roster(Some("203.0.113.10".to_owned())).cluster_metadata(Some(0)); + let metadata = roster(Some("203.0.113.10".to_owned())).cluster_metadata(Some(0), None); assert_eq!(metadata.nodes[0].ip, "203.0.113.10"); } #[test] fn cluster_metadata_falls_back_to_replica_ip() { - let metadata = roster(None).cluster_metadata(Some(0)); + let metadata = roster(None).cluster_metadata(Some(0), None); assert_eq!(metadata.nodes[0].ip, "10.0.0.1"); } #[test] fn cluster_metadata_normalizes_advertised_hostname_to_lowercase() { - let metadata = roster(Some("Broker.Example.COM".to_owned())).cluster_metadata(Some(0)); + let metadata = + roster(Some("Broker.Example.COM".to_owned())).cluster_metadata(Some(0), None); assert_eq!(metadata.nodes[0].ip, "broker.example.com"); } @@ -208,7 +239,8 @@ mod tests { #[test] fn cluster_metadata_canonicalizes_advertised_ipv6_address() { for equivalent_address in ["2001:DB8::1", "[2001:db8::1]"] { - let metadata = roster(Some(equivalent_address.to_owned())).cluster_metadata(Some(0)); + let metadata = + roster(Some(equivalent_address.to_owned())).cluster_metadata(Some(0), None); assert_eq!( metadata.nodes[0].ip, "2001:db8::1", @@ -219,11 +251,29 @@ mod tests { #[test] fn cluster_metadata_passes_unparsable_replica_ip_verbatim() { - let mut cluster_roster = roster(None); - cluster_roster.nodes[0].ip = "iggy_node".to_owned(); + let mut node = node_config(None); + node.ip = "iggy_node".to_owned(); - let metadata = cluster_roster.cluster_metadata(Some(0)); + let metadata = roster_of(node).cluster_metadata(Some(0), None); assert_eq!(metadata.nodes[0].ip, "iggy_node"); } + + #[test] + fn cluster_metadata_serves_the_selector_address_to_a_matching_client() { + let mut node = node_config(Some("203.0.113.10".to_owned())); + node.advertised_addresses = vec![AdvertisedAddressSelector { + client_cidr: "10.0.0.0/16".to_owned(), + address: "10.0.0.1".to_owned(), + }]; + let cluster_roster = roster_of(node); + + let in_network = + cluster_roster.cluster_metadata(Some(0), Some("10.0.9.9".parse().unwrap())); + assert_eq!(in_network.nodes[0].ip, "10.0.0.1"); + + let out_of_network = + cluster_roster.cluster_metadata(Some(0), Some("198.51.100.7".parse().unwrap())); + assert_eq!(out_of_network.nodes[0].ip, "203.0.113.10"); + } } diff --git a/core/server-ng/src/dispatch.rs b/core/server-ng/src/dispatch.rs index 99908fd6e..26a4d68fc 100644 --- a/core/server-ng/src/dispatch.rs +++ b/core/server-ng/src/dispatch.rs @@ -105,6 +105,7 @@ use shard::{ }; use std::cell::RefCell; use std::collections::{HashMap, HashSet, VecDeque}; +use std::net::IpAddr; use std::rc::Rc; use std::sync::Arc; use tracing::{debug, warn}; @@ -1353,7 +1354,15 @@ async fn handle_non_replicated_request<B, MJ, S>( handle_sync_consumer_group(shard, transport_client_id, &request).await; } _ => { - let roster = sessions.borrow().cluster_roster(); + let (roster, client_ip) = { + let sessions = sessions.borrow(); + ( + sessions.cluster_roster(), + sessions + .connection_address(transport_client_id) + .map(|address| address.ip()), + ) + }; handle_default_non_replicated( shard, transport_client_id, @@ -1361,13 +1370,14 @@ async fn handle_non_replicated_request<B, MJ, S>( &request, user_id, &roster, + client_ip, ) .await; } } } -#[allow(clippy::future_not_send)] +#[allow(clippy::future_not_send, clippy::too_many_arguments)] async fn handle_default_non_replicated<B, MJ, S>( shard: &Rc<ShellShard<B, MJ, S>>, transport_client_id: u128, @@ -1375,6 +1385,7 @@ async fn handle_default_non_replicated<B, MJ, S>( request: &Message<RequestHeader>, user_id: Option<u32>, roster: &ClusterRoster, + client_ip: Option<IpAddr>, ) where B: ShellBus, MJ: JournalHandle + 'static, @@ -1388,7 +1399,14 @@ async fn handle_default_non_replicated<B, MJ, S>( send_non_replicated_deny(shard, request, transport_client_id, error.as_code()).await; return; } - match build_non_replicated_response(shard, code, request_body(request), user_id, roster) { + match build_non_replicated_response( + shard, + code, + request_body(request), + user_id, + roster, + client_ip, + ) { Ok(response) => { let commit = current_metadata_commit(shard); let reply = response.into_reply( diff --git a/core/server-ng/src/http.rs b/core/server-ng/src/http.rs index 23946535c..e36f1b674 100644 --- a/core/server-ng/src/http.rs +++ b/core/server-ng/src/http.rs @@ -48,11 +48,13 @@ use std::sync::Arc; use std::sync::atomic::AtomicU64; use axum::Router; +use axum::extract::connect_info::Connected; use axum::extract::{DefaultBodyLimit, Request}; use axum::http::{HeaderName, HeaderValue, Method, StatusCode, Version, header::CONNECTION}; use axum::middleware::{Next, from_fn, from_fn_with_state}; use axum::response::Response; use axum::routing::{delete, get, post, put}; +use compio::net::TcpListener; use configs::http::{HttpConfig, HttpCorsConfig}; use configs::ng_cluster::{ClusterConfig, TransportPorts, http_forwarding_key_material}; use configs::server_ng::NgSystemConfig; @@ -146,7 +148,7 @@ pub async fn start( roster: ClusterRoster { enabled: cluster.enabled, name: cluster.name.clone(), - nodes: cluster.nodes.clone(), + nodes: cluster.nodes.iter().cloned().map(Into::into).collect(), self_ip: bound_addr.ip().to_string(), // The self node reports the live bound HTTP port; the other client // ports arrive resolved from the caller. @@ -189,9 +191,12 @@ pub async fn start( info!(address = %bound_addr, "server-ng HTTP listener started"); let shutdown = shard.bus.token(); let handle = compio::runtime::spawn(async move { - if let Err(error) = cyper_axum::serve(listener, router) - .with_graceful_shutdown(async move { shutdown.wait().await }) - .await + if let Err(error) = cyper_axum::serve( + listener, + router.into_make_service_with_connect_info::<ClientAddr>(), + ) + .with_graceful_shutdown(async move { shutdown.wait().await }) + .await { error!(%error, "server-ng HTTP listener terminated with error"); } @@ -202,6 +207,23 @@ pub async fn start( Ok(()) } +/// Connect-info payload carrying the peer socket address of an HTTP client. +/// +/// The plain listener records it through axum's connect-info make-service +/// (the [`Connected`] impl below); the TLS path cannot (its hand-rolled serve +/// loop bypasses the make-service), so `tls::serve` injects the identical +/// `ConnectInfo<ClientAddr>` extension per connection instead. Handlers read +/// it through the `Identity` extractor's `client_ip`, which picks the +/// advertised address a client is told about - never authorization. +#[derive(Debug, Clone, Copy)] +pub struct ClientAddr(pub SocketAddr); + +impl Connected<cyper_axum::IncomingStream<'_, TcpListener>> for ClientAddr { + fn connect_info(stream: cyper_axum::IncomingStream<'_, TcpListener>) -> Self { + Self(*stream.remote_addr()) + } +} + /// Health-probe path. Public and pre-auth, and the one success route reached /// without proving a credential, so the `Iggy-View` layer withholds the /// cluster-internal view number here (see the response layer below). diff --git a/core/server-ng/src/http/error.rs b/core/server-ng/src/http/error.rs index 28f5f56c2..33e14a6ec 100644 --- a/core/server-ng/src/http/error.rs +++ b/core/server-ng/src/http/error.rs @@ -25,7 +25,7 @@ use axum::Json; use axum::http::header::{LOCATION, RETRY_AFTER}; use axum::http::{HeaderValue, StatusCode}; use axum::response::{IntoResponse, Response}; -use configs::ng_cluster::{AdvertisedAddress, ClusterNodeConfig}; +use configs::ng_cluster::ResolvedClusterNode; use iggy_binary_protocol::Operation; use iggy_common::IggyError; use serde::{Deserialize, Serialize}; @@ -525,17 +525,20 @@ fn primary_redirect_response(location: &str) -> Response { /// Build the `Location` for a 307 redirect of a linearizable read to the VSR /// primary: `<scheme>://<host>:<http-port><path_and_query>`. The scheme is the /// redirecting node's own listener scheme (uniform cluster HTTP config, same -/// assumption the forward hop makes). `None` when the primary does not resolve -/// from the roster, so the caller fails closed to a 503 rather than pointing at -/// an unreachable target. Pure (no consensus or axum dependency) so the -/// redirect target is unit-tested in isolation. +/// assumption the forward hop makes). `client_ip` is the redirected client's +/// peer address, so the `Location` host comes from the primary's +/// per-client-network selectors when one matches. `None` when the primary does +/// not resolve from the roster, so the caller fails closed to a 503 rather +/// than pointing at an unreachable target. Pure (no consensus or axum +/// dependency) so the redirect target is unit-tested in isolation. pub(in crate::http) fn primary_redirect_location( roster: &ClusterRoster, primary_index: u8, scheme: &str, path_and_query: &str, + client_ip: Option<IpAddr>, ) -> Option<String> { - let authority = primary_advertised_http_authority(roster, primary_index)?; + let authority = primary_advertised_http_authority(roster, primary_index, client_ip)?; Some(format!("{scheme}://{authority}{path_and_query}")) } @@ -547,29 +550,33 @@ pub(in crate::http) fn primary_http_socket( primary_index: u8, ) -> Option<SocketAddr> { let (node, http_port) = primary_node(roster, primary_index)?; - let ip = node.ip.parse::<IpAddr>().ok()?; + let ip = node.config.ip.parse::<IpAddr>().ok()?; Some(SocketAddr::new(ip, http_port)) } -/// Resolve the client-facing HTTP authority (`host:port`) for a redirect. The -/// advertised address is preferred, with the private roster IP retained as -/// the compatibility fallback. [`AdvertisedAddress::authority`] brackets IPv6 -/// hosts and passes hostnames through, so the redirect URL stays valid; a -/// host that is neither a valid IP nor a valid hostname yields `None` so -/// callers fail closed. -fn primary_advertised_http_authority(roster: &ClusterRoster, primary_index: u8) -> Option<String> { +/// Resolve the client-facing HTTP authority (`host:port`) for a redirect +/// through [`ResolvedClusterNode::advertised_for`]: a client-network selector +/// match first, then the catch-all advertised address, then the private +/// roster IP as the compatibility fallback. `AdvertisedAddress::authority` +/// brackets IPv6 hosts and passes hostnames through, so the redirect URL +/// stays valid; a host that is neither a valid IP nor a valid hostname yields +/// `None` so callers fail closed. +fn primary_advertised_http_authority( + roster: &ClusterRoster, + primary_index: u8, + client_ip: Option<IpAddr>, +) -> Option<String> { let (node, http_port) = primary_node(roster, primary_index)?; - let host = node.advertised_address.as_deref().unwrap_or(&node.ip); - let address = host.parse::<AdvertisedAddress>().ok()?; + let address = node.advertised_for(client_ip)?; Some(address.authority(http_port)) } -fn primary_node(roster: &ClusterRoster, primary_index: u8) -> Option<(&ClusterNodeConfig, u16)> { +fn primary_node(roster: &ClusterRoster, primary_index: u8) -> Option<(&ResolvedClusterNode, u16)> { let node = roster .nodes .iter() - .find(|node| node.replica_id == primary_index)?; - let http_port = node.ports.http?; + .find(|node| node.config.replica_id == primary_index)?; + let http_port = node.config.ports.http?; Some((node, http_port)) } @@ -577,7 +584,7 @@ fn primary_node(roster: &ClusterRoster, primary_index: u8) -> Option<(&ClusterNo mod tests { use super::*; - use configs::ng_cluster::TransportPorts; + use configs::ng_cluster::{ClusterNodeConfig, TransportPorts}; const READ_PATH: &str = "/streams?consistency=linearizable"; fn node(replica_id: u8, ip: &str, http: Option<u16>) -> ClusterNodeConfig { @@ -585,6 +592,7 @@ mod tests { name: format!("node-{replica_id}"), ip: ip.to_owned(), advertised_address: None, + advertised_addresses: Vec::new(), replica_id, ports: TransportPorts { tcp: None, @@ -600,7 +608,7 @@ mod tests { ClusterRoster { enabled: true, name: "test-cluster".to_owned(), - nodes, + nodes: nodes.into_iter().map(Into::into).collect(), self_ip: "127.0.0.1".to_owned(), self_ports: TransportPorts::default(), metadata_view: std::sync::Arc::new(std::sync::atomic::AtomicU64::new( @@ -616,7 +624,7 @@ mod tests { node(1, "10.0.0.2", Some(8090)), ]); assert_eq!( - primary_redirect_location(&roster, 1, "http", READ_PATH), + primary_redirect_location(&roster, 1, "http", READ_PATH, None), Some("http://10.0.0.2:8090/streams?consistency=linearizable".to_owned()) ); } @@ -625,7 +633,7 @@ mod tests { fn primary_redirect_location_uses_the_listener_scheme() { let roster = roster(vec![node(0, "10.0.0.1", Some(8080))]); assert_eq!( - primary_redirect_location(&roster, 0, "https", READ_PATH), + primary_redirect_location(&roster, 0, "https", READ_PATH, None), Some("https://10.0.0.1:8080/streams?consistency=linearizable".to_owned()) ); } @@ -634,7 +642,7 @@ mod tests { fn primary_redirect_location_is_none_when_no_node_matches_primary_index() { let roster = roster(vec![node(0, "10.0.0.1", Some(8080))]); assert_eq!( - primary_redirect_location(&roster, 2, "http", READ_PATH), + primary_redirect_location(&roster, 2, "http", READ_PATH, None), None ); } @@ -643,7 +651,7 @@ mod tests { fn primary_redirect_location_is_none_when_primary_has_no_http_port() { let roster = roster(vec![node(0, "10.0.0.1", None)]); assert_eq!( - primary_redirect_location(&roster, 0, "http", READ_PATH), + primary_redirect_location(&roster, 0, "http", READ_PATH, None), None ); } @@ -652,7 +660,7 @@ mod tests { fn primary_redirect_location_is_none_for_empty_roster() { let roster = roster(Vec::new()); assert_eq!( - primary_redirect_location(&roster, 0, "http", READ_PATH), + primary_redirect_location(&roster, 0, "http", READ_PATH, None), None ); } @@ -661,7 +669,7 @@ mod tests { fn primary_redirect_location_brackets_ipv6_host() { let roster = roster(vec![node(0, "::1", Some(8080))]); assert_eq!( - primary_redirect_location(&roster, 0, "http", READ_PATH), + primary_redirect_location(&roster, 0, "http", READ_PATH, None), Some("http://[::1]:8080/streams?consistency=linearizable".to_owned()) ); } @@ -673,7 +681,7 @@ mod tests { let roster = roster(vec![primary]); assert_eq!( - primary_redirect_location(&roster, 0, "https", READ_PATH), + primary_redirect_location(&roster, 0, "https", READ_PATH, None), Some("https://[2001:db8::1]:8080/streams?consistency=linearizable".to_owned()) ); } @@ -685,11 +693,45 @@ mod tests { let roster = roster(vec![primary]); assert_eq!( - primary_redirect_location(&roster, 0, "https", READ_PATH), + primary_redirect_location(&roster, 0, "https", READ_PATH, None), Some("https://broker-1.example.com:8080/streams?consistency=linearizable".to_owned()) ); } + #[test] + fn primary_redirect_location_uses_the_selector_address_for_a_matching_client() { + let mut primary = node(0, "10.0.0.1", Some(8080)); + primary.advertised_address = Some("203.0.113.1".to_owned()); + primary.advertised_addresses = vec![configs::ng_cluster::AdvertisedAddressSelector { + client_cidr: "10.0.0.0/16".to_owned(), + address: "10.0.0.1".to_owned(), + }]; + let roster = roster(vec![primary]); + + assert_eq!( + primary_redirect_location( + &roster, + 0, + "https", + READ_PATH, + Some("10.0.9.9".parse().unwrap()) + ), + Some("https://10.0.0.1:8080/streams?consistency=linearizable".to_owned()), + "an in-network client must be redirected to the selector address" + ); + assert_eq!( + primary_redirect_location( + &roster, + 0, + "https", + READ_PATH, + Some("198.51.100.7".parse().unwrap()) + ), + Some("https://203.0.113.1:8080/streams?consistency=linearizable".to_owned()), + "an out-of-network client must stay on the catch-all address" + ); + } + #[test] fn primary_http_socket_uses_private_roster_ip() { let mut primary = node(0, "10.0.0.1", Some(8080)); diff --git a/core/server-ng/src/http/extractor.rs b/core/server-ng/src/http/extractor.rs index eb52a8321..c98983ce2 100644 --- a/core/server-ng/src/http/extractor.rs +++ b/core/server-ng/src/http/extractor.rs @@ -17,9 +17,10 @@ //! Bearer-credential extractor for protected shard-0 HTTP routes. +use std::net::IpAddr; use std::rc::Rc; -use axum::extract::FromRequestParts; +use axum::extract::{ConnectInfo, FromRequestParts}; use axum::http::HeaderMap; use axum::http::header::AUTHORIZATION; use axum::http::request::Parts; @@ -30,6 +31,7 @@ use super::HttpState; use super::error::AuthError; use super::session::HttpSession; use crate::auth::verify_pat_credentials_with_expiry; +use crate::http::ClientAddr; /// Bearer scheme prefix in the `Authorization` header. const BEARER: &str = "Bearer "; @@ -96,6 +98,12 @@ pub struct Identity { /// `Location` for its 307 redirect to the primary. Empty only when the URI /// carries neither, which a routed read never is. pub path_and_query: String, + /// Transport-level peer IP, from the [`ClientAddr`] connect info both + /// serve paths install. Used only to pick the advertised address a client + /// is told about (cluster metadata, the 307 redirect `Location`) - never + /// for authorization. `None` (defensive; the extension is always present + /// today) degrades to the catch-all advertised address. + pub client_ip: Option<IpAddr>, } impl FromRequestParts<HttpState> for Identity { @@ -121,9 +129,14 @@ impl FromRequestParts<HttpState> for Identity { .path_and_query() .map(|value| value.as_str().to_owned()) .unwrap_or_default(); + let client_ip = parts + .extensions + .get::<ConnectInfo<ClientAddr>>() + .map(|ConnectInfo(address)| address.0.ip()); Ok(Self { user_id, path_and_query, + client_ip, }) } } diff --git a/core/server-ng/src/http/handlers.rs b/core/server-ng/src/http/handlers.rs index f97f3a53b..c2a90eab9 100644 --- a/core/server-ng/src/http/handlers.rs +++ b/core/server-ng/src/http/handlers.rs @@ -599,12 +599,13 @@ pub(in crate::http) async fn get_snapshot( /// the per-op authorization gate and the consistency gate, and serves from the /// roster captured at listener start plus the sync consensus getters, so it /// never touches the metadata STM, consensus, or a VSR session and stays fully -/// synchronous. +/// synchronous. The caller's peer IP picks each node's advertised address from +/// its per-client-network selectors. pub(in crate::http) async fn get_cluster_metadata( State(state): State<HttpState>, - _identity: Identity, + identity: Identity, ) -> Json<ClusterMetadata> { - Json(state.build_cluster_metadata()) + Json(state.build_cluster_metadata(identity.client_ip)) } /// `GET /clients`: list every connected client across all shards as the same diff --git a/core/server-ng/src/http/reads.rs b/core/server-ng/src/http/reads.rs index e84cb9944..9246ad610 100644 --- a/core/server-ng/src/http/reads.rs +++ b/core/server-ng/src/http/reads.rs @@ -65,7 +65,7 @@ pub(in crate::http) fn authorize_read( .authorize(|permissioner| rule(permissioner, identity.user_id)) .map_err(ReadError::Rejected)?; if consistency == Consistency::Linearizable && !state.is_metadata_primary() { - return Err(state.not_primary_read_error(&identity.path_and_query)); + return Err(state.not_primary_read_error(&identity.path_and_query, identity.client_ip)); } Ok(()) } @@ -98,6 +98,7 @@ pub(in crate::http) async fn read_local( body, Some(identity.user_id), &state.roster, + identity.client_ip, ) .map_err(ReadError::Rejected)? { diff --git a/core/server-ng/src/http/state.rs b/core/server-ng/src/http/state.rs index 3971f9aeb..7efb8bcc4 100644 --- a/core/server-ng/src/http/state.rs +++ b/core/server-ng/src/http/state.rs @@ -21,6 +21,7 @@ use std::cell::{Cell, RefCell}; use std::collections::HashMap; +use std::net::IpAddr; use std::rc::Rc; use std::sync::Arc; @@ -124,8 +125,15 @@ impl HttpInner { /// current VSR primary's HTTP address when it resolves from the roster, else /// fail closed to the 503. The target is the roster node whose `replica_id` /// equals `primary_index(view)`; an absent consensus, an unmatched id, or a - /// port-less node all fall back to [`ReadError::NotPrimary`]. - pub(in crate::http) fn not_primary_read_error(&self, path_and_query: &str) -> ReadError { + /// port-less node all fall back to [`ReadError::NotPrimary`]. `client_ip` + /// picks the primary's advertised address from its per-client-network + /// selectors, so the redirected client lands on the address for its own + /// network. + pub(in crate::http) fn not_primary_read_error( + &self, + path_and_query: &str, + client_ip: Option<IpAddr>, + ) -> ReadError { let location = self .shard .plane @@ -139,6 +147,7 @@ impl HttpInner { primary_index, self.forward.scheme, path_and_query, + client_ip, ) }); location.map_or(ReadError::NotPrimary, ReadError::RedirectToPrimary) @@ -397,7 +406,12 @@ impl HttpInner { /// shared [`ClusterRoster`] assembly. The leader marking comes from this /// shard's consensus view; the HTTP listener is shard-0-only, so consensus is /// always present and every roster read carries real leader/follower roles. - pub(in crate::http) fn build_cluster_metadata(&self) -> ClusterMetadata { + /// `client_ip` picks each node's advertised address from its + /// per-client-network selectors. + pub(in crate::http) fn build_cluster_metadata( + &self, + client_ip: Option<IpAddr>, + ) -> ClusterMetadata { let primary_index = self .shard .plane @@ -405,7 +419,7 @@ impl HttpInner { .consensus .as_ref() .map(|consensus| consensus.primary_index(consensus.view())); - self.roster.cluster_metadata(primary_index) + self.roster.cluster_metadata(primary_index, client_ip) } } diff --git a/core/server-ng/src/http/tls.rs b/core/server-ng/src/http/tls.rs index 830e79d22..c987c69dc 100644 --- a/core/server-ng/src/http/tls.rs +++ b/core/server-ng/src/http/tls.rs @@ -41,6 +41,7 @@ use std::time::Duration; use async_channel::{Receiver, Sender}; use axum::Router; +use axum::extract::ConnectInfo; use compio::net::{TcpListener, TcpStream}; use compio::runtime::JoinHandle; use compio::tls::{TlsAcceptor, TlsStream}; @@ -54,8 +55,10 @@ use message_bus::ShutdownToken; use message_bus::transports::tls::{ TlsServerCredentials, install_default_crypto_provider, load_pem, }; +use tower_http::add_extension::AddExtension; use tracing::{debug, error}; +use crate::http::ClientAddr; use crate::server_error::ServerNgError; /// hyper's auto-builder serves whichever protocol the client selects via @@ -124,6 +127,9 @@ pub fn spawn_accept_pump( /// shutdown the pump drops its sender, this loop ends, and the in-flight /// connection tasks drain via their own shutdown clone. pub async fn serve(connections: Receiver<Handshaken>, router: Router, shutdown: ShutdownToken) { + // `with_state(())` finalizes the routes eagerly, once for the whole + // listener; per-connection clones share the finalized internals. + let router: Router = router.with_state(()); while let Ok((tls, peer)) = connections.recv().await { let router = router.clone(); let shutdown = shutdown.clone(); @@ -143,8 +149,16 @@ async fn serve_connection( let io = HyperStream::new_tls(tls); // `Router<()>` already maps the incoming body to axum's `Body` in its own // `Service` impl, so it serves hyper's `Request<Incoming>` directly - no - // `map_request` shim. `with_state(())` finalizes the routes eagerly. - let service = TowerToHyperService::new(router.with_state(())); + // `map_request` shim. + // + // This hand-rolled loop bypasses axum's connect-info make-service (the + // plain listener's source of the peer address), so stamp the identical + // `ConnectInfo<ClientAddr>` extension on every request of this connection + // here - the extractors cannot tell the two paths apart. `AddExtension` + // wraps the shared router as one thin per-request insert; `Router::layer` + // would rebuild every route's boxed service on each connection. + let service = + TowerToHyperService::new(AddExtension::new(router, ConnectInfo(ClientAddr(peer)))); let builder = Builder::new(LocalExecutor); // `serve_connection_with_upgrades` borrows `builder`, so it must outlive // `conn`; keep it bound rather than inlined. diff --git a/core/server-ng/src/responses.rs b/core/server-ng/src/responses.rs index 6cc585230..2254b1ce1 100644 --- a/core/server-ng/src/responses.rs +++ b/core/server-ng/src/responses.rs @@ -87,6 +87,7 @@ use server_common::send_messages2::{COMMAND_HEADER_SIZE, SendMessages2Header}; use server_common::sharding::IggyNamespace; use shard::ConnectedClientInfo; use std::cell::RefCell; +use std::net::IpAddr; use std::path::PathBuf; use std::rc::Rc; use std::sync::{Arc, OnceLock}; @@ -463,13 +464,17 @@ where /// `user_id` is the authenticated caller, used only by the identity-scoped /// reads (currently the PAT list); every other arm ignores it. Authorization -/// stays with the per-transport gates that run before this builder. +/// stays with the per-transport gates that run before this builder. `client_ip` +/// is the caller's transport-level peer address, used only by the +/// cluster-metadata read to pick each node's advertised address; `None` +/// degrades to the catch-all address. pub(crate) fn build_non_replicated_response<B, MJ, S>( shard: &Rc<ShellShard<B, MJ, S>>, code: u32, body: &[u8], user_id: Option<u32>, roster: &ClusterRoster, + client_ip: Option<IpAddr>, ) -> Result<NonReplicatedResponse, IggyError> where B: ShellBus, @@ -479,7 +484,7 @@ where { match code { GET_CLUSTER_METADATA_CODE => Ok(NonReplicatedResponse::Bytes( - build_cluster_metadata_response(roster, shard).to_bytes(), + build_cluster_metadata_response(roster, shard, client_ip).to_bytes(), )), GET_STATS_CODE => Ok(NonReplicatedResponse::Bytes( build_stats_response(shard)?.to_bytes(), @@ -597,6 +602,7 @@ where fn build_cluster_metadata_response<B, MJ, S>( roster: &ClusterRoster, shard: &Rc<ShellShard<B, MJ, S>>, + client_ip: Option<IpAddr>, ) -> ClusterMetadataResponse where B: ShellBus, @@ -621,7 +627,7 @@ where .then_some(primary_index) }) .or_else(|| roster.current_primary_index()); - let metadata = roster.cluster_metadata(primary_index); + let metadata = roster.cluster_metadata(primary_index, client_ip); ClusterMetadataResponse { name: metadata.name, nodes: metadata diff --git a/core/server-ng/src/session_manager.rs b/core/server-ng/src/session_manager.rs index 9033aa98b..b7e63698a 100644 --- a/core/server-ng/src/session_manager.rs +++ b/core/server-ng/src/session_manager.rs @@ -291,6 +291,17 @@ impl SessionManager { } } + /// The transport-level peer address a connection arrived from, recorded by + /// [`Self::ensure_connection`] for every transport. The non-replicated + /// read path uses it to pick the advertised address a client is told + /// about; `None` (unknown connection) degrades to the catch-all address. + #[must_use] + pub fn connection_address(&self, connection_id: u128) -> Option<SocketAddr> { + self.connections + .get(&connection_id) + .map(|conn| conn.address) + } + /// Look up the authenticated user id for a connection. #[must_use] pub fn get_user_id(&self, connection_id: u128) -> Option<u32> {
