This is an automated email from the ASF dual-hosted git repository.
mmodzelewski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new d9635b6d7 feat(server-ng): forward follower HTTP requests to VSR
primary (#3744)
d9635b6d7 is described below
commit d9635b6d71dacbe6be1702a0723df50f5574b789
Author: Hubert Gruszecki <[email protected]>
AuthorDate: Fri Jul 24 15:44:12 2026 +0200
feat(server-ng): forward follower HTTP requests to VSR primary (#3744)
A load balancer round-robining HTTP requests across cluster nodes
breaks the web UI: followers answer every control-plane write with
a transient not-a-leader 503 the browser cannot act on.
Followers now re-issue such requests to the metadata primary over
an internal cyper client and relay the response on the original
connection. Retries happen only for outcomes that provably never
entered the pipeline (connect failure, TransientNotAccepted reply,
307 redirect re-resolved from the local roster), so a request can
never be applied twice. submit_gated latches NotCommitted across
replay frames instead of surfacing the last frame's code, so a
may-have-committed op can never masquerade as safely-retryable.
The relayed response body is streamed under a hard cap, bounding
length-less replies as well as declared ones.
Cross-node token validity: when no JWT secret is configured, the
HS256 signing key is derived from the cluster shared secret via
blake3, so any node can verify tokens issued by another. The
forward hop pins the local TLS certificate; plaintext is allowed
and documented. The 307 linearizable-read redirect now shares the
forward path's roster resolution and scheme, so it points at
https targets correctly.
Response metadata headers follow RFC 6648: the new iggy-forwarded
hop guard is born unprefixed, and the existing x-iggy-view and
x-iggy-durability headers are renamed to iggy-view and
iggy-durability (server-ng is unreleased, so nothing breaks).
Partition-plane requests (messages, consumer offsets) still run
on the receiving node; forwarding them needs per-partition-group
targeting, marked with a TODO.
---
Cargo.lock | 1 +
Cargo.toml | 2 +-
bdd/docker-compose.vsr.yml | 7 +
core/configs/src/server_ng_config/cluster.rs | 19 +
core/configs/src/server_ng_config/validators.rs | 95 ++++
core/integration/tests/server/http_vsr.rs | 4 +-
core/server-ng/Cargo.toml | 1 +
core/server-ng/config.toml | 17 +-
core/server-ng/src/http.rs | 104 ++--
core/server-ng/src/http/error.rs | 130 +++--
core/server-ng/src/http/extractor.rs | 20 +-
core/server-ng/src/http/forward.rs | 632 ++++++++++++++++++++++++
core/server-ng/src/http/handlers.rs | 6 +-
core/server-ng/src/http/jwt.rs | 47 +-
core/server-ng/src/http/reply.rs | 51 +-
core/server-ng/src/http/state.rs | 36 +-
core/server-ng/src/http/submit.rs | 43 +-
core/server-ng/src/server_error.rs | 2 +
18 files changed, 1099 insertions(+), 118 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 4bc2420ba..7591cdbf3 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -11823,6 +11823,7 @@ dependencies = [
"async_zip",
"axum",
"axum-server",
+ "blake3",
"bytemuck",
"bytes",
"chrono",
diff --git a/Cargo.toml b/Cargo.toml
index 8d363b005..b03900ab1 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -146,7 +146,7 @@ csv = "1.4.0"
ctor = "1.0.9"
ctrlc = { version = "3.5", features = ["termination"] }
cucumber = "0.23"
-cyper = { version = "0.9.0", features = ["rustls"], default-features = false }
+cyper = { version = "0.9.0", features = ["rustls", "stream"], default-features
= false }
cyper-axum = { version = "0.9.0" }
cyper-core = { version = "0.9.0", default-features = false }
darling = "0.23"
diff --git a/bdd/docker-compose.vsr.yml b/bdd/docker-compose.vsr.yml
index 15d6ef1ed..af414d60b 100644
--- a/bdd/docker-compose.vsr.yml
+++ b/bdd/docker-compose.vsr.yml
@@ -49,6 +49,13 @@ x-vsr-cluster-env: &vsr-cluster-env
IGGY_CLUSTER_NODES_1_IP: 172.28.0.102
IGGY_CLUSTER_NODES_0_PORTS_TCP_REPLICA: "8191"
IGGY_CLUSTER_NODES_1_PORTS_TCP_REPLICA: "8192"
+ # http.enabled with cluster.enabled requires a JWT key every node can
+ # verify; cluster auth provides it (derived from the shared secret).
+ # Enabling auth activates follower-to-primary forwarding, so the config
+ # validator requires ports.http on every roster node - those ports are set
+ # in docker-compose.cluster.yml (merged for the leader_redirection flow).
+ IGGY_CLUSTER_AUTH_ENABLED: "true"
+ IGGY_CLUSTER_AUTH_SHARED_SECRET: "bdd-vsr-cluster-shared-secret-0123456789"
services:
iggy-server:
diff --git a/core/configs/src/server_ng_config/cluster.rs
b/core/configs/src/server_ng_config/cluster.rs
index b8dc2d7f4..44b565f58 100644
--- a/core/configs/src/server_ng_config/cluster.rs
+++ b/core/configs/src/server_ng_config/cluster.rs
@@ -24,6 +24,7 @@
use super::defaults::SERVER_NG_CONFIG;
use crate::ConfigurationError;
+use crate::http::HttpJwtConfig;
use configs::ConfigEnv;
use iggy_common::{IggyDuration, Validatable};
use serde::{Deserialize, Serialize};
@@ -164,6 +165,24 @@ pub struct TransportPorts {
pub tcp_replica: Option<u16>,
}
+/// Whether cluster-wide JWT key material exists: a configured `http.jwt`
+/// secret, or the signing key derived from the cluster PSK. When it does, a
+/// bearer minted on any node verifies on every node - the invariant
+/// follower-to-primary HTTP forwarding depends on. Callers gate `http.enabled`
+/// themselves; this covers only the key material.
+///
+/// Single source for both the boot-time config validator and the server-ng
+/// runtime forwarding gate. If the two ever disagree the validator's roster
+/// http-port guarantee is silently bypassed: forwarding would activate against
+/// a node the validator never required to expose an http port, and every
+/// forward through it fails closed with a 503.
+pub fn http_forwarding_key_material(jwt: &HttpJwtConfig, cluster:
&ClusterConfig) -> bool {
+ cluster.enabled
+ && ((cluster.auth.enabled && !cluster.auth.shared_secret.is_empty())
+ || !jwt.encoding_secret.is_empty()
+ || !jwt.decoding_secret.is_empty())
+}
+
impl Validatable<ConfigurationError> for ClusterConfig {
fn validate(&self) -> Result<(), ConfigurationError> {
if !self.enabled {
diff --git a/core/configs/src/server_ng_config/validators.rs
b/core/configs/src/server_ng_config/validators.rs
index 1c9745b8c..4736d9055 100644
--- a/core/configs/src/server_ng_config/validators.rs
+++ b/core/configs/src/server_ng_config/validators.rs
@@ -26,6 +26,7 @@
//! net.
use super::COMPONENT_NG;
+use super::cluster::http_forwarding_key_material;
use super::server_ng::{ExtraConfig, NamespaceConfig, ServerNgConfig};
use crate::ConfigurationError;
use err_trail::ErrContext;
@@ -137,6 +138,25 @@ impl Validatable<ConfigurationError> for ServerNgConfig {
return Err(ConfigurationError::InvalidConfigurationValue);
}
+ // Without key material forwarding is disabled (followers answer a
+ // transient 503) and the server still boots, so it is not required
+ // here. When it IS present the operator opted into forwarding, and the
+ // roster must support it: a node listed without an http port would
+ // silently degrade every forward through it to a fail-closed 503.
+ let http_forwarding_active =
+ self.http.enabled && http_forwarding_key_material(&self.http.jwt,
&self.cluster);
+ if http_forwarding_active {
+ for node in &self.cluster.nodes {
+ if node.ports.http.is_none() {
+ eprintln!(
+ "cluster node '{}' has no ports.http; every node needs
one when http.enabled so followers can forward to the primary",
+ node.name
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+ }
+ }
+
if topic_size < self.system.segment.size.as_bytes_u64() {
eprintln!(
"system.topic.max_size ({} B) must be >= system.segment.size
({} B)",
@@ -272,6 +292,7 @@ impl Validatable<ConfigurationError> for NamespaceConfig {
#[cfg(test)]
mod tests {
+ use super::super::cluster::{ClusterNodeConfig, TransportPorts};
use super::*;
use figment::Figment;
use figment::providers::{Format, Toml};
@@ -409,4 +430,78 @@ mod tests {
let cfg = https_config("cert.pem", "key.pem");
assert!(cfg.validate().is_ok());
}
+
+ fn cluster_node(replica_id: u8, http: Option<u16>) -> ClusterNodeConfig {
+ ClusterNodeConfig {
+ name: format!("node-{replica_id}"),
+ ip: "127.0.0.1".to_string(),
+ replica_id,
+ ports: TransportPorts {
+ tcp: Some(8090 + u16::from(replica_id)),
+ quic: None,
+ http,
+ websocket: None,
+ tcp_replica: Some(9090 + u16::from(replica_id)),
+ },
+ }
+ }
+
+ fn clustered_http_config(nodes: Vec<ClusterNodeConfig>) -> ServerNgConfig {
+ let mut cfg = ServerNgConfig::default();
+ cfg.http.enabled = true;
+ cfg.cluster.enabled = true;
+ cfg.cluster.name = "test-cluster".to_string();
+ cfg.cluster.nodes = nodes;
+ cfg
+ }
+
+ // Keyless cluster+http boots: forwarding degrades to off instead of
+ // failing the whole server.
+ #[test]
+ fn validate_accepts_cluster_http_without_jwt_secret_or_cluster_auth() {
+ let cfg = clustered_http_config(vec![
+ cluster_node(0, Some(3000)),
+ cluster_node(1, Some(3001)),
+ ]);
+ assert!(cfg.validate().is_ok());
+ }
+
+ // Without key material forwarding is off, so the roster http-port
+ // requirement does not apply either.
+ #[test]
+ fn validate_accepts_keyless_cluster_http_with_portless_roster_node() {
+ let cfg = clustered_http_config(vec![cluster_node(0, Some(3000)),
cluster_node(1, None)]);
+ assert!(cfg.validate().is_ok());
+ }
+
+ #[test]
+ fn validate_accepts_cluster_http_with_configured_jwt_secrets() {
+ let mut cfg = clustered_http_config(vec![
+ cluster_node(0, Some(3000)),
+ cluster_node(1, Some(3001)),
+ ]);
+ cfg.http.jwt.encoding_secret =
"0123456789abcdef0123456789abcdef".to_string();
+ cfg.http.jwt.decoding_secret =
"0123456789abcdef0123456789abcdef".to_string();
+ assert!(cfg.validate().is_ok());
+ }
+
+ #[test]
+ fn validate_accepts_cluster_http_with_cluster_auth_as_jwt_key_source() {
+ let mut cfg = clustered_http_config(vec![
+ cluster_node(0, Some(3000)),
+ cluster_node(1, Some(3001)),
+ ]);
+ cfg.cluster.auth.enabled = true;
+ cfg.cluster.auth.shared_secret =
"0123456789abcdef0123456789abcdef".to_string();
+ assert!(cfg.validate().is_ok());
+ }
+
+ #[test]
+ fn validate_rejects_cluster_http_when_a_roster_node_has_no_http_port() {
+ let mut cfg =
+ clustered_http_config(vec![cluster_node(0, Some(3000)),
cluster_node(1, None)]);
+ cfg.cluster.auth.enabled = true;
+ cfg.cluster.auth.shared_secret =
"0123456789abcdef0123456789abcdef".to_string();
+ assert!(cfg.validate().is_err());
+ }
}
diff --git a/core/integration/tests/server/http_vsr.rs
b/core/integration/tests/server/http_vsr.rs
index 2018d4ae7..fe4dfc886 100644
--- a/core/integration/tests/server/http_vsr.rs
+++ b/core/integration/tests/server/http_vsr.rs
@@ -18,7 +18,7 @@
//! HTTP data-plane gate for server-ng: produce, poll, and consumer-offset
//! routes exercised over raw `reqwest` (not the SDK HTTP client) so the wire
//! contract itself is under test - exact status codes, the
-//! `x-iggy-durability` header, the body-size cap, and cross-request isolation
+//! `iggy-durability` header, the body-size cap, and cross-request isolation
//! of concurrent produces on one login session.
use crate::server::http_client::HttpClient;
@@ -44,7 +44,7 @@ const PARTITION_ID: u32 = 0;
/// (`Consumer::default()` would carry numeric id 0, not 1).
const CONSUMER_ID: u32 = 1;
-const DURABILITY_HEADER: &str = "x-iggy-durability";
+const DURABILITY_HEADER: &str = "iggy-durability";
const DURABILITY_REPLICATED_MEMORY: &str = "replicated-memory";
const DURABILITY_NONE: &str = "none";
diff --git a/core/server-ng/Cargo.toml b/core/server-ng/Cargo.toml
index b42c2a3cb..6659db887 100644
--- a/core/server-ng/Cargo.toml
+++ b/core/server-ng/Cargo.toml
@@ -97,6 +97,7 @@ async-channel = { workspace = true }
async_zip = { workspace = true }
axum = { workspace = true }
axum-server = { workspace = true }
+blake3 = { workspace = true }
bytemuck = { workspace = true }
bytes = { workspace = true }
chrono = { workspace = true }
diff --git a/core/server-ng/config.toml b/core/server-ng/config.toml
index 832e6b33e..7a185a80d 100644
--- a/core/server-ng/config.toml
+++ b/core/server-ng/config.toml
@@ -34,6 +34,12 @@ interval = "1 m"
# Determines if the HTTP server is active.
# `true` enables the server, allowing it to handle HTTP requests.
# `false` disables the server, preventing it from handling HTTP requests.
+# In cluster mode, followers forward control-plane requests (streams, topics,
+# users, ...) to the current primary when a cluster-wide JWT key exists (see
+# http.jwt / cluster.auth below).
+# TODO: forwarding does not cover the partition-plane APIs yet - message
+# produce and consumer-offset writes are never forwarded and must reach the
+# partition's primary node directly (message polls read locally on any node).
enabled = true
# Specifies the network address and port for the HTTP server.
@@ -75,9 +81,9 @@ allowed_origins = ["*"]
allowed_headers = ["content-type", "authorization"]
# Headers that browsers are allowed to access in CORS responses.
-# `x-iggy-view` carries the current VSR view number; exposing it lets browser
+# `iggy-view` carries the current VSR view number; exposing it lets browser
# clients read it on cross-origin responses.
-exposed_headers = ["x-iggy-view"]
+exposed_headers = ["iggy-view"]
# Determines if credentials like cookies or HTTP auth can be included in CORS
requests.
# `true` allows credentials to be included, useful for authenticated sessions;
@@ -120,6 +126,10 @@ not_before = "0 s"
# Secret key for encoding JWTs.
# If left empty, a secure random secret will be generated on each server start.
+# In cluster mode a configured secret (identical on every node) makes bearers
+# valid cluster-wide and activates follower-to-primary HTTP forwarding; with
+# cluster.auth enabled the key is instead derived from the shared PSK. Without
+# either, tokens are node-local and forwarding stays disabled.
encoding_secret = ""
# Secret key for decoding JWTs.
@@ -589,6 +599,9 @@ heartbeat_timeout = "5s"
# When true, every replica peer must complete the authenticated handshake or be
# rejected, and shared_secret becomes mandatory. Off by default = legacy
# unauthenticated replica traffic. Enabling it is a coordinated-restart change.
+# With http enabled and no http.jwt secrets configured, the PSK also becomes
+# the JWT key source, making bearers valid cluster-wide and activating
+# follower-to-primary HTTP forwarding.
enabled = false
# Cluster-wide pre-shared key, >= 32 bytes of CSPRNG output, byte-identical on
diff --git a/core/server-ng/src/http.rs b/core/server-ng/src/http.rs
index 37a5535b4..0f4668cf3 100644
--- a/core/server-ng/src/http.rs
+++ b/core/server-ng/src/http.rs
@@ -25,6 +25,7 @@
mod admission;
mod error;
mod extractor;
+mod forward;
mod handlers;
mod jwks;
mod jwt;
@@ -48,16 +49,16 @@ use std::sync::atomic::AtomicU64;
use axum::Router;
use axum::extract::{DefaultBodyLimit, Request};
use axum::http::{HeaderName, HeaderValue, Method};
-use axum::middleware::{Next, from_fn};
+use axum::middleware::{Next, from_fn, from_fn_with_state};
use axum::routing::{delete, get, post, put};
use configs::http::{HttpConfig, HttpCorsConfig};
-use configs::ng_cluster::{ClusterConfig, TransportPorts};
+use configs::ng_cluster::{ClusterConfig, TransportPorts,
http_forwarding_key_material};
use configs::server_ng::NgSystemConfig;
use iggy_common::IggyError;
use message_bus::client_listener;
use send_wrapper::SendWrapper;
use tower_http::cors::{AllowOrigin, CorsLayer};
-use tracing::{error, info};
+use tracing::{error, info, warn};
use crate::bootstrap::ServerNgShard;
use crate::cluster_meta::ClusterRoster;
@@ -97,7 +98,28 @@ pub async fn start(
system_config: Arc<NgSystemConfig>,
self_ports: TransportPorts,
) -> Result<(), ServerNgError> {
- let jwt = JwtManager::build(&http_config.jwt)?;
+ // In cluster mode with no configured JWT secret the signing key derives
+ // from the cluster PSK, so a bearer minted on any node verifies on every
+ // node - the invariant follower-to-primary forwarding depends on.
+ let cluster_psk =
+ (cluster.enabled && cluster.auth.enabled &&
!cluster.auth.shared_secret.is_empty())
+ .then_some(cluster.auth.shared_secret.as_str());
+ let jwt = JwtManager::build(&http_config.jwt, cluster_psk)?;
+ // Forwarding needs a bearer every node can verify; without key material it
+ // degrades to off (followers answer the transient 503) instead of failing
+ // the boot, so keyless clusters still serve HTTP node-locally.
+ let forwarding_active = http_forwarding_key_material(&http_config.jwt,
cluster);
+ if cluster.enabled && !forwarding_active {
+ warn!(
+ "cluster mode with http enabled but no http.jwt secrets and no
cluster.auth: bearers are node-local and follower-to-primary forwarding is
disabled - control-plane writes on followers answer a transient 503; configure
http.jwt encoding/decoding secrets or enable cluster.auth (identical on every
node) to activate forwarding"
+ );
+ }
+ // Saturating: a configured limit past the pointer width (32-bit target,
+ // >4 GiB value) clamps to the largest enforceable cap instead of wrapping.
+ let max_request_size =
+
usize::try_from(http_config.max_request_size.as_bytes_u64()).unwrap_or(usize::MAX);
+ let forward =
+ forward::build_forward_state(&http_config.tls, max_request_size,
forwarding_active)?;
// Validated before bind so a bad [http.cors] fails boot before the socket
// opens and the "started" log prints.
let cors = http_config
@@ -130,11 +152,8 @@ pub async fn start(
metadata_view:
Arc::new(AtomicU64::new(crate::cluster_meta::METADATA_VIEW_UNKNOWN)),
},
in_flight_writes: Cell::new(0),
+ forward,
}));
- // Saturating: a configured limit past the pointer width (32-bit target,
- // >4 GiB value) clamps to the largest enforceable cap instead of wrapping.
- let max_request_size =
-
usize::try_from(http_config.max_request_size.as_bytes_u64()).unwrap_or(usize::MAX);
let router = router(state, max_request_size, cors, http_config.web_ui);
if http_config.tls.enabled {
@@ -163,13 +182,21 @@ pub async fn start(
}
/// Health-probe path. Public and pre-auth, and the one success route reached
-/// without proving a credential, so the `X-Iggy-View` layer withholds the
+/// without proving a credential, so the `Iggy-View` layer withholds the
/// cluster-internal view number here (see the response layer below).
const PING_PATH: &str = "/ping";
/// Assemble the shard-0 router: unauthenticated health + login routes plus the
/// authenticated REST surface.
///
+/// The surface is split by consensus dependency. The control-plane routes -
+/// whose writes all commit through the metadata consensus group - carry the
+/// `forward_to_primary` route layer, so on a follower they are relayed to the
+/// primary instead of failing with a transient 503 (reads under that layer
+/// still serve locally unless `?consistency=linearizable`). The local routes
+/// never need the metadata primary: health, the login/refresh flows (STM read
+/// + JWT mint), node-local reads, and the partition-plane routes.
+///
/// `max_request_size` becomes the router-wide `DefaultBodyLimit` (413 past
/// it), exactly like the legacy server: it bounds the per-request term of the
/// admission math - what one body may cost in bytes and decode CPU - while
@@ -181,24 +208,17 @@ const PING_PATH: &str = "/ping";
/// `Authorization` header and matches none of the method routes - would 401 or
/// 405 if it reached the router; the outermost `CorsLayer` answers it first
/// instead, and stamps the CORS response headers over every reply, including
-/// the inner layer's `x-iggy-view`.
+/// the inner layer's `iggy-view`.
fn router(
state: HttpState,
max_request_size: usize,
cors: Option<CorsLayer>,
web_ui: bool,
) -> Router {
- // Cloned for the response layer so `X-Iggy-View` reads the live view per
+ // Cloned for the response layer so `Iggy-View` reads the live view per
// response; the original `state` is moved into `with_state` below.
let view_source = state.clone();
- let router = Router::new()
- .route(PING_PATH, get(ping))
- .route("/users/login", post(login_user))
- .route("/users/refresh-token", post(refresh_token))
- .route(
- "/personal-access-tokens/login",
- post(login_with_personal_access_token),
- )
+ let forwardable = Router::new()
.route("/users", get(get_users).post(create_user))
.route(
"/users/{user_id}",
@@ -235,6 +255,35 @@ fn router(
"/streams/{stream_id}/topics/{topic_id}/partitions/{partition_id}",
delete(delete_segments),
)
+ .route(
+ "/streams/{stream_id}/topics/{topic_id}/consumer-groups",
+ get(get_cgs).post(create_cg),
+ )
+ .route(
+
"/streams/{stream_id}/topics/{topic_id}/consumer-groups/{group_id}",
+ get(get_cg).delete(delete_cg),
+ )
+ .route("/personal-access-tokens", get(get_pats).post(create_pat))
+ .route("/personal-access-tokens/{name}", delete(delete_pat))
+ .route_layer(from_fn_with_state(
+ state.clone(),
+ forward::forward_to_primary,
+ ));
+ // The partition-plane routes (produce, consumer-offset writes) stay local:
+ // each partition is its own consensus group whose primary can diverge from
+ // the metadata primary, so forwarding them to the metadata primary would
+ // livelock whenever the two disagree.
+ // TODO: forward partition-plane writes to their own partition group's
+ // primary (requires resolving the target partition from the request before
+ // dispatch, and rewriting balanced partitioning to an explicit partition).
+ let local = Router::new()
+ .route(PING_PATH, get(ping))
+ .route("/users/login", post(login_user))
+ .route("/users/refresh-token", post(refresh_token))
+ .route(
+ "/personal-access-tokens/login",
+ post(login_with_personal_access_token),
+ )
.route(
"/streams/{stream_id}/topics/{topic_id}/messages",
get(poll_messages).post(send_messages),
@@ -247,21 +296,14 @@ fn router(
"/streams/{stream_id}/topics/{topic_id}/consumer-offsets/{consumer_id}",
delete(delete_consumer_offset),
)
- .route(
- "/streams/{stream_id}/topics/{topic_id}/consumer-groups",
- get(get_cgs).post(create_cg),
- )
- .route(
-
"/streams/{stream_id}/topics/{topic_id}/consumer-groups/{group_id}",
- get(get_cg).delete(delete_cg),
- )
- .route("/personal-access-tokens", get(get_pats).post(create_pat))
- .route("/personal-access-tokens/{name}", delete(delete_pat))
.route("/stats", get(get_stats))
.route("/snapshot", post(get_snapshot))
.route("/cluster/metadata", get(get_cluster_metadata))
.route("/clients", get(get_clients))
- .route("/clients/{client_id}", get(get_client))
+ .route("/clients/{client_id}", get(get_client));
+ let router = Router::new()
+ .merge(forwardable)
+ .merge(local)
.with_state(state)
.layer(DefaultBodyLimit::max(max_request_size))
.layer(from_fn(move |request: Request, next: Next| {
@@ -452,7 +494,7 @@ mod tests {
allowed_methods: vec!["GET".to_owned(), "POST".to_owned()],
allowed_origins: vec!["*".to_owned()],
allowed_headers: vec!["content-type".to_owned(),
"authorization".to_owned()],
- exposed_headers: vec!["x-iggy-view".to_owned()],
+ exposed_headers: vec!["iggy-view".to_owned()],
allow_credentials: false,
allow_private_network: false,
}
diff --git a/core/server-ng/src/http/error.rs b/core/server-ng/src/http/error.rs
index 63fbe935a..511a1dd22 100644
--- a/core/server-ng/src/http/error.rs
+++ b/core/server-ng/src/http/error.rs
@@ -84,7 +84,19 @@ impl IntoResponse for CustomError {
}
_ => StatusCode::BAD_REQUEST,
};
- (status_code, Json(ErrorResponse::from_error(&error)))
+ let response =
+ (status_code,
Json(ErrorResponse::from_error(&error))).into_response();
+ // Transient 503s are retryable, so the advisory Retry-After
hint
+ // rides along, matching the other transient 503 bodies
+ // (`service_unavailable`, `server_busy`).
+ if matches!(
+ error,
+ IggyError::TransientNotCommitted |
IggyError::TransientNotAccepted
+ ) {
+ with_retry_after(response)
+ } else {
+ response
+ }
}
Self::ResourceNotFound => (
StatusCode::NOT_FOUND,
@@ -94,9 +106,9 @@ impl IntoResponse for CustomError {
reason: "Resource not found".to_string(),
field: None,
}),
- ),
+ )
+ .into_response(),
}
- .into_response()
}
}
@@ -138,8 +150,15 @@ impl ErrorResponse {
/// body every other route error uses), while a VSR session that cannot be
/// established right now is a transient server condition (503) and must never
/// masquerade as an auth failure.
-pub enum AuthError {
+pub(in crate::http) enum AuthError {
Unauthenticated(IggyError),
+ /// The Register provably never entered the consensus pipeline (not
+ /// primary, not caught up, or the prepare queue was full), so the request
+ /// is safe to re-issue anywhere. Rendered with the `TransientNotAccepted`
+ /// body so a forwarding follower recognizes it as retryable against a
+ /// re-resolved primary; a plain client sees the same retryable 503 either
+ /// way.
+ SessionNotAccepted,
SessionUnavailable,
}
@@ -158,11 +177,15 @@ impl IntoResponse for AuthError {
// JWT middleware (empty body), so this is deliberately richer, not
// byte-identical to legacy.
Self::Unauthenticated(error) =>
CustomError::from(error).into_response(),
- // A fresh session could not be established: the Register did not
- // commit (no caught-up primary, pipeline full, or a view-change
- // cancel), or the session table is at `MAX_HTTP_SESSIONS` and
- // refused the fresh registration. Transient server condition ->
503,
- // retryable.
+ Self::SessionNotAccepted => {
+
CustomError::from(IggyError::TransientNotAccepted).into_response()
+ }
+ // A fresh session could not be established: the Register was
+ // canceled with its commit outcome unknown, or the session table
+ // is at `MAX_HTTP_SESSIONS` and refused the fresh registration.
+ // Transient server condition -> 503, retryable by the CLIENT only
+ // (a forwarder must not re-issue an unknown-outcome Register under
+ // this node's session budget on the caller's behalf).
Self::SessionUnavailable => service_unavailable(),
}
}
@@ -276,7 +299,7 @@ fn partition_write_timeout_response(operation: Operation)
-> Response {
const RETRY_AFTER_SECONDS: u64 = 1;
/// Attach the advisory [`RETRY_AFTER_SECONDS`] hint to a retryable 429/503.
-fn with_retry_after(mut response: Response) -> Response {
+pub(in crate::http) fn with_retry_after(mut response: Response) -> Response {
response
.headers_mut()
.insert(RETRY_AFTER, HeaderValue::from(RETRY_AFTER_SECONDS));
@@ -286,7 +309,7 @@ fn with_retry_after(mut response: Response) -> Response {
/// Render an `ErrorResponse` body for `status`, tagged with `code` / `reason`
/// and no field, so every hand-built HTTP error the routes return parses as
the
/// one error schema clients already handle.
-fn error_response(status: StatusCode, code: &str, reason: &str) -> Response {
+pub(in crate::http) fn error_response(status: StatusCode, code: &str, reason:
&str) -> Response {
(
status,
Json(ErrorResponse {
@@ -301,9 +324,9 @@ fn error_response(status: StatusCode, code: &str, reason:
&str) -> Response {
/// Shared 504 rendering for an in-band request the partition plane did not
/// answer in time, shaped like every other HTTP error (`ErrorResponse`) so
-/// clients parse one error schema. Consumed by the partition-write reply wait
-/// and the partition reads ([`ReadError::Timeout`]).
-fn gateway_timeout_response(code: &str, reason: &str) -> Response {
+/// clients parse one error schema. Consumed by the partition-write reply wait,
+/// the partition reads ([`ReadError::Timeout`]), and the forward attempt
bound.
+pub(in crate::http) fn gateway_timeout_response(code: &str, reason: &str) ->
Response {
error_response(StatusCode::GATEWAY_TIMEOUT, code, reason)
}
@@ -465,26 +488,38 @@ fn primary_redirect_response(location: &str) -> Response {
}
/// Build the `Location` for a 307 redirect of a linearizable read to the VSR
-/// primary: `http://<host>:<http-port><path_and_query>`. `None` when the
roster
-/// has no node at `primary_index`, that node exposes no HTTP port, or its `ip`
-/// is not a valid address, so the caller fails closed to a 503 rather than
-/// pointing at an unreachable target. Formats through [`SocketAddr`] so an
IPv6
-/// host is bracketed (`http://[::1]:8080/...`) rather than left ambiguous.
Pure
-/// (no consensus or axum dependency) so the redirect target is unit-tested in
-/// isolation.
+/// 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.
pub(in crate::http) fn primary_redirect_location(
roster: &ClusterRoster,
primary_index: u8,
+ scheme: &str,
path_and_query: &str,
) -> Option<String> {
+ let socket = primary_http_socket(roster, primary_index)?;
+ Some(format!("{scheme}://{socket}{path_and_query}"))
+}
+
+/// Resolve the VSR primary's HTTP socket from the static roster: the node
+/// whose `replica_id` equals `primary_index`, its `ports.http`, and its `ip`
+/// parsed strictly. `None` on any miss so callers fail closed. Formatting the
+/// returned [`SocketAddr`] brackets an IPv6 host (`[::1]:8080`) rather than
+/// leaving it ambiguous.
+pub(in crate::http) fn primary_http_socket(
+ roster: &ClusterRoster,
+ primary_index: u8,
+) -> Option<SocketAddr> {
let node = roster
.nodes
.iter()
.find(|node| node.replica_id == primary_index)?;
let http_port = node.ports.http?;
let ip = node.ip.parse::<IpAddr>().ok()?;
- let socket = SocketAddr::new(ip, http_port);
- Some(format!("http://{socket}{path_and_query}";))
+ Some(SocketAddr::new(ip, http_port))
}
#[cfg(test)]
@@ -529,35 +564,74 @@ mod tests {
node(1, "10.0.0.2", Some(8090)),
]);
assert_eq!(
- primary_redirect_location(&roster, 1, READ_PATH),
+ primary_redirect_location(&roster, 1, "http", READ_PATH),
Some("http://10.0.0.2:8090/streams?consistency=linearizable".to_owned())
);
}
+ #[test]
+ 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),
+
Some("https://10.0.0.1:8080/streams?consistency=linearizable".to_owned())
+ );
+ }
+
#[test]
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, READ_PATH), None);
+ assert_eq!(
+ primary_redirect_location(&roster, 2, "http", READ_PATH),
+ None
+ );
}
#[test]
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, READ_PATH), None);
+ assert_eq!(
+ primary_redirect_location(&roster, 0, "http", READ_PATH),
+ None
+ );
}
#[test]
fn primary_redirect_location_is_none_for_empty_roster() {
let roster = roster(Vec::new());
- assert_eq!(primary_redirect_location(&roster, 0, READ_PATH), None);
+ assert_eq!(
+ primary_redirect_location(&roster, 0, "http", READ_PATH),
+ None
+ );
}
#[test]
fn primary_redirect_location_brackets_ipv6_host() {
let roster = roster(vec![node(0, "::1", Some(8080))]);
assert_eq!(
- primary_redirect_location(&roster, 0, READ_PATH),
+ primary_redirect_location(&roster, 0, "http", READ_PATH),
Some("http://[::1]:8080/streams?consistency=linearizable".to_owned())
);
}
+
+ #[test]
+ fn transient_not_committed_renders_503_with_retry_after() {
+ let response =
CustomError::from(IggyError::TransientNotCommitted).into_response();
+ assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
+ assert!(response.headers().contains_key(RETRY_AFTER));
+ }
+
+ #[test]
+ fn transient_not_accepted_renders_503_with_retry_after() {
+ let response =
CustomError::from(IggyError::TransientNotAccepted).into_response();
+ assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
+ assert!(response.headers().contains_key(RETRY_AFTER));
+ }
+
+ #[test]
+ fn business_error_renders_without_retry_after() {
+ let response =
CustomError::from(IggyError::UserAlreadyExists).into_response();
+ assert_eq!(response.status(), StatusCode::BAD_REQUEST);
+ assert!(!response.headers().contains_key(RETRY_AFTER));
+ }
}
diff --git a/core/server-ng/src/http/extractor.rs
b/core/server-ng/src/http/extractor.rs
index bcd523658..eb52a8321 100644
--- a/core/server-ng/src/http/extractor.rs
+++ b/core/server-ng/src/http/extractor.rs
@@ -20,6 +20,7 @@
use std::rc::Rc;
use axum::extract::FromRequestParts;
+use axum::http::HeaderMap;
use axum::http::header::AUTHORIZATION;
use axum::http::request::Parts;
use iggy_common::{IggyError, PersonalAccessToken};
@@ -59,7 +60,7 @@ impl FromRequestParts<HttpState> for Authenticated {
parts: &mut Parts,
state: &HttpState,
) -> Result<Self, Self::Rejection> {
- let bearer = bearer_token(parts)?;
+ let bearer = bearer_token(&parts.headers)?;
// Both `resolve_credential` (its JWT verify may await a `!Send` JWKS
// fetch through cyper) and `resolve_session` (`Rc`-based, `!Send`)
must
@@ -104,7 +105,7 @@ impl FromRequestParts<HttpState> for Identity {
parts: &mut Parts,
state: &HttpState,
) -> Result<Self, Self::Rejection> {
- let bearer = bearer_token(parts)?;
+ let bearer = bearer_token(&parts.headers)?;
// Verify only. The session key and expiry `resolve_credential` also
// returns feed the write path's session table; a read discards them.
@@ -129,10 +130,11 @@ impl FromRequestParts<HttpState> for Identity {
/// Extract the raw bearer token from `Authorization: Bearer <token>`, or
reject
/// as `AccessTokenMissing` (the 401 both extractors share). The `?` at each
call
-/// site converts the `IggyError` into the extractor's `AuthError`.
-fn bearer_token(parts: &Parts) -> Result<&str, IggyError> {
- parts
- .headers
+/// site converts the `IggyError` into the extractor's `AuthError`. Takes the
+/// header map (not `Parts`) so the forward middleware, which holds a full
+/// `Request`, runs the identical extraction.
+pub(in crate::http) fn bearer_token(headers: &HeaderMap) -> Result<&str,
IggyError> {
+ headers
.get(AUTHORIZATION)
.and_then(|value| value.to_str().ok())
.and_then(|value| value.strip_prefix(BEARER))
@@ -147,8 +149,10 @@ fn bearer_token(parts: &Parts) -> Result<&str, IggyError> {
/// the key is stable and collision-free while the raw secret never enters the
/// table. The JWT verify is `async` and `!Send` (a trusted-issuer token may
/// fetch the issuer's JWKS on a cache miss), so both call sites drive it
inside
-/// a `SendWrapper`; the PAT check is local and synchronous.
-async fn resolve_credential(
+/// a `SendWrapper`; the PAT check is local and synchronous. Also the forward
+/// middleware's verify-only gate: a follower proves the bearer before relaying
+/// a request, without minting a session.
+pub(in crate::http) async fn resolve_credential(
state: &HttpState,
bearer: &str,
) -> Result<(String, u32, u64), AuthError> {
diff --git a/core/server-ng/src/http/forward.rs
b/core/server-ng/src/http/forward.rs
new file mode 100644
index 000000000..08f8608eb
--- /dev/null
+++ b/core/server-ng/src/http/forward.rs
@@ -0,0 +1,632 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Follower-side forwarding of consensus-needing requests to the VSR
+//! metadata primary.
+//!
+//! A load balancer that round-robins HTTP across the cluster lands writes and
+//! linearizable reads on followers, which cannot serve them: a follower can
+//! neither Register a VSR session nor commit a control-plane op. Instead of
+//! failing those requests with a transient 503, the middleware here re-issues
+//! them against the current primary's HTTP listener and relays the primary's
+//! response on the original connection, so any node answers any request.
+//!
+//! Scope: the middleware is attached (via `route_layer`) only to the
+//! control-plane routes, whose ops all commit through the metadata consensus
+//! group and therefore share one forward target. Partition-plane writes
+//! (produce, consumer-offset writes) are excluded: each partition is its own
+//! consensus group whose primary can diverge from the metadata primary, so
+//! forwarding them needs per-group target resolution.
+//!
+//! Safety model, in order:
+//! - The bearer is verified locally (verify-only, no session mint) before any
+//! bytes leave this node, so an unauthenticated caller cannot make a
+//! follower relay junk to the primary. A PAT is checked against this node's
+//! local metadata view, so a PAT minted on the primary and not yet
+//! replicated here answers 401 until replication catches up - fail-closed
+//! on purpose, the price of keeping the pre-forward auth gate.
+//! - The primary re-authenticates and re-authorizes the forwarded request
+//! through its ordinary extractor stack; the forward marker header is a loop
+//! guard only and never a trust input.
+//! - A forwarded request is retried only when it provably never entered the
+//! primary's pipeline: a connect-phase failure, a 503 whose body carries the
+//! `TransientNotAccepted` code, or a 307 from a stale target. Every other
+//! outcome - including a 503 carrying `TransientNotCommitted`, whose op may
+//! still commit - is relayed as-is, because re-issuing it under a fresh
+//! session would defeat the consensus dedup and double-apply the op.
+//! - The target is always resolved from the local roster + consensus view,
+//! never from a response `Location`, and the client follows no redirects.
+
+use std::cell::Cell;
+use std::net::SocketAddr;
+use std::path::Path;
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+use axum::body::{Body, to_bytes};
+use axum::extract::{Request, State};
+use axum::http::header::{AUTHORIZATION, CONTENT_TYPE, RETRY_AFTER};
+use axum::http::request::Parts;
+use axum::http::{HeaderMap, HeaderName, HeaderValue, Method, StatusCode};
+use axum::middleware::Next;
+use axum::response::{IntoResponse, Response};
+use bytes::Bytes;
+use configs::http::HttpTlsConfig;
+use consensus::MetadataHandle;
+use futures::StreamExt;
+use iggy_common::IggyError;
+use message_bus::transports::tls::{install_default_crypto_provider, load_pem};
+use rustls::client::danger::{HandshakeSignatureValid, ServerCertVerified,
ServerCertVerifier};
+use rustls::crypto::{CryptoProvider, WebPkiSupportedAlgorithms};
+use rustls::pki_types::{CertificateDer, ServerName, UnixTime};
+use rustls::{CertificateError, DigitallySignedStruct, SignatureScheme};
+use send_wrapper::SendWrapper;
+use serde::Deserialize;
+use tracing::{debug, warn};
+
+use crate::http::HttpState;
+use crate::http::error::{
+ CustomError, error_response, gateway_timeout_response,
primary_http_socket, with_retry_after,
+};
+use crate::http::extractor::{bearer_token, resolve_credential};
+use crate::http::state::{HttpInner, VIEW_HEADER};
+use crate::server_error::ServerNgError;
+
+/// Marker stamped on every forwarded request. Loop guard only: a node that is
+/// not primary and sees it answers the transient 503 instead of forwarding
+/// again, so a stale view can never chain hops. It is client-spoofable by
+/// design - spoofing it at a follower is a self-inflicted 503, and the primary
+/// ignores it - and it must never gate auth, authz, or admission.
+const FORWARDED_HEADER: HeaderName = HeaderName::from_static("iggy-forwarded");
+const FORWARDED_VALUE: HeaderValue = HeaderValue::from_static("1");
+
+/// Wall-clock bound on one forward attempt (send + primary processing + body
+/// read). Above the primary's own 30s in-flight transient replay budget, so a
+/// legitimately slow commit is answered rather than cut mid-flight; without
+/// this cap a hung primary would park the connection for the whole retry
+/// budget with no per-attempt bound (the HTTP client itself has no timeout).
+const FORWARD_ATTEMPT_TIMEOUT: Duration = Duration::from_secs(35);
+
+/// Cadence between retryable attempts, mirroring the binary SDKs' in-client
+/// replay loop and the local submit path's transient replay interval.
+const FORWARD_RETRY_INTERVAL: Duration = Duration::from_millis(50);
+
+/// Budget across retryable attempts. Sized to ride out a full view change
+/// (detection is up to the heartbeat timeout, default 5s, plus election
+/// rounds) a few times over; on exhaustion the caller gets a retryable 503.
+const FORWARD_RETRY_DEADLINE: Duration = Duration::from_secs(30);
+
+/// Cap on concurrent forwards held by this node. Deliberately its own budget:
+/// a forward parks no reply slot and touches none of the shard bus machinery
+/// the partition-write caps protect, and counting forwards against those caps
+/// would let a slow primary starve this node's own direct clients.
+const MAX_IN_FLIGHT_FORWARDS: u32 = 128;
+
+/// Bound on a relayed response body, enforced twice: against a declared
+/// `content-length` before the read, and as a running cap on the streamed
+/// bytes so a length-less reply is bounded too. Forwarded routes answer
+/// entity JSON, not message batches (poll and snapshot are served locally),
+/// so this is a backstop, not a working limit.
+const RESPONSE_BODY_LIMIT: usize = 64 * 1024 * 1024;
+
+/// Pre-allocation hint cap for a relayed body: honest control-plane replies
+/// fit well under this, and a mis-declared content-length must not reserve
+/// the full [`RESPONSE_BODY_LIMIT`] up front.
+const RESPONSE_CAPACITY_HINT: usize = 64 * 1024;
+
+/// Response headers copied from the primary's reply. Everything else is
+/// dropped, which subsumes the RFC 7230 hop-by-hop set: the relayed response
+/// is rebuilt, never streamed, so upstream `connection` / `transfer-encoding`
+/// semantics cannot leak to the client. `iggy-view` is included so the
+/// relayed response carries the serving primary's view, not this follower's
+/// (the view layer only fills the header when absent).
+const RELAYED_RESPONSE_HEADERS: [HeaderName; 3] = [CONTENT_TYPE, RETRY_AFTER,
VIEW_HEADER];
+
+/// Per-node forwarding context hung off `HttpInner`: the outbound client
+/// (pinned-cert TLS when the listener serves HTTPS), the scheme it dials, the
+/// request-body buffer bound, and the in-flight budget.
+pub(in crate::http) struct ForwardState {
+ /// False when no cluster-wide bearer key material exists (no configured
+ /// JWT secret, no cluster PSK): a forwarded bearer would 401 on the
+ /// primary, so the middleware passes through and followers answer with
+ /// the transient 503 instead.
+ active: bool,
+ client: cyper::Client,
+ /// Also read by the 307 redirect builder: the primary is assumed to serve
+ /// the same scheme as this node (uniform cluster HTTP config).
+ pub(in crate::http) scheme: &'static str,
+ body_limit: usize,
+ in_flight: Cell<u32>,
+}
+
+/// Build the [`ForwardState`] at listener startup.
+///
+/// With `http.tls.enabled` the forward hop dials `https` and verifies the peer
+/// against this node's OWN certificate chain (exact-DER pin): cluster nodes
+/// are expected to share the HTTP certificate, and a fixed-roster deployment
+/// makes pinning strictly stronger than name-based verification against
+/// config-listed IPs. Per-node distinct certificates fail closed at the
+/// handshake. Plaintext deployments dial plain `http`, which puts the relayed
+/// bearer on the node-to-node link exactly as exposed as it already is on the
+/// client-to-node link - TLS is the remedy for both.
+///
+/// # Errors
+///
+/// [`ServerNgError::ListenerCredentials`] when TLS is enabled but the PEM
+/// files cannot be loaded, [`ServerNgError::HttpForwardClient`] when the
+/// outbound client cannot be built.
+pub(in crate::http) fn build_forward_state(
+ tls: &HttpTlsConfig,
+ body_limit: usize,
+ active: bool,
+) -> Result<ForwardState, ServerNgError> {
+ // Unconditional: cyper's rustls connector resolves the process default
+ // provider even when the client only ever dials plain http.
+ install_default_crypto_provider();
+ let builder = cyper::Client::builder()
+ // The retry loop re-resolves the primary from the local roster; a
+ // followed `Location` would let the peer steer the bearer anywhere.
+ .redirect(cyper::redirect::Policy::none());
+ let (builder, scheme) = if tls.enabled {
+ let credentials =
+ load_pem(Path::new(&tls.cert_file),
Path::new(&tls.key_file)).map_err(|source| {
+ ServerNgError::ListenerCredentials {
+ transport: "http.tls",
+ source,
+ }
+ })?;
+ // load_pem guarantees a non-empty chain; this is the no-panic
+ // path for the unreachable empty case.
+ let pinned = credentials.cert_chain.into_iter().next().ok_or_else(|| {
+ ServerNgError::HttpForwardClient {
+ reason: "TLS certificate chain is empty".to_string(),
+ }
+ })?;
+ let algorithms = CryptoProvider::get_default()
+ // Installed above; absence is unreachable.
+ .expect("default crypto provider installed")
+ .signature_verification_algorithms;
+ let config = rustls::ClientConfig::builder()
+ .dangerous()
+ .with_custom_certificate_verifier(Arc::new(PinnedCertVerifier {
pinned, algorithms }))
+ .with_no_client_auth();
+ (builder.use_rustls(Arc::new(config)), "https")
+ } else {
+ // Explicit config even for plain http: cyper's implicit rustls
+ // backend eagerly loads the system CA store at build() and fails
+ // boot on CA-less hosts (minimal container images), although this
+ // client never dials https. Empty roots skip that load and keep an
+ // accidental https dial fail-closed.
+ let config = rustls::ClientConfig::builder()
+ .with_root_certificates(rustls::RootCertStore::empty())
+ .with_no_client_auth();
+ (builder.use_rustls(Arc::new(config)), "http")
+ };
+ let client = builder
+ .build()
+ .map_err(|source| ServerNgError::HttpForwardClient {
+ reason: source.to_string(),
+ })?;
+ Ok(ForwardState {
+ active,
+ client,
+ scheme,
+ body_limit,
+ in_flight: Cell::new(0),
+ })
+}
+
+/// Route-layer middleware for the control-plane routes: pass through on the
+/// primary and for local (non-linearizable) reads, otherwise forward to the
+/// primary and relay its response.
+///
+/// The `!Send` internals (roster/consensus reads, the compio-bound HTTP
+/// client) are bridged with `SendWrapper` exactly like every handler: sound
+/// because the listener pins all of this to shard 0's thread.
+pub(in crate::http) async fn forward_to_primary(
+ State(state): State<HttpState>,
+ request: Request,
+ next: Next,
+) -> Response {
+ SendWrapper::new(forward_or_pass(state, request, next)).await
+}
+
+async fn forward_or_pass(state: HttpState, request: Request, next: Next) ->
Response {
+ if !state.forward.active || state.is_metadata_primary() {
+ return next.run(request).await;
+ }
+ // Reads default to the local STM and stay on this node; only an explicit
+ // linearizable read must reach the primary. An encoded or malformed
+ // `consistency` value falls through to the handler, whose own gate still
+ // answers 307/503, so a miss here degrades, never breaks.
+ if request.method() == Method::GET &&
!wants_linearizable(request.uri().query()) {
+ return next.run(request).await;
+ }
+ if request.headers().contains_key(FORWARDED_HEADER) {
+ // One hop max. The peer that forwarded here re-resolves the primary
+ // and retries; the transient body code tells it the request never
+ // entered any pipeline.
+ return
CustomError::from(IggyError::TransientNotAccepted).into_response();
+ }
+ // Verify-only auth gate (no VSR session mint): a garbage bearer dies here
+ // instead of being buffered and relayed, so unauthenticated traffic cannot
+ // use followers to amplify load onto the primary. The primary still runs
+ // its full extractor on what arrives.
+ let bearer = match bearer_token(request.headers()) {
+ Ok(bearer) => bearer,
+ Err(error) => return CustomError::from(error).into_response(),
+ };
+ if let Err(rejection) = resolve_credential(&state, bearer).await {
+ return rejection.into_response();
+ }
+ let Some(_guard) = ForwardGuard::admit(&state.forward.in_flight) else {
+ return with_retry_after(error_response(
+ StatusCode::SERVICE_UNAVAILABLE,
+ "forward_busy",
+ "node is at its forward budget; retry with backoff",
+ ));
+ };
+ forward(&state, request).await
+}
+
+/// Buffer the request and drive forward attempts until one yields a relayable
+/// outcome or the retry budget runs out.
+async fn forward(state: &HttpInner, request: Request) -> Response {
+ let (parts, body) = request.into_parts();
+ // The router-wide `DefaultBodyLimit` only annotates the request; it is
+ // enforced by whoever consumes the body, so the bound is passed explicitly
+ // here or the buffer would be unbounded.
+ let Ok(body) = to_bytes(body, state.forward.body_limit).await else {
+ return error_response(
+ StatusCode::PAYLOAD_TOO_LARGE,
+ "payload_too_large",
+ "request body exceeds http.max_request_size",
+ );
+ };
+ let path_and_query = parts
+ .uri
+ .path_and_query()
+ .map_or("/", |path_and_query| path_and_query.as_str());
+ let deadline = Instant::now() + FORWARD_RETRY_DEADLINE;
+ loop {
+ let outcome = match primary_socket(state) {
+ // No resolvable primary (mid-election, or a roster hole): count it
+ // as a retryable attempt so a completing election is picked up.
+ None => AttemptOutcome::Retry,
+ Some(socket) => {
+ let url = format!("{}://{socket}{path_and_query}",
state.forward.scheme);
+ attempt(state, &parts, &body, &url).await
+ }
+ };
+ match outcome {
+ AttemptOutcome::Relay(response) => return response,
+ AttemptOutcome::Retry => {
+ let remaining =
deadline.saturating_duration_since(Instant::now());
+ if remaining.is_zero() {
+ warn!(
+ path = parts.uri.path(),
+ "forward retry budget exhausted without a reachable
primary"
+ );
+ return with_retry_after(error_response(
+ StatusCode::SERVICE_UNAVAILABLE,
+ "no_reachable_primary",
+ "no primary accepted the request within the forward
budget; retry",
+ ));
+ }
+
compio::time::sleep(FORWARD_RETRY_INTERVAL.min(remaining)).await;
+ }
+ }
+ }
+}
+
+enum AttemptOutcome {
+ /// Terminal: hand this response to the client.
+ Relay(Response),
+ /// The request provably never entered a pipeline; re-resolve and retry.
+ Retry,
+}
+
+/// Run one forward attempt end to end (connect, send, read the full reply)
+/// under [`FORWARD_ATTEMPT_TIMEOUT`].
+async fn attempt(state: &HttpInner, parts: &Parts, body: &Bytes, url: &str) ->
AttemptOutcome {
+ let builder = match state.forward.client.request(parts.method.clone(),
url) {
+ Ok(builder) => builder,
+ Err(error) => {
+ warn!(%error, "forward request build failed");
+ return AttemptOutcome::Relay(bad_gateway());
+ }
+ };
+ let request = builder
+ .headers(forwarded_headers(&parts.headers))
+ .body(body.clone())
+ .build();
+ let attempt = async {
+ let response = match state.forward.client.execute(request).await {
+ Ok(response) => response,
+ Err(error) => return classify_transport_error(&error),
+ };
+ let status = response.status();
+ // Only the relayed subset survives; the response is consumed by the
+ // body stream below, so the values are pulled out first.
+ let relayed_headers: Vec<(HeaderName, HeaderValue)> =
RELAYED_RESPONSE_HEADERS
+ .into_iter()
+ .filter_map(|name| {
+ let value = response.headers().get(&name)?.clone();
+ Some((name, value))
+ })
+ .collect();
+ let declared = response.content_length();
+ if declared.is_some_and(|length| length > RESPONSE_BODY_LIMIT as u64) {
+ warn!(?declared, "relayed response exceeds the body bound");
+ return AttemptOutcome::Relay(bad_gateway());
+ }
+ // Streamed with a running cap so a length-less reply is bounded by
+ // the limit, not merely by the attempt timeout. The capacity hint is
+ // clamped to RESPONSE_CAPACITY_HINT so a mis-declared content-length
+ // cannot pre-reserve the full bound. The running cap still bounds the
+ // real total.
+ let mut body = Vec::with_capacity(
+ declared
+ .and_then(|length| usize::try_from(length).ok())
+ .unwrap_or(0)
+ .min(RESPONSE_CAPACITY_HINT),
+ );
+ let mut stream = response.bytes_stream();
+ while let Some(chunk) = stream.next().await {
+ let chunk = match chunk {
+ Ok(chunk) => chunk,
+ Err(error) => {
+ warn!(%error, "forward response body read failed; outcome
unknown");
+ return AttemptOutcome::Relay(bad_gateway());
+ }
+ };
+ if body.len() + chunk.len() > RESPONSE_BODY_LIMIT {
+ warn!(
+ received = body.len() + chunk.len(),
+ "relayed response exceeds the body bound"
+ );
+ return AttemptOutcome::Relay(bad_gateway());
+ }
+ body.extend_from_slice(&chunk);
+ }
+ classify_reply(status, relayed_headers, Bytes::from(body))
+ };
+ match compio::time::timeout(FORWARD_ATTEMPT_TIMEOUT, attempt).await {
+ // Elapsed: the request may be mid-commit on the primary. Outcome
+ // unknown, so never retried - 504, same contract as a local commit
+ // wait that timed out.
+ Err(_elapsed) => AttemptOutcome::Relay(gateway_timeout_response(
+ "forward_timeout",
+ "the primary did not answer the forwarded request in time; the
outcome is unknown",
+ )),
+ Ok(outcome) => outcome,
+ }
+}
+
+/// Copy the forwardable request headers: the bearer (the primary
+/// re-authenticates it) and the content type. Everything else - including any
+/// client-supplied forward marker, which `forward_or_pass` already bounced -
+/// is dropped, then the loop-guard marker is stamped fresh.
+fn forwarded_headers(request_headers: &HeaderMap) -> HeaderMap {
+ let mut headers = HeaderMap::new();
+ for name in [AUTHORIZATION, CONTENT_TYPE] {
+ if let Some(value) = request_headers.get(&name) {
+ headers.insert(name, value.clone());
+ }
+ }
+ headers.insert(FORWARDED_HEADER, FORWARDED_VALUE);
+ headers
+}
+
+/// Grade a transport-level failure. Only a connect-phase error - the request
+/// was never written - may retry; anything later (reset mid-request, a broken
+/// body read) leaves the outcome unknown and must surface, because the
+/// primary may have committed the op and a re-issue would double-apply it.
+/// A pooled connection that dies before the request is written also lands in
+/// the relay arm (hyper exposes no sound never-sent predicate): that spurious
+/// 502 is the safe side, and hyper's own canceled-request retry for buffered
+/// bodies absorbs most of it.
+fn classify_transport_error(error: &cyper::Error) -> AttemptOutcome {
+ if let cyper::Error::HyperClient(client_error) = error
+ && client_error.is_connect()
+ {
+ debug!(%error, "forward connect failed; re-resolving primary");
+ return AttemptOutcome::Retry;
+ }
+ warn!(%error, "forward transport error after connect; outcome unknown");
+ AttemptOutcome::Relay(bad_gateway())
+}
+
+/// Grade a complete reply from the target.
+///
+/// A 307 means the target itself was not primary and knows a better one; the
+/// retry re-resolves from the LOCAL view instead of trusting the `Location`.
+/// A 503 is retried only when its body carries the `TransientNotAccepted`
+/// code (never entered a pipeline; also what the hop guard answers) - a
+/// `TransientNotCommitted` 503 may still commit and is relayed untouched.
+/// Everything else is the primary's answer and is relayed.
+fn classify_reply(
+ status: StatusCode,
+ relayed_headers: Vec<(HeaderName, HeaderValue)>,
+ body: Bytes,
+) -> AttemptOutcome {
+ if status == StatusCode::TEMPORARY_REDIRECT {
+ return AttemptOutcome::Retry;
+ }
+ if status == StatusCode::SERVICE_UNAVAILABLE &&
is_transient_not_accepted_body(&body) {
+ return AttemptOutcome::Retry;
+ }
+ let mut response = Response::new(Body::from(body));
+ *response.status_mut() = status;
+ for (name, value) in relayed_headers {
+ response.headers_mut().insert(name, value);
+ }
+ AttemptOutcome::Relay(response)
+}
+
+/// True when a 503 body is the JSON `ErrorResponse` whose `id` is the
+/// `TransientNotAccepted` code. Unparsable or foreign bodies are NOT
+/// transient: when in doubt the reply is relayed, never retried.
+fn is_transient_not_accepted_body(body: &[u8]) -> bool {
+ #[derive(Deserialize)]
+ struct ErrorId {
+ id: u32,
+ }
+ serde_json::from_slice::<ErrorId>(body)
+ .is_ok_and(|error| error.id ==
IggyError::TransientNotAccepted.as_code())
+}
+
+/// HTTP socket of the current metadata primary, from the live consensus view
+/// and the static roster. `None` mid-election or when the roster has no HTTP
+/// address for the primary.
+fn primary_socket(state: &HttpInner) -> Option<SocketAddr> {
+ let consensus = state.shard.plane.metadata().consensus.as_ref()?;
+ let primary_index = consensus.primary_index(consensus.view());
+ primary_http_socket(&state.roster, primary_index)
+}
+
+fn wants_linearizable(query: Option<&str>) -> bool {
+ query.is_some_and(|query| {
+ query
+ .split('&')
+ .any(|pair| pair == "consistency=linearizable")
+ })
+}
+
+fn bad_gateway() -> Response {
+ error_response(
+ StatusCode::BAD_GATEWAY,
+ "forward_failed",
+ "forwarding to the primary failed after the request was sent; the
outcome is unknown",
+ )
+}
+
+/// RAII admission against [`MAX_IN_FLIGHT_FORWARDS`]; releases on drop, so a
+/// client disconnect mid-forward frees the slot.
+struct ForwardGuard<'a> {
+ in_flight: &'a Cell<u32>,
+}
+
+impl<'a> ForwardGuard<'a> {
+ fn admit(in_flight: &'a Cell<u32>) -> Option<Self> {
+ if in_flight.get() >= MAX_IN_FLIGHT_FORWARDS {
+ return None;
+ }
+ in_flight.set(in_flight.get() + 1);
+ Some(Self { in_flight })
+ }
+}
+
+impl Drop for ForwardGuard<'_> {
+ fn drop(&mut self) {
+ self.in_flight.set(self.in_flight.get() - 1);
+ }
+}
+
+/// Exact-DER pin against this node's own end-entity certificate.
Presented-leaf
+/// equality replaces chain building and name checks on purpose (config-listed
+/// IPs rarely appear as SANs in operator certs); handshake signatures are
still
+/// verified with the provider's algorithms, so possession of the pinned
+/// certificate's private key remains required.
+#[derive(Debug)]
+struct PinnedCertVerifier {
+ pinned: CertificateDer<'static>,
+ algorithms: WebPkiSupportedAlgorithms,
+}
+
+impl ServerCertVerifier for PinnedCertVerifier {
+ fn verify_server_cert(
+ &self,
+ end_entity: &CertificateDer<'_>,
+ _intermediates: &[CertificateDer<'_>],
+ _server_name: &ServerName<'_>,
+ _ocsp_response: &[u8],
+ _now: UnixTime,
+ ) -> Result<ServerCertVerified, rustls::Error> {
+ if self.pinned == *end_entity {
+ Ok(ServerCertVerified::assertion())
+ } else {
+ Err(rustls::Error::InvalidCertificate(
+ CertificateError::UnknownIssuer,
+ ))
+ }
+ }
+
+ fn verify_tls12_signature(
+ &self,
+ message: &[u8],
+ cert: &CertificateDer<'_>,
+ dss: &DigitallySignedStruct,
+ ) -> Result<HandshakeSignatureValid, rustls::Error> {
+ rustls::crypto::verify_tls12_signature(message, cert, dss,
&self.algorithms)
+ }
+
+ fn verify_tls13_signature(
+ &self,
+ message: &[u8],
+ cert: &CertificateDer<'_>,
+ dss: &DigitallySignedStruct,
+ ) -> Result<HandshakeSignatureValid, rustls::Error> {
+ rustls::crypto::verify_tls13_signature(message, cert, dss,
&self.algorithms)
+ }
+
+ fn supported_verify_schemes(&self) -> Vec<SignatureScheme> {
+ self.algorithms.supported_schemes()
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn linearizable_query_detected_only_on_exact_pair() {
+ assert!(wants_linearizable(Some("consistency=linearizable")));
+ assert!(wants_linearizable(Some("foo=bar&consistency=linearizable")));
+ assert!(!wants_linearizable(Some("consistency=serializable")));
+ assert!(!wants_linearizable(Some("consistency=LINEARIZABLE")));
+ assert!(!wants_linearizable(None));
+ }
+
+ #[test]
+ fn transient_not_accepted_body_matches_only_its_code() {
+ let accepted = format!(
+
r#"{{"id":{},"code":"transient_not_accepted","reason":"x","field":null}}"#,
+ IggyError::TransientNotAccepted.as_code()
+ );
+ let committed = format!(
+
r#"{{"id":{},"code":"transient_not_committed","reason":"x","field":null}}"#,
+ IggyError::TransientNotCommitted.as_code()
+ );
+ assert!(is_transient_not_accepted_body(accepted.as_bytes()));
+ assert!(!is_transient_not_accepted_body(committed.as_bytes()));
+ assert!(!is_transient_not_accepted_body(b"not json"));
+ assert!(!is_transient_not_accepted_body(b"{}"));
+ }
+
+ #[test]
+ fn forward_guard_caps_and_releases() {
+ let in_flight = Cell::new(0);
+ let guards: Vec<_> = (0..MAX_IN_FLIGHT_FORWARDS)
+ .map(|_| ForwardGuard::admit(&in_flight).expect("under cap"))
+ .collect();
+ assert!(ForwardGuard::admit(&in_flight).is_none());
+ drop(guards);
+ assert_eq!(in_flight.get(), 0);
+ assert!(ForwardGuard::admit(&in_flight).is_some());
+ }
+}
diff --git a/core/server-ng/src/http/handlers.rs
b/core/server-ng/src/http/handlers.rs
index 6ca70039d..be9bf50f8 100644
--- a/core/server-ng/src/http/handlers.rs
+++ b/core/server-ng/src/http/handlers.rs
@@ -151,7 +151,7 @@ const HTTP_READ_CLIENT_ID: u128 = 0;
/// Response header attesting what durability a produce response proves:
/// [`DURABILITY_REPLICATED_MEMORY`] after an awaited quorum commit,
/// [`DURABILITY_NONE`] for a `?ack=none` fire-and-forget.
-const DURABILITY_HEADER: HeaderName =
HeaderName::from_static("x-iggy-durability");
+const DURABILITY_HEADER: HeaderName =
HeaderName::from_static("iggy-durability");
const DURABILITY_REPLICATED_MEMORY: &str = "replicated-memory";
@@ -1116,8 +1116,8 @@ pub(in crate::http) async fn get_consumer_offset(
/// consensus (at-least-once, no dedup, no session gate - concurrent produces
/// on one credential are legal), and the committed reply comes back through
/// the session's in-process reply slot rather than a submit return value.
-/// The default answers 201 + `X-Iggy-Durability: replicated-memory` only
-/// after the quorum commit; `?ack=none` answers 202 + `X-Iggy-Durability:
+/// The default answers 201 + `Iggy-Durability: replicated-memory` only
+/// after the quorum commit; `?ack=none` answers 202 + `Iggy-Durability:
/// none` immediately after dispatch.
pub(in crate::http) async fn send_messages(
State(state): State<HttpState>,
diff --git a/core/server-ng/src/http/jwt.rs b/core/server-ng/src/http/jwt.rs
index 91f1830bc..c5053e147 100644
--- a/core/server-ng/src/http/jwt.rs
+++ b/core/server-ng/src/http/jwt.rs
@@ -38,7 +38,7 @@ use iggy_common::{IggyDuration, IggyError, IggyExpiry,
IggyTimestamp, UserId};
use jsonwebtoken::{Algorithm, DecodingKey, EncodingKey, Header, Validation,
decode, encode};
use serde::{Deserialize, Deserializer, Serialize, Serializer};
use server_common::crypto;
-use tracing::{debug, error, warn};
+use tracing::{debug, error, info, warn};
use uuid::Uuid;
use super::jwks::JwksClient;
@@ -47,6 +47,15 @@ use super::jwks::JwksClient;
/// Matches the legacy server so both behave identically on an empty secret.
const GENERATED_SECRET_LEN: Range<usize> = 32..64;
+/// BLAKE3 `derive_key` context for turning the cluster PSK into the HS256
+/// signing key when no JWT secret is configured in cluster mode. A distinct
+/// context string keeps this key cryptographically independent from the
+/// replica-auth MAC subkey derived from the same PSK. Deliberate coupling:
+/// with this fallback a PSK compromise also yields the token-signing key, and
+/// rotating the PSK invalidates all bearers; operators who want the domains
+/// decoupled configure an explicit `http.jwt` secret, which always wins.
+const JWT_KEY_CONTEXT: &str = "apache-iggy server-ng http-jwt v1
psk->hs256-key";
+
/// Expiry stamp used for a non-expiring token: far enough out to never trip
/// `exp` validation, small enough to fit `u32`. Mirrors the legacy server.
const NEVER_EXPIRE_SECS: u32 = 1_000_000_000;
@@ -74,12 +83,17 @@ impl JwtManager {
/// ephemeral secret, or mirror whichever half is set) and loading any
/// configured trusted issuers (keyed by normalized issuer URL).
///
+ /// `cluster_psk`, when present, replaces the random-mint fallback with a
+ /// key derived from the cluster shared secret, so every node verifies
+ /// every node's bearers - the invariant follower-to-primary forwarding
+ /// depends on. A configured secret always wins over it.
+ ///
/// # Errors
///
/// Returns [`IggyError`] if the configured algorithm is unsupported or a
/// secret cannot be turned into an encoding/decoding key.
- pub fn build(config: &HttpJwtConfig) -> Result<Self, IggyError> {
- let config = normalize_secrets(config.clone());
+ pub fn build(config: &HttpJwtConfig, cluster_psk: Option<&str>) ->
Result<Self, IggyError> {
+ let config = normalize_secrets(config.clone(), cluster_psk);
let algorithm = config.get_algorithm()?;
let encoding_key = config.get_encoding_key()?;
let decoding_key = config.get_decoding_key()?;
@@ -330,12 +344,27 @@ fn normalize_issuer_url(url: &str) -> String {
/// Fill in an unconfigured JWT secret exactly like the legacy HTTP server:
/// both empty -> one random ephemeral secret; one empty -> mirror the other;
/// both set but different under an HMAC algorithm -> warn (they must match).
-fn normalize_secrets(mut config: HttpJwtConfig) -> HttpJwtConfig {
+fn normalize_secrets(mut config: HttpJwtConfig, cluster_psk: Option<&str>) ->
HttpJwtConfig {
match (
config.encoding_secret.is_empty(),
config.decoding_secret.is_empty(),
) {
(true, true) => {
+ if let Some(psk) = cluster_psk {
+ // Cluster-wide deterministic key: hex of the derived 256-bit
+ // subkey, so every node signs and verifies identically and a
+ // follower-forwarded bearer is valid on the primary. Hex (not
+ // raw bytes) because the secret travels the same String path
+ // a configured secret does.
+ let derived = blake3::derive_key(JWT_KEY_CONTEXT,
psk.as_bytes());
+ let secret =
blake3::Hash::from_bytes(derived).to_hex().to_string();
+ info!(
+ "JWT secrets are not configured - derived a cluster-wide
secret from cluster.auth.shared_secret; tokens are valid on every node and
rotate with the PSK"
+ );
+ config.encoding_secret.clone_from(&secret);
+ config.decoding_secret = secret;
+ return config;
+ }
let secret = crypto::generate_secret(GENERATED_SECRET_LEN);
let redacted: String = secret.chars().take(3).collect();
warn!(
@@ -477,7 +506,7 @@ mod tests {
async fn token_presented_before_its_nbf_is_rejected() {
// not_before well past the leeway window, so the freshly issued token
// is not yet valid and decode must reject it.
- let manager = JwtManager::build(&config("3600 s", "5
s")).expect("builds");
+ let manager = JwtManager::build(&config("3600 s", "5 s"),
None).expect("builds");
let token = manager.generate(7).expect("issues");
assert!(
manager.decode(&token.access_token).await.is_err(),
@@ -489,7 +518,7 @@ mod tests {
async fn token_at_its_nbf_is_accepted() {
// Default not_before (0s) => nbf == iat, so the token is valid now and
// enabling nbf validation does not reject a normally issued token.
- let manager = JwtManager::build(&config("0 s", "5
s")).expect("builds");
+ let manager = JwtManager::build(&config("0 s", "5 s"),
None).expect("builds");
let token = manager.generate(7).expect("issues");
let claims = manager
.decode(&token.access_token)
@@ -509,7 +538,7 @@ mod tests {
}]),
..HttpJwtConfig::default()
};
- match JwtManager::build(&jwt) {
+ match JwtManager::build(&jwt, None) {
Err(IggyError::InvalidConfiguration) => {}
Err(other) => panic!("expected InvalidConfiguration, got
{other:?}"),
Ok(_) => panic!("build must reject a trusted issuer mapping to the
root user"),
@@ -527,7 +556,7 @@ mod tests {
}]),
..HttpJwtConfig::default()
};
- match JwtManager::build(&jwt) {
+ match JwtManager::build(&jwt, None) {
Err(IggyError::InvalidConfiguration) => {}
Err(other) => panic!("expected InvalidConfiguration, got
{other:?}"),
Ok(_) => panic!("build must reject a trusted issuer with an empty
issuer"),
@@ -545,7 +574,7 @@ mod tests {
}]),
..HttpJwtConfig::default()
};
- match JwtManager::build(&jwt) {
+ match JwtManager::build(&jwt, None) {
Err(IggyError::InvalidConfiguration) => {}
Err(other) => panic!("expected InvalidConfiguration, got
{other:?}"),
Ok(_) => panic!("build must reject a trusted issuer with an empty
jwks_url"),
diff --git a/core/server-ng/src/http/reply.rs b/core/server-ng/src/http/reply.rs
index 21feb3d81..a9708983a 100644
--- a/core/server-ng/src/http/reply.rs
+++ b/core/server-ng/src/http/reply.rs
@@ -96,18 +96,25 @@ pub(in crate::http) fn committed_payload(
}
}
-/// True when a reply-shaped frame is the primary's pre-consensus
-/// `TransientNotCommitted` rejection (`[count=1][index=0][57]`, see
-/// `build_result_rejection_reply`): the op did not commit, so the write path
-/// must replay the same request id rather than grade it as a committed
-/// result or advance the session gate.
-pub(in crate::http) fn is_transient_not_committed(reply:
&Message<GenericHeader>) -> bool {
- matches!(
- result_code(reply_body(reply)),
- Some(code)
- if code == IggyError::TransientNotCommitted.as_code()
- || code == IggyError::TransientNotAccepted.as_code()
- )
+/// The transient variant of a reply-shaped pre-consensus rejection frame
+/// (`[count=1][index=0][code]`, see `build_result_rejection_reply`), or `None`
+/// for a committed outcome. Either transient means the op did not commit, so
+/// the write path must replay the same request id rather than grade it as a
+/// committed result or advance the session gate. The two codes are kept
+/// distinct because they exhaust differently: `TransientNotAccepted` never
+/// entered the pipeline and is safe to re-issue anywhere, while
+/// `TransientNotCommitted` may still commit and only a same-session same-id
+/// replay is safe.
+pub(in crate::http) fn transient_code(reply: &Message<GenericHeader>) ->
Option<IggyError> {
+ match result_code(reply_body(reply)) {
+ Some(code) if code == IggyError::TransientNotCommitted.as_code() => {
+ Some(IggyError::TransientNotCommitted)
+ }
+ Some(code) if code == IggyError::TransientNotAccepted.as_code() => {
+ Some(IggyError::TransientNotAccepted)
+ }
+ _ => None,
+ }
}
/// The reply body past the generic header, bounded by the header's `size`.
@@ -368,11 +375,25 @@ mod tests {
IggyError::TransientNotCommitted.as_code(),
)
.into_generic();
- assert!(is_transient_not_committed(&reply));
+ assert_eq!(
+ transient_code(&reply),
+ Some(IggyError::TransientNotCommitted)
+ );
assert!(matches!(
committed_payload(&reply),
Err(WriteError::Rejected(IggyError::TransientNotCommitted))
));
+
+ let not_accepted = consensus::build_result_rejection_reply(
+ request.header(),
+ 9,
+ IggyError::TransientNotAccepted.as_code(),
+ )
+ .into_generic();
+ assert_eq!(
+ transient_code(¬_accepted),
+ Some(IggyError::TransientNotAccepted)
+ );
}
/// Genuine committed outcomes must advance the gate, so neither a success
@@ -389,7 +410,7 @@ mod tests {
body.extend_from_slice(b"payload");
let success =
build_reply_from_bytes(request.header(), 42, 7, 9,
&Bytes::from(body)).into_generic();
- assert!(!is_transient_not_committed(&success));
+ assert_eq!(transient_code(&success), None);
let Ok(payload) = committed_payload(&success) else {
panic!("success section must grade ok");
};
@@ -401,7 +422,7 @@ mod tests {
IggyError::UserAlreadyExists.as_code(),
)
.into_generic();
- assert!(!is_transient_not_committed(&rejected));
+ assert_eq!(transient_code(&rejected), None);
assert!(matches!(
committed_payload(&rejected),
Err(WriteError::Rejected(IggyError::UserAlreadyExists))
diff --git a/core/server-ng/src/http/state.rs b/core/server-ng/src/http/state.rs
index 777518946..f596c3ad0 100644
--- a/core/server-ng/src/http/state.rs
+++ b/core/server-ng/src/http/state.rs
@@ -31,6 +31,7 @@ use consensus::{MetadataHandle, VsrConsensus};
use futures::channel::oneshot;
use iggy_common::{ClusterMetadata, IggyTimestamp};
use message_bus::InstanceToken;
+use metadata::MetadataSubmitError;
use send_wrapper::SendWrapper;
use tokio::sync::Mutex;
use tracing::warn;
@@ -39,6 +40,7 @@ use crate::bootstrap::ServerNgShard;
use crate::cluster_meta::ClusterRoster;
use crate::dispatch::submit_register_on_owner;
use crate::http::error::{AuthError, ReadError, primary_redirect_location};
+use crate::http::forward::ForwardState;
use crate::http::jwt::JwtManager;
use crate::http::session::{
BarrierEntry, FIRST_REQUEST_ID, HttpSession, MAX_HTTP_SESSIONS,
RegistrationBarrier,
@@ -48,8 +50,10 @@ use crate::http::session::{
/// Response header carrying the current VSR view number. Stamped by
/// `insert_view_header` on success and redirect responses only (never on
/// errors, and the router suppresses it on `/ping`) while this node has live
-/// consensus.
-const VIEW_HEADER: HeaderName = HeaderName::from_static("x-iggy-view");
+/// consensus. Fill-if-absent: a response relayed from the primary already
+/// carries the SERVING node's view, which must win over the relaying
+/// follower's possibly-stale one.
+pub(in crate::http) const VIEW_HEADER: HeaderName =
HeaderName::from_static("iggy-view");
/// Axum router state: shard-0's [`HttpInner`] behind an `Rc`, `!Send` yet
/// bridged into axum's `Send + Sync` requirement by `SendWrapper`. Sound
@@ -82,6 +86,9 @@ pub(in crate::http) struct HttpInner {
/// by [`MAX_IN_FLIGHT_WRITES_GLOBAL`]. Only [`InFlightWriteGuard`] touches
/// it, so every admission is paired with exactly one release.
pub(in crate::http) in_flight_writes: Cell<u32>,
+ /// Follower-to-primary forwarding context: outbound client, scheme, body
+ /// bound, and its own in-flight budget (see `http::forward`).
+ pub(in crate::http) forward: ForwardState,
}
impl HttpInner {
@@ -114,7 +121,12 @@ impl HttpInner {
.as_ref()
.and_then(|consensus| {
let primary_index = consensus.primary_index(consensus.view());
- primary_redirect_location(&self.roster, primary_index,
path_and_query)
+ primary_redirect_location(
+ &self.roster,
+ primary_index,
+ self.forward.scheme,
+ path_and_query,
+ )
});
location.map_or(ReadError::NotPrimary, ReadError::RedirectToPrimary)
}
@@ -228,7 +240,18 @@ impl HttpInner {
.map_err(|_| AuthError::SessionUnavailable)?
.map_err(|error| {
warn!(?error, "server-ng HTTP: VSR Register submit failed");
- AuthError::SessionUnavailable
+ match error {
+ // The Register never entered the pipeline, so re-issuing
+ // it anywhere is safe; the transient-not-accepted body
+ // tells a forwarding peer to retry against the current
+ // primary. `Canceled` / `InProgress` mean a prepare may
+ // still commit cluster-wide, so they stay the plain
+ // unavailable 503.
+ MetadataSubmitError::NotPrimary
+ | MetadataSubmitError::NotCaughtUp
+ | MetadataSubmitError::PipelineFull =>
AuthError::SessionNotAccepted,
+ _ => AuthError::SessionUnavailable,
+ }
})?;
Ok(Rc::new(HttpSession {
key,
@@ -294,9 +317,12 @@ pub(in crate::http) fn insert_view_header(state:
&HttpInner, mut response: Respo
return response;
}
if let Some(consensus) = state.shard.plane.metadata().consensus.as_ref() {
+ // Fill-if-absent: a relayed response already carries the serving
+ // primary's view, which must not be overwritten with this follower's.
response
.headers_mut()
- .insert(VIEW_HEADER, HeaderValue::from(consensus.view()));
+ .entry(VIEW_HEADER)
+ .or_insert(HeaderValue::from(consensus.view()));
}
response
}
diff --git a/core/server-ng/src/http/submit.rs
b/core/server-ng/src/http/submit.rs
index f343f546f..1f7cc2971 100644
--- a/core/server-ng/src/http/submit.rs
+++ b/core/server-ng/src/http/submit.rs
@@ -38,7 +38,7 @@ use crate::dispatch::{
use crate::http::admission::admit_partition_write;
use crate::http::error::{PartitionWriteError, WriteError};
use crate::http::reply::{
- classify_partition_reply, committed_payload, eviction_error,
is_transient_not_committed,
+ classify_partition_reply, committed_payload, eviction_error,
transient_code,
};
use crate::http::session::HttpSession;
use crate::http::state::HttpInner;
@@ -189,6 +189,7 @@ async fn submit_gated(
let request_header = *message.header();
let deadline = Instant::now() + TRANSIENT_RETRY_DEADLINE;
let mut request = message;
+ let mut saw_not_committed = false;
let reply = loop {
// The submit consumes the request; keep a byte-identical copy for a
// possible replay. Re-running the rewrites instead would mint a fresh
@@ -197,22 +198,36 @@ async fn submit_gated(
let Some(reply) = submit_client_request_on_owner(shard, request).await
else {
return Err(WriteError::Unavailable);
};
- if reply.header().command != Command2::Reply ||
!is_transient_not_committed(&reply) {
+ let transient = (reply.header().command == Command2::Reply)
+ .then(|| transient_code(&reply))
+ .flatten();
+ let Some(transient) = transient else {
break reply;
- }
- // Pre-consensus `TransientNotCommitted`: replay the SAME request id,
- // mirroring the binary SDKs' in-client loop. Safe to replay - the
- // dominant emissions never entered the pipeline, and the view-change
- // cancel is dedup-idempotent (the client table serves the cached
- // reply). The gate stays held across the replay on purpose: the
- // request keeps its serialization turn, and a queued same-session
- // write would only hit the same transient.
+ };
+ saw_not_committed |= matches!(transient,
IggyError::TransientNotCommitted);
+ // Pre-consensus transient frame: replay the SAME request id, mirroring
+ // the binary SDKs' in-client loop. Safe to replay - the dominant
+ // emissions never entered the pipeline, and the view-change cancel is
+ // dedup-idempotent (the client table serves the cached reply). The
+ // gate stays held across the replay on purpose: the request keeps its
+ // serialization turn, and a queued same-session write would only hit
+ // the same transient.
let remaining = deadline.saturating_duration_since(Instant::now());
if remaining.is_zero() {
- // Budget exhausted with the op still not committed: surface the
- // transient code, which renders as a retryable 503, never the
- // catch-all 400.
- return Err(WriteError::Rejected(IggyError::TransientNotCommitted));
+ // Budget exhausted: surface a retryable 503 (never the catch-all
+ // 400), with the code sticky across frames. Once ANY frame was
+ // `TransientNotCommitted` the op may still commit cluster-wide (a
+ // view-change-canceled prepare can reach quorum and be inherited
+ // by the new primary), so a later `TransientNotAccepted` frame -
+ // this node losing the primary role mid-replay - must not
+ // downgrade it: `TransientNotAccepted` licenses a forwarding
+ // follower to re-issue at the new primary under a fresh session,
+ // which would double-apply the possibly-committed op.
+ return Err(WriteError::Rejected(if saw_not_committed {
+ IggyError::TransientNotCommitted
+ } else {
+ transient
+ }));
}
compio::time::sleep(TRANSIENT_RETRY_INTERVAL.min(remaining)).await;
request = retry_request;
diff --git a/core/server-ng/src/server_error.rs
b/core/server-ng/src/server_error.rs
index 041420812..f2987c6e0 100644
--- a/core/server-ng/src/server_error.rs
+++ b/core/server-ng/src/server_error.rs
@@ -177,6 +177,8 @@ pub enum ServerNgError {
#[source]
source: std::io::Error,
},
+ #[error("failed to build the HTTP forward client: {reason}")]
+ HttpForwardClient { reason: String },
#[error("failed to construct IggyShard from bootstrap inputs")]
ShardConstruction(#[source] ShardCtorError),
#[error("{} shard thread(s) failed: {}", failures.len(),
format_shard_failures(failures))]
