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