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 e757e16a2bbb9d15e7b9f59c7daf0a9fe889c8f3 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 | 932 ++++++++++++++++++++- core/configs/src/server_ng_config/defaults.rs | 1 + core/configs/src/server_ng_config/validators.rs | 1 + core/configs_derive/src/config_env.rs | 22 +- .../tests/server/cluster_metadata_vsr.rs | 229 ++++- core/integration/tests/server/http_client.rs | 19 +- core/integration/tests/server/http_tls.rs | 112 ++- core/server-ng/config.toml | 27 + core/server-ng/src/bootstrap.rs | 9 +- core/server-ng/src/cluster_meta.rs | 125 ++- core/server-ng/src/dispatch.rs | 24 +- core/server-ng/src/http.rs | 37 +- core/server-ng/src/http/error.rs | 102 ++- 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 | 15 +- core/server-ng/src/responses.rs | 12 +- core/server-ng/src/session_manager.rs | 11 + 23 files changed, 1600 insertions(+), 128 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..2296def6e 100644 --- a/core/configs/src/server_ng_config/cluster.rs +++ b/core/configs/src/server_ng_config/cluster.rs @@ -27,8 +27,10 @@ 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::cmp::Reverse; use std::fmt; use std::net::{IpAddr, Ipv6Addr, SocketAddr}; use std::str::FromStr; @@ -367,6 +369,13 @@ 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)] + #[config_env(max_elements = 16)] + 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 +384,144 @@ 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. +/// 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 and catch-all parsed +/// once, built wherever a roster is assembled for serving clients +/// (listener/shard start). Per-request resolution never re-parses config +/// strings: everything is snapshotted here, so mutating the source config +/// after conversion has no effect on what clients are told. Entries that do +/// not parse are dropped at build time; validation already rejects them +/// whenever the cluster is enabled, and a disabled cluster never consults +/// the roster. +#[derive(Debug, Clone)] +pub struct ResolvedClusterNode { + config: ClusterNodeConfig, + /// Truncated, canonicalized selector networks with their parsed + /// addresses, in declaration order. + selectors: Vec<(IpNet, AdvertisedAddress)>, + /// Parsed catch-all: [`ClusterNodeConfig::advertised_address`], else the + /// roster [`ClusterNodeConfig::ip`]. `None` when the configured value + /// does not parse - a set `advertised_address` never falls through to + /// the private roster ip. + catch_all: Option<AdvertisedAddress>, +} + +impl From<ClusterNodeConfig> for ResolvedClusterNode { + fn from(config: ClusterNodeConfig) -> Self { + let selectors = config + .advertised_addresses + .iter() + .filter_map(|selector| { + let network = selector.client_cidr.parse::<IpNet>().ok()?; + let address = selector.address.parse::<AdvertisedAddress>().ok()?; + Some((canonical_ip_net(network.trunc()), address)) + }) + .collect(); + let catch_all = match config.advertised_address.as_deref() { + Some(advertised_address) => advertised_address.parse().ok(), + None => config.ip.parse().ok(), + }; + Self { + config, + selectors, + catch_all, + } + } +} + +impl ResolvedClusterNode { + /// The roster entry this node was built from. Read-only: resolution runs + /// on the boot-parsed snapshot, never on the config strings. + #[must_use] + pub fn config(&self) -> &ClusterNodeConfig { + &self.config + } + + /// The client-facing address for a client connecting from `client_ip`: + /// longest-prefix match over the selector networks, then the parsed + /// catch-all. `None` when no selector matches and the catch-all did not + /// parse; callers choose whether to fail closed (redirect URLs) or to + /// publish [`Self::raw_advertised_fallback`] verbatim (cluster metadata). + #[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(self.catch_all.as_ref()) + } + + /// The catch-all ladder ([`ClusterNodeConfig::advertised_address`], else + /// the roster [`ClusterNodeConfig::ip`]) as configured, unparsed. Cluster + /// metadata publishes this verbatim when [`Self::advertised_for`] finds + /// nothing: the roster `ip` is only validated non-empty, and Docker + /// service names with underscores exist in the wild. + #[must_use] + pub fn raw_advertised_fallback(&self) -> &str { + self.config + .advertised_address + .as_deref() + .unwrap_or(&self.config.ip) + } + + /// Longest-prefix match over the boot-parsed selector networks. 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. `min_by_key` keeps the first of equal-length matches, so + /// resolution stays declaration-order deterministic even though a + /// validated config cannot produce two matching networks of equal length + /// (equal-length distinct networks are disjoint, duplicates are + /// rejected). + fn selector_address(&self, client_ip: IpAddr) -> Option<&AdvertisedAddress> { + let client_ip = client_ip.to_canonical(); + self.selectors + .iter() + .filter(|(network, _)| network.contains(&client_ip)) + .min_by_key(|(network, _)| Reverse(network.prefix_len())) + .map(|(_, address)| address) + } +} + +/// 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 +893,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 +984,104 @@ impl Validatable<ConfigurationError> for ClusterConfig { None => node.ip.parse::<AdvertisedAddress>().ok(), }; - 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) - } - }; - if !inserted { + // Selector CIDRs and addresses feed clients the same way the + // catch-all advertised address does, so they get the same strict + // parse. 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. + // Parsed before the catch-all enters the conflict pool because + // every entry's effective client set depends on the node's full + // selector list. + let mut selectors = Vec::with_capacity(node.advertised_addresses.len()); + 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); + } + }; + selectors.push((client_cidr, address)); + } + let selector_ranges: Vec<ClientAddressRange> = selectors + .iter() + .map(|(network, _)| ClientAddressRange::from(network)) + .collect(); + + // Endpoint conflicts are checked across every node's selectors + // and catch-all on effective client sets - the clients an entry + // actually wins after this node's longest-prefix match. Two + // nodes may reuse one host:port as long as the winning sets stay + // disjoint (that is the feature, and it includes a per-subnet + // override shadowing the same node's wider selector); a conflict + // means some client wins both entries and would resolve both + // nodes to one endpoint. The catch-all is an implicit + // match-everything-else selector, so it pools the same way. A + // roster ip that fails the strict parse skips the pool: it can + // never equal a parsed host, and two raw ips sharing host:port + // are already rejected by the bind-endpoint check above. + if let Some(address) = &client_address { + let catch_all_clients = EffectiveClients::for_catch_all(&selector_ranges); + 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: None, + clients: catch_all_clients.clone(), + host: address.clone(), + port: *port, + }, + )?; + } + } + } + + for (selector_index, (client_cidr, address)) in selectors.iter().enumerate() { + let sibling_ranges: Vec<ClientAddressRange> = selector_ranges + .iter() + .enumerate() + .filter(|(other_index, _)| *other_index != selector_index) + .map(|(_, range)| *range) + .collect(); + let clients = EffectiveClients::for_selector(client_cidr, &sibling_ranges); + 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), + clients: clients.clone(), + host: address.clone(), + port: *port, + }, + )?; + } } } } @@ -933,6 +1157,204 @@ impl Validatable<ConfigurationError> for ClusterConfig { } } +/// One advertised client endpoint and the clients it wins, pooled by +/// [`ClusterConfig::validate`] so selectors and catch-all conflict-check +/// against each other. `network: None` is the catch-all (`advertised_address`, +/// or the roster `ip` as fallback); `clients` is the entry's effective set +/// after its node's longest-prefix shadowing. +struct AdvertisedEndpoint<'roster> { + node_name: &'roster str, + transport: &'static str, + network: Option<IpNet>, + clients: EffectiveClients, + host: AdvertisedAddress, + port: u16, +} + +impl AdvertisedEndpoint<'_> { + /// True when some client would resolve both entries to one host:port on + /// two different nodes. Effective client sets already encode each node's + /// longest-prefix shadowing, so nested networks conflict only where the + /// wider entry still wins some client that the other node's entry also + /// wins. Entries of one node never conflict: their effective sets are + /// disjoint by construction. + fn conflicts_with(&self, other: &Self) -> bool { + self.node_name != other.node_name + && self.port == other.port + && self.host == other.host + && self.clients.overlaps(&other.clients) + } + + fn authority(&self) -> String { + self.host.authority(self.port) + } + + fn network_description(&self) -> String { + match self.network { + Some(network) => format!("client_cidr {network}"), + None => "every client network (catch-all)".to_owned(), + } + } +} + +/// The clients an advertised entry actually wins under its node's +/// longest-prefix match, built by [`ClusterConfig::validate`] for the +/// cross-node conflict scan. +#[derive(Clone)] +struct EffectiveClients { + /// Sorted disjoint ranges of winning client addresses. + ranges: Vec<ClientAddressRange>, + /// The catch-all also wins clients whose peer address the transport + /// could not produce ([`ResolvedClusterNode::advertised_for`] with no + /// client IP), so two catch-alls overlap even when selectors cover both + /// address families. + serves_unknown_peers: bool, +} + +impl EffectiveClients { + /// A selector wins its network minus the sibling networks nested inside + /// it (longer prefixes take the node's LPM). `sibling_ranges` must + /// exclude the selector's own network. + fn for_selector(network: &IpNet, sibling_ranges: &[ClientAddressRange]) -> Self { + Self { + ranges: ClientAddressRange::from(network).subtract_nested(sibling_ranges), + serves_unknown_peers: false, + } + } + + /// The catch-all wins every client no selector matches, in both address + /// families, plus unknown-peer clients. + fn for_catch_all(selector_ranges: &[ClientAddressRange]) -> Self { + let mut ranges = ClientAddressRange::FULL_IPV4.subtract_nested(selector_ranges); + ranges.extend(ClientAddressRange::FULL_IPV6.subtract_nested(selector_ranges)); + Self { + ranges, + serves_unknown_peers: true, + } + } + + fn overlaps(&self, other: &Self) -> bool { + if self.serves_unknown_peers && other.serves_unknown_peers { + return true; + } + self.ranges.iter().any(|range| { + other.ranges.iter().any(|other_range| { + range.is_ipv4 == other_range.is_ipv4 + && range.first <= other_range.last + && other_range.first <= range.last + }) + }) + } +} + +/// Inclusive range of client addresses within one family. Client IPs +/// canonicalize to v4 before matching, so v4 and v6 networks match disjoint +/// client populations and a range never spans families. +#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +struct ClientAddressRange { + is_ipv4: bool, + first: u128, + last: u128, +} + +impl ClientAddressRange { + const FULL_IPV4: Self = Self { + is_ipv4: true, + first: 0, + last: u32::MAX as u128, + }; + const FULL_IPV6: Self = Self { + is_ipv4: false, + first: 0, + last: u128::MAX, + }; + + /// `self` minus every range nested inside it, as sorted disjoint + /// leftovers. CIDR networks are nested or disjoint, never partially + /// overlapping, so a range outside `self` is either disjoint from it + /// (subtracts nothing) or contains it (a shorter prefix, which loses + /// LPM and also subtracts nothing). + fn subtract_nested(self, ranges: &[Self]) -> Vec<Self> { + let mut nested: Vec<Self> = ranges + .iter() + .filter(|range| { + range.is_ipv4 == self.is_ipv4 + && range.first >= self.first + && range.last <= self.last + }) + .copied() + .collect(); + nested.sort_unstable(); + let mut remaining = Vec::new(); + let mut cursor = Some(self.first); + for nested_range in nested { + let Some(next_free) = cursor else { break }; + if nested_range.first > next_free { + remaining.push(Self { + is_ipv4: self.is_ipv4, + first: next_free, + last: nested_range.first - 1, + }); + } + cursor = nested_range + .last + .checked_add(1) + .map(|after| after.max(next_free)); + } + if let Some(next_free) = cursor + && next_free <= self.last + { + remaining.push(Self { + is_ipv4: self.is_ipv4, + first: next_free, + last: self.last, + }); + } + remaining + } +} + +impl From<&IpNet> for ClientAddressRange { + fn from(network: &IpNet) -> Self { + match network { + IpNet::V4(network) => Self { + is_ipv4: true, + first: u128::from(u32::from(network.network())), + last: u128::from(u32::from(network.broadcast())), + }, + IpNet::V6(network) => Self { + is_ipv4: false, + first: u128::from(network.network()), + last: u128::from(network.broadcast()), + }, + } + } +} + +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 {}); their effective client sets overlap after longest-prefix shadowing, 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::*; @@ -996,6 +1418,26 @@ mod tests { "unexpected deserialization error: {error}" ); } + + #[test] + fn advertised_addresses_env_expansion_is_capped() { + // The selectors Vec nests inside the nodes Vec, so the derive's + // index ceilings multiply; without the field's max_elements cap the + // default of 256x256 adds ~131k leaked mappings to every boot. + let mappings = <ClusterConfig as configs::ConfigEnvMappings>::env_mappings(); + assert!( + mappings + .iter() + .any(|mapping| mapping.env_name.contains("ADVERTISED_ADDRESSES_15_")), + "selector index 15 must stay reachable by env override" + ); + assert!( + !mappings + .iter() + .any(|mapping| mapping.env_name.contains("ADVERTISED_ADDRESSES_16_")), + "selector env expansion must stop at max_elements = 16" + ); + } } #[cfg(test)] @@ -1075,6 +1517,167 @@ 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 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 equal_prefix_matches_resolve_deterministically_to_first_declared() { + // No validated config reaches this state: these networks truncate to + // one /16, which validation rejects as a duplicate. Pinned anyway so + // a future relaxation of that rule cannot make resolution + // order-dependent. + 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() + )) + ); + } +} + #[cfg(test)] mod cluster_validate_tests { use super::*; @@ -1084,11 +1687,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 +2117,291 @@ 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: n1 has no longer prefix of + // its own shadowing the /16 overlap, so a 10.0.0.0/16 client wins + // n1's /8 and n2's /16, resolving both 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/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_nested_cidr_reuse_shadowed_by_same_node_longer_prefix() { + // n1's /16 selector shadows its /8 within 10.0.0.0/16, so n1's /8 + // entry wins only 10.0.0.0/8 minus 10.0.0.0/16 - disjoint from n2's + // /16. No client resolves 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/8", "192.0.2.10"), + selector("10.0.0.0/16", "192.0.2.20"), + ]; + 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_ok()); + } + + #[test] + fn validate_rejects_partially_shadowed_nested_cidr_reuse() { + // n1's /24 shadow carves only part of the /16 overlap: a client in + // 10.0.0.0/16 outside 10.0.0.0/24 still wins n1's /8 entry and n2's + // /16 entry, resolving 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/8", "192.0.2.10"), + selector("10.0.0.0/24", "192.0.2.20"), + ]; + 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_catch_all_reuse_shadowed_by_same_node_selector() { + // n1's /16 selector shadows its catch-all within 10.0.0.0/16, so + // the catch-all never wins a client inside n2's /24. Without the + // shadow the same pair conflicts (see + // validate_rejects_selector_endpoint_clashing_with_another_nodes_catch_all). + 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.20")]; + 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/24", "192.0.2.10")]; + n2.ports.tcp = Some(8090); + + assert!(cfg(vec![n1, n2]).validate().is_ok()); + } + + #[test] + fn validate_accepts_selector_reusing_a_fully_shadowed_catch_all_address() { + // n1's selectors cover both address families, so its catch-all wins + // known peers nowhere; only unknown-peer clients reach it, and they + // never match n2's selector. + 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("0.0.0.0/0", "192.0.2.20"), + selector("::/0", "192.0.2.30"), + ]; + 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_ok()); + } + + #[test] + fn validate_accepts_catch_all_spelling_another_nodes_selector_address_when_self_shadowed() { + // Split-network NAT roster: n2's catch-all spells n1's 10/8 selector + // address, but n2's own 10/8 selector shadows its catch-all inside + // 10/8 (outside it n1 serves its own catch-all), so no client + // resolves 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/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_address = Some("192.0.2.10".to_owned()); + n2.advertised_addresses = vec![selector("10.0.0.0/8", "192.0.2.20")]; + n2.ports.tcp = Some(8090); + + assert!(cfg(vec![n1, n2]).validate().is_ok()); + } + + #[test] + fn validate_rejects_duplicate_catch_all_even_when_fully_shadowed() { + // A client whose peer address the transport cannot produce always + // falls to the catch-all, so duplicate catch-alls conflict even when + // selectors cover every known network. + 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("0.0.0.0/0", "192.0.2.20"), + selector("::/0", "192.0.2.30"), + ]; + 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.advertised_addresses = vec![ + selector("0.0.0.0/0", "192.0.2.40"), + selector("::/0", "192.0.2.50"), + ]; + 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/configs_derive/src/config_env.rs b/core/configs_derive/src/config_env.rs index 33f15e9f6..69dbc0c85 100644 --- a/core/configs_derive/src/config_env.rs +++ b/core/configs_derive/src/config_env.rs @@ -20,16 +20,18 @@ use proc_macro2::TokenStream as TokenStream2; use quote::{format_ident, quote}; use syn::{DeriveInput, Generics, Ident, Type}; -/// Maximum number of array elements to generate env var mappings for. +/// Default number of array elements to generate env var mappings for. /// /// For `Vec<T>` fields, mappings are generated for indices 0 through 255 /// (e.g., `FIELD_0_NAME`, ..., `FIELD_255_NAME`). Environment variables for /// indices beyond this limit are silently ignored. /// -/// The ceiling mirrors `cluster.nodes` (the only known Vec<nested> -/// configuration surface), which the validator allows up to 255 replicas. -/// Keeping the two in sync means every node index reachable by the validator -/// is also reachable by env overrides. +/// The ceiling mirrors `cluster.nodes`, which the validator allows up to 255 +/// replicas. Keeping the two in sync means every node index reachable by the +/// validator is also reachable by env overrides. Vecs nested inside another +/// Vec multiply their ceilings (256 x 256 entries, each `Box::leak`ed at +/// first use), so nested-vec fields like `cluster.nodes.advertised_addresses` +/// must lower theirs with `#[config_env(max_elements = N)]`. const MAX_ARRAY_ELEMENTS: usize = 256; /// Container-level attributes for `#[config_env(...)]` @@ -88,6 +90,11 @@ struct FieldOpts { /// Treat this field as a leaf value (not a nested config struct) #[darling(default)] leaf: bool, + + /// Cap the array-index expansion of a `Vec` field below + /// [`MAX_ARRAY_ELEMENTS`]. Ignored on non-Vec fields. + #[darling(default)] + max_elements: Option<usize>, } /// Represents a single environment variable mapping (leaf field) @@ -111,6 +118,8 @@ struct NestedFieldInfo { field_env_segment: String, /// The field name for config path construction field_name: String, + /// Array-index expansion ceiling for Vec fields + max_elements: usize, } pub fn generate_impl(input: &DeriveInput) -> TokenStream2 { @@ -261,7 +270,7 @@ fn generate_struct_impl( if info.is_vec { // For Vec fields, expand with array indices - let max_elements = MAX_ARRAY_ELEMENTS; + let max_elements = info.max_elements; quote! { let nested_mappings = <#ty as configs::ConfigEnvMappings>::env_mappings(); for i in 0..#max_elements { @@ -533,6 +542,7 @@ fn collect_mappings(fields: &[FieldOpts]) -> (Vec<EnvMapping>, Vec<NestedFieldIn is_vec, field_env_segment, field_name, + max_elements: field.max_elements.unwrap_or(MAX_ARRAY_ELEMENTS), }); } } diff --git a/core/integration/tests/server/cluster_metadata_vsr.rs b/core/integration/tests/server/cluster_metadata_vsr.rs index f15ad4416..2011f6f1d 100644 --- a/core/integration/tests/server/cluster_metadata_vsr.rs +++ b/core/integration/tests/server/cluster_metadata_vsr.rs @@ -17,10 +17,29 @@ //! 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` - never the decoy selector on a +//! foreign network. Every selector here advertises `localhost`: it is the +//! one hostname the leader-aware SDK client's address comparison rewrites to +//! `127.0.0.1`, so any other value makes that client redial the advertised +//! host on every fresh connect - including the harness readiness probe, +//! which would wedge startup on an unresolvable name. A non-`localhost` +//! hostname selector therefore cannot run end to end until the SDK-side +//! comparison is fixed; that flow is pinned at unit level instead +//! (`cluster_meta.rs`, `cluster.rs`). + +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 +93,211 @@ 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. Note `localhost` is also the one +/// hostname that client's `is_same_address` check rewrites to `127.0.0.1`, +/// so binary-transport tests must use it; the raw-hostname flow is pinned by +/// the HTTP-only [`HOSTNAME_SELECTOR_ADDRESS`] test instead. +const SELECTOR_ADDRESS: &str = "localhost"; + +/// Decoy selector on a network no test client connects from: its address in +/// metadata would mean CIDR matching is broken (matching everything, or +/// matching the server's own bind address instead of the peer), not that +/// selector precedence works. +const DECOY_CIDR: &str = "10.0.0.0/8"; +const DECOY_ADDRESS: &str = "203.0.113.99"; + +/// A 2-node cluster whose roster gives every node a loopback-CIDR selector +/// plus the decoy selector that must never match. 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 { + for (suffix, value) in [ + ("0_CLIENT_CIDR", LOOPBACK_CIDR), + ("0_ADDRESS", SELECTOR_ADDRESS), + ("1_CLIENT_CIDR", DECOY_CIDR), + ("1_ADDRESS", DECOY_ADDRESS), + ] { + harness.node_mut(node).add_env( + format!("IGGY_CLUSTER_NODES_{roster_entry}_ADVERTISED_ADDRESSES_{suffix}"), + value, + ); + } + } + } + harness +} + +/// Assert every roster node reports `expected_address` to this loopback +/// client; without the selector the same roster publishes `127.0.0.1`, and +/// broken CIDR matching would surface the decoy instead. +fn assert_advertised_addresses( + nodes: impl IntoIterator<Item = (String, String)>, + expected_address: &str, +) { + let mut seen = 0; + for (name, ip) in nodes { + assert_eq!( + ip, expected_address, + "node '{name}' must advertise its loopback selector address to a loopback client" + ); + seen += 1; + } + assert_eq!(seen, SELECTOR_CLUSTER_NODES, "full roster must be reported"); +} + +/// Extract `(name, ip)` per node from a `/cluster/metadata` HTTP reply. +async fn http_metadata_nodes(http: &HttpClient) -> Vec<(String, String)> { + 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"); + 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() +} + +#[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_advertised_addresses( + metadata + .nodes + .iter() + .map(|node| (node.name.clone(), node.ip.clone())), + SELECTOR_ADDRESS, + ); +} + +#[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; + assert_advertised_addresses(http_metadata_nodes(&http).await, SELECTOR_ADDRESS); +} + +/// 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 - + // once, before the settle loop: every login commits a Register through + // metadata consensus, so logging in per round would race the election the + // loop waits out, and login retries would eat the shared settle budget. + let mut sessions = Vec::with_capacity(SELECTOR_CLUSTER_NODES); + for node in 0..SELECTOR_CLUSTER_NODES { + let addr = harness.node(node).http_addr().expect("node http address"); + sessions.push(HttpClient::login_root_no_redirect(format!("http://{addr}";)).await); + } + + let deadline = Instant::now() + LOGIN_TIMEOUT; + let mut verdicts: Vec<(usize, StatusCode, Option<String>)>; + loop { + verdicts = Vec::new(); + for (node, session) in sessions.iter().enumerate() { + 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/integration/tests/server/http_tls.rs b/core/integration/tests/server/http_tls.rs index 80bd58910..f36e4a058 100644 --- a/core/integration/tests/server/http_tls.rs +++ b/core/integration/tests/server/http_tls.rs @@ -21,7 +21,11 @@ //! client over the wire and proves the response was actually served over //! HTTP/2, negotiated via ALPN. A failure here (h1 fallback, handshake //! rejection, or a plaintext bind) is a real defect in that path, not a test -//! artifact. +//! artifact. The `/cluster/metadata` step additionally proves the TLS serve +//! loop's hand-stamped `ConnectInfo<ClientAddr>` reaches the extractors: the +//! plain listener gets that extension from axum's connect-info make-service, +//! the TLS loop stamps it itself, and only a client-IP-dependent response +//! can tell a missing stamp from a working one. use std::path::PathBuf; use std::time::{Duration, Instant}; @@ -39,6 +43,26 @@ const READY_TIMEOUT: Duration = Duration::from_secs(15); const READY_RETRY_INTERVAL: Duration = Duration::from_millis(50); const REQUEST_TIMEOUT: Duration = Duration::from_secs(30); +/// Selector marker distinct from the roster ip (`127.0.0.1`): it appears in +/// metadata only when the request carried the client's peer address, so it +/// distinguishes a working `ConnectInfo` stamp from the catch-all fallback a +/// missing stamp silently degrades to. `localhost` specifically, because the +/// cluster readiness probe logs in over the leader-aware binary client, +/// which redials any other advertised hostname and would wedge startup on an +/// unresolvable name. +const TLS_SELECTOR_ADDRESS: &str = "localhost"; + +/// Selectors covering the loopback client in both address families +/// (`localhost` may resolve to either). +const TLS_SELECTOR_ENVS: [(&str, &str); 4] = [ + ("0_CLIENT_CIDR", "127.0.0.0/8"), + ("0_ADDRESS", TLS_SELECTOR_ADDRESS), + ("1_CLIENT_CIDR", "::1/128"), + ("1_ADDRESS", TLS_SELECTOR_ADDRESS), +]; + +const TLS_CLUSTER_NODES: usize = 2; + /// Absolute path to a repo loopback cert asset. The spawned server's CWD is a /// temp dir, so relative paths break; `CARGO_MANIFEST_DIR` is the integration /// crate, whose sibling `../certs` holds the checked-in loopback material. @@ -47,35 +71,49 @@ fn cert_asset(file: &str) -> PathBuf { .unwrap_or_else(|error| panic!("canonicalize repo cert asset {file}: {error}")) } -/// Boot iggy-server-ng with `[http.tls]` enabled against the repo loopback -/// cert, then prove a real HTTPS request is served over HTTP/2. Single node: -/// the HTTP listener is shard-0-only and cluster formation is irrelevant to -/// what we are proving, so a standalone node keeps the signal clean. +/// Boot an iggy-server-ng cluster with `[http.tls]` enabled against the repo +/// loopback cert, then prove a real HTTPS request is served over HTTP/2. Two +/// nodes rather than one because the `/cluster/metadata` assertion below +/// needs an enabled cluster roster (a single-node harness runs with the +/// cluster disabled, where metadata synthesizes a self node and selectors +/// are inert); the HTTPS requests themselves only ever talk to node 0. #[tokio::test] #[serial_test::parallel] async fn given_http_tls_enabled_when_pinging_should_serve_https_over_http2() { let mut harness = TestHarness::builder() .default_server() - .cluster_nodes(1) + .cluster_nodes(TLS_CLUSTER_NODES) .build() .expect("build TLS harness"); let cert = cert_asset("iggy_cert.pem"); let key = cert_asset("iggy_key.pem"); - harness - .server_mut() - .add_env("IGGY_HTTP_TLS_ENABLED", "true"); - harness - .server_mut() - .add_env("IGGY_HTTP_TLS_CERT_FILE", cert.display().to_string()); - harness - .server_mut() - .add_env("IGGY_HTTP_TLS_KEY_FILE", key.display().to_string()); + for node in 0..TLS_CLUSTER_NODES { + harness + .node_mut(node) + .add_env("IGGY_HTTP_TLS_ENABLED", "true"); + harness + .node_mut(node) + .add_env("IGGY_HTTP_TLS_CERT_FILE", cert.display().to_string()); + harness + .node_mut(node) + .add_env("IGGY_HTTP_TLS_KEY_FILE", key.display().to_string()); + // The roster (selectors included) must be identical on every node; + // the selectors back the ConnectInfo-stamp assertion below. + for roster_entry in 0..TLS_CLUSTER_NODES { + for (suffix, value) in TLS_SELECTOR_ENVS { + harness.node_mut(node).add_env( + format!("IGGY_CLUSTER_NODES_{roster_entry}_ADVERTISED_ADDRESSES_{suffix}"), + value, + ); + } + } + } harness .start() .await - .expect("start server-ng with HTTPS enabled"); + .expect("start server-ng cluster with HTTPS enabled"); let addr = harness .server() @@ -149,10 +187,46 @@ async fn given_http_tls_enabled_when_pinging_should_serve_https_over_http2() { "login response must also be HTTP/2" ); let identity: IdentityInfo = login.json().await.expect("decode IdentityInfo over HTTPS"); + let token = identity + .access_token + .expect("login over HTTPS must return an access token"); assert!( - identity - .access_token - .is_some_and(|token| !token.token.is_empty()), + !token.token.is_empty(), "login over HTTPS must return a non-empty access token" ); + + // ConnectInfo-stamp proof: the TLS serve loop stamps the peer address by + // hand (`http::tls::serve_connection`), and `Identity.client_ip` silently + // degrades to the catch-all when the extension is missing. Only the + // selector address in the metadata reply proves the stamp reached the + // handler; the catch-all here is the roster ip `127.0.0.1`. + let metadata = client + .get(format!("{base_url}/cluster/metadata")) + .bearer_auth(&token.token) + .send() + .await + .expect("cluster metadata request over HTTPS"); + assert_eq!( + metadata.status(), + StatusCode::OK, + "cluster metadata must answer 200 over HTTPS" + ); + let metadata: serde_json::Value = metadata.json().await.expect("decode cluster metadata"); + let nodes = metadata["nodes"] + .as_array() + .expect("metadata must carry a nodes array"); + assert_eq!( + nodes.len(), + TLS_CLUSTER_NODES, + "full roster must be reported" + ); + for node in nodes { + assert_eq!( + node["ip"].as_str(), + Some(TLS_SELECTOR_ADDRESS), + "HTTPS metadata must serve the loopback selector address for node {}, \ + proving the TLS path stamped the client peer address", + node["name"] + ); + } } diff --git a/core/server-ng/config.toml b/core/server-ng/config.toml index af6d3c9ce..52cc0c124 100644 --- a/core/server-ng/config.toml +++ b/core/server-ng/config.toml @@ -739,6 +739,33 @@ 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; clients matching no selector +# fall back to 'advertised_address', then 'ip'. Matching is per address +# family: '0.0.0.0/0' matches no IPv6 client and '::/0' matches no IPv4 +# client, so covering both families takes one selector per family (or the +# catch-all). IPv4-mapped IPv6 CIDRs ('::ffff:10.0.0.0/104') match like +# their IPv4 form only at prefix length 96 or longer; shorter ones match +# native IPv6 clients only. 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..10d17b83d 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,18 @@ 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`]). +/// Metadata deliberately does NOT fail closed like the redirect path: a host +/// that parses as neither IP nor hostname (the roster `ip` is only validated +/// non-empty - Docker service names with underscores exist in the wild) +/// publishes verbatim via [`ResolvedClusterNode::raw_advertised_fallback`]. +fn client_host(node: &ResolvedClusterNode, client_ip: Option<IpAddr>) -> String { + node.advertised_for(client_ip).map_or_else( + || node.raw_advertised_fallback().to_owned(), + ToString::to_string, + ) } const fn role_for(primary_index: Option<u8>, replica_id: u8) -> ClusterNodeRole { @@ -167,40 +181,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 +234,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 +246,47 @@ 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"); + } + + // Unit-level pin for the hostname-selector flow: it cannot run end to + // end because the leader-aware SDK client redials any advertised + // hostname other than `localhost` on every fresh connect, wedging the + // integration harness readiness probe on an unresolvable name (see + // cluster_metadata_vsr.rs). + #[test] + fn cluster_metadata_serves_a_hostname_selector_normalized_to_lowercase() { + 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: "Broker.Internal.Test".to_owned(), + }]; + + let metadata = roster_of(node).cluster_metadata(Some(0), Some("10.0.9.9".parse().unwrap())); + + assert_eq!(metadata.nodes[0].ip, "broker.internal.test"); + } } 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..82b3fd997 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). @@ -327,7 +349,12 @@ fn router( None => router, }; - merge_web_ui(router, web_ui) + // `with_state(())` finalizes every route eagerly, once for the whole + // listener - including the web-ui routes merged after the stateful + // `with_state` above, which would otherwise stay boxed handlers that + // axum rebuilds per request. Identity on already-finalized routes, so + // both the plain and the TLS serve paths share the finalized form. + merge_web_ui(router, web_ui).with_state(()) } /// The control-plane route table: every write here commits through the diff --git a/core/server-ng/src/http/error.rs b/core/server-ng/src/http/error.rs index 28f5f56c2..df52466c5 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,35 @@ 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. This is the fail-closed caller: a host that is neither a +/// valid IP nor a valid hostname yields `None` and the redirect becomes a +/// 503 rather than a `Location` pointing at an unparsable target (cluster +/// metadata makes the opposite choice and publishes such a host verbatim). +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 +586,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 +594,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 +610,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 +626,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 +635,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 +644,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 +653,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 +662,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 +671,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 +683,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 +695,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..b4e0137a5 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 @@ -143,8 +146,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> {
