v0.32.6_fix1

This commit is contained in:
Mateusz Gruszczyński committed 2026-10-14 08:19:09 +02:00
1 parent a3d183bd4b
commit e3de084149
34 files changed
+559 -85

No files matched your search

+55 -2
View File
@@ -6,7 +6,7 @@ use crate::config::{Backend, UpstreamProtocol};
use anyhow::{Context, Result};
use arc_swap::ArcSwap;
use axum::body::Body;
use http::{Request, Response};
use http::{Request, Response, Version};
use hyper::body::Incoming;
use hyper_rustls::HttpsConnectorBuilder;
use hyper_util::{
@@ -50,6 +50,10 @@ struct RotatingClient {
generation_started_ms: AtomicU64,
rotations: AtomicU64,
rotating: AtomicBool,
observed_version: AtomicU64,
observed_at_ms: AtomicU64,
observed_h1: AtomicU64,
observed_h2: AtomicU64,
}
/// Backend HTTP client with bounded generation lifetime.
@@ -71,6 +75,14 @@ pub struct ClientRotationStats {
pub rotations: u64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct UpstreamProtocolStats {
pub last_version: Option<&'static str>,
pub observed_at_ms: u64,
pub http1_responses: u64,
pub http2_responses: u64,
}
impl std::fmt::Debug for ProxyClient {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
@@ -91,13 +103,25 @@ impl ProxyClient {
generation_started_ms: AtomicU64::new(monotonic_millis()),
rotations: AtomicU64::new(0),
rotating: AtomicBool::new(false),
observed_version: AtomicU64::new(0),
observed_at_ms: AtomicU64::new(0),
observed_h1: AtomicU64::new(0),
observed_h2: AtomicU64::new(0),
}),
}
}
pub async fn request(&self, request: Request<Body>) -> Result<Response<Incoming>, ClientError> {
let client = self.inner.client_for_request();
client.request(request).await
let response = client.request(request).await;
if let Ok(value) = &response {
self.inner.observe_protocol(value.version());
}
response
}
pub fn protocol_stats(&self) -> UpstreamProtocolStats {
self.inner.protocol_stats()
}
pub fn rotation_stats(&self) -> ClientRotationStats {
@@ -119,6 +143,35 @@ impl ProxyClient {
}
impl RotatingClient {
fn observe_protocol(&self, version: Version) {
let code = match version {
Version::HTTP_2 => 2,
Version::HTTP_10 | Version::HTTP_11 => 1,
_ => return,
};
if code == 2 {
self.observed_h2.fetch_add(1, Ordering::Relaxed);
} else {
self.observed_h1.fetch_add(1, Ordering::Relaxed);
}
self.observed_version.store(code, Ordering::Relaxed);
self.observed_at_ms
.store(crate::logging::now_ms(), Ordering::Relaxed);
}
fn protocol_stats(&self) -> UpstreamProtocolStats {
UpstreamProtocolStats {
last_version: match self.observed_version.load(Ordering::Relaxed) {
1 => Some("h1"),
2 => Some("h2"),
_ => None,
},
observed_at_ms: self.observed_at_ms.load(Ordering::Relaxed),
http1_responses: self.observed_h1.load(Ordering::Relaxed),
http2_responses: self.observed_h2.load(Ordering::Relaxed),
}
}
fn stats(&self) -> ClientRotationStats {
let now = monotonic_millis();
ClientRotationStats {
+17
View File
@@ -11,6 +11,23 @@ mod tests {
.unwrap()
}
#[test]
fn protocol_observation_requires_a_real_http_response_version() {
let backend = backend("protocol-observed", 18090);
let client = client(&backend, AddressPolicy::new(&backend).unwrap(), false).unwrap();
assert_eq!(client.protocol_stats().last_version, None);
let shared = client.clone();
// Exercise the response-version recorder independently of the network;
// a live TLS ALPN acceptance test must still run on staging.
shared.inner.observe_protocol(Version::HTTP_11);
shared.inner.observe_protocol(Version::HTTP_2);
let observed = client.protocol_stats();
assert_eq!(observed.last_version, Some("h2"));
assert_eq!(observed.http1_responses, 1);
assert_eq!(observed.http2_responses, 1);
assert!(observed.observed_at_ms > 0);
}
#[test]
fn generation_recycles_after_request_budget() {
let backend = backend("rotation-test", 18081);