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:34 UTC
(iggy) branch advertised-addresses created (now 8e13d5f5c)
This is an automated email from the ASF dual-hosted git repository.
mmodzelewski pushed a change to branch advertised-addresses
in repository https://gitbox.apache.org/repos/asf/iggy.git
at 8e13d5f5c feat(server-ng): add per-client-network advertised address selectors
This branch includes the following new commits:
new 8e13d5f5c feat(server-ng): add per-client-network advertised address selectors
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iggy) 01/01: feat(server-ng): add per-client-network advertised address selectors
Posted by ma...@apache.org.
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> {