From ed2139502f3cb41f2a200042c9efe9dcf303eed2 Mon Sep 17 00:00:00 2001 From: Alexey <247128645+axkurcom@users.noreply.github.com> Date: Sat, 31 Oct 2026 13:22:59 +0300 Subject: [PATCH] X-Telemt-Retry-After-Ms added --- src/api/config_store.rs | 8 ++--- src/config/load.rs | 4 +-- src/logging/file.rs | 6 ++-- src/maestro/listeners/web_overload.rs | 12 ++++++- src/proxy/masking.rs | 2 +- src/proxy/user_admission.rs | 10 ++++-- src/quota_state.rs | 1 - src/transport/middle_proxy/health.rs | 8 ++--- .../middle_proxy/health/floor_plan.rs | 2 -- .../middle_proxy/health/idle_refresh.rs | 1 - src/transport/middle_proxy/pool_refill.rs | 1 - .../middle_proxy/pool_reinit/reconcile.rs | 2 -- src/transport/middle_proxy/pool_writer.rs | 5 ++- .../middle_proxy/pool_writer/runtime.rs | 2 -- src/util/secure_fs/mod.rs | 4 +-- src/web/bridge/behavior_tests.js | 33 +++++++++++++++++++ src/web/bridge/request.js | 2 ++ src/web/http/diagnostic_tests.rs | 4 +++ src/web/http/response.rs | 5 +++ src/web/trace/sanitize.rs | 1 + 20 files changed, 81 insertions(+), 32 deletions(-) diff --git a/src/api/config_store.rs b/src/api/config_store.rs index 85feb2e..372db2e 100644 --- a/src/api/config_store.rs +++ b/src/api/config_store.rs @@ -15,12 +15,12 @@ mod atomic; pub(in crate::api) use atomic::{write_atomic, write_atomic_if_unchanged}; #[cfg(test)] -use persistence::{find_toml_table_bounds, render_access_section, save_sections_to_disk}; -#[cfg(test)] pub(in crate::api) use persistence::save_access_sections_to_disk; +#[cfg(test)] +use persistence::{find_toml_table_bounds, render_access_section, save_sections_to_disk}; pub(in crate::api) use persistence::{ - render_server_listeners, render_top_level_section, - save_access_sections_to_disk_if_revision, upsert_toml_table, + render_server_listeners, render_top_level_section, save_access_sections_to_disk_if_revision, + upsert_toml_table, }; #[derive(Clone, Copy, Debug, PartialEq, Eq)] diff --git a/src/config/load.rs b/src/config/load.rs index 6eb420a..6b28fbc 100644 --- a/src/config/load.rs +++ b/src/config/load.rs @@ -40,9 +40,7 @@ mod validation; #[cfg(test)] use self::includes::normalize_config_path; -use self::includes::{ - hash_rendered_snapshot, preprocess_includes, read_config_source, -}; +use self::includes::{hash_rendered_snapshot, preprocess_includes, read_config_source}; use self::normalize::{ is_valid_ad_tag, is_valid_tls_domain_name, normalize_domain_to_ascii, normalize_exclusive_mask_target, normalize_mask_host_to_ascii, parse_exclusive_mask_target, diff --git a/src/logging/file.rs b/src/logging/file.rs index 13f0fa7..211a651 100644 --- a/src/logging/file.rs +++ b/src/logging/file.rs @@ -1,8 +1,8 @@ -#[cfg(not(unix))] -use std::fs::OpenOptions; -use std::fs::File; #[cfg(any(test, not(unix)))] use std::fs; +use std::fs::File; +#[cfg(not(unix))] +use std::fs::OpenOptions; use std::io::{self, Write}; use std::path::{Path, PathBuf}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; diff --git a/src/maestro/listeners/web_overload.rs b/src/maestro/listeners/web_overload.rs index 22a7b58..1d225b2 100644 --- a/src/maestro/listeners/web_overload.rs +++ b/src/maestro/listeners/web_overload.rs @@ -13,7 +13,7 @@ use crate::web::manager::{HttpConnectionAdmissionError, WebProcessRuntime}; use crate::web::telemetry::{WebHttpConnectionOverloadOutcome, WebRejectionReason}; /// Exact bounded retryable response emitted before HTTP request parsing. -pub(super) const SERVICE_UNAVAILABLE_RESPONSE: &[u8] = b"HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nCache-Control: no-store\r\nRetry-After: 1\r\nConnection: close\r\n\r\n"; +pub(super) const SERVICE_UNAVAILABLE_RESPONSE: &[u8] = b"HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nCache-Control: no-store\r\nRetry-After: 1\r\nX-Telemt-Retry-After-Ms: 250\r\nConnection: close\r\n\r\n"; /// Handles one accepted WEB socket outside ordinary connection capacity. #[allow(clippy::too_many_arguments)] @@ -164,6 +164,11 @@ mod tests { let mut bytes = Vec::new(); client.read_to_end(&mut bytes).await.unwrap(); assert_eq!(bytes, super::SERVICE_UNAVAILABLE_RESPONSE); + assert!( + bytes + .windows(b"X-Telemt-Retry-After-Ms: 250\r\n".len()) + .any(|window| window == b"X-Telemt-Retry-After-Ms: 250\r\n") + ); } #[tokio::test] @@ -191,6 +196,11 @@ mod tests { let mut bytes = Vec::new(); client.read_to_end(&mut bytes).await.unwrap(); assert_eq!(bytes, super::SERVICE_UNAVAILABLE_RESPONSE); + assert!( + bytes + .windows(b"X-Telemt-Retry-After-Ms: 250\r\n".len()) + .any(|window| window == b"X-Telemt-Retry-After-Ms: 250\r\n") + ); assert_eq!( runtime .telemetry() diff --git a/src/proxy/masking.rs b/src/proxy/masking.rs index aa11fb7..4f5bdc8 100644 --- a/src/proxy/masking.rs +++ b/src/proxy/masking.rs @@ -83,9 +83,9 @@ pub(in crate::proxy) use handler::handle_bad_client_with_shared_resolver; use interfaces::*; use relay::*; use target::*; -pub(crate) use timing::sample_lognormal_percentile_bounded; #[cfg(test)] use timing::mask_outcome_target_budget; +pub(crate) use timing::sample_lognormal_percentile_bounded; use timing::{wait_mask_connect_budget_if_needed, wait_mask_outcome_budget}; #[cfg(test)] diff --git a/src/proxy/user_admission.rs b/src/proxy/user_admission.rs index 773b5e4..9629fe7 100644 --- a/src/proxy/user_admission.rs +++ b/src/proxy/user_admission.rs @@ -98,13 +98,19 @@ pub(crate) struct UserMutationResult { /// Number of live owners cancelled by the mutation. #[cfg_attr( not(test), - expect(dead_code, reason = "Mutation results are asserted by admission regression tests") + expect( + dead_code, + reason = "Mutation results are asserted by admission regression tests" + ) )] pub(crate) cancelled: usize, /// Whether the effective enabled state changed from enabled to disabled. #[cfg_attr( not(test), - expect(dead_code, reason = "Mutation results are asserted by admission regression tests") + expect( + dead_code, + reason = "Mutation results are asserted by admission regression tests" + ) )] pub(crate) newly_disabled: bool, } diff --git a/src/quota_state.rs b/src/quota_state.rs index 1122847..07c3fe0 100644 --- a/src/quota_state.rs +++ b/src/quota_state.rs @@ -180,7 +180,6 @@ impl QuotaStateOwner { wait_for_blocking_io(task).await } - /// Persists deletion while retaining the already-owned lifecycle gate. pub(crate) async fn remove_user_locked( &self, diff --git a/src/transport/middle_proxy/health.rs b/src/transport/middle_proxy/health.rs index e5d9898..a020711 100644 --- a/src/transport/middle_proxy/health.rs +++ b/src/transport/middle_proxy/health.rs @@ -68,7 +68,6 @@ struct FamilyFloorPlan { warm_writers_current: usize, } - #[cfg(test)] struct ScheduledReconnects<'a> { inflight: &'a mut HashMap<(i32, IpFamily), usize>, @@ -77,7 +76,6 @@ struct ScheduledReconnects<'a> { #[cfg(test)] impl ScheduledReconnects<'_> { - fn reserve(&mut self, key: (i32, IpFamily)) { self.keys.push(key); *self.inflight.entry(key).or_insert(0) += 1; @@ -119,17 +117,15 @@ mod recovery; // Independent stale draining-writer watchdog. mod zombie_watchdog; -pub use drain::me_drain_timeout_enforcer; #[cfg(test)] pub(in crate::transport::middle_proxy) use drain::health_drain_close_budget; +pub use drain::me_drain_timeout_enforcer; pub(in crate::transport::middle_proxy) use drain::reap_draining_writers; use family::{check_family, update_family_runtime_state}; use floor_plan::build_family_floor_plan; use idle_refresh::maybe_refresh_idle_writer_for_dc; pub use monitor::me_health_monitor; -use recovery::{ - has_bound_clients_on_endpoint, maybe_rotate_single_endpoint_shadow, -}; +use recovery::{has_bound_clients_on_endpoint, maybe_rotate_single_endpoint_shadow}; pub use zombie_watchdog::me_zombie_writer_watchdog; #[cfg(test)] diff --git a/src/transport/middle_proxy/health/floor_plan.rs b/src/transport/middle_proxy/health/floor_plan.rs index 6599654..4c81289 100644 --- a/src/transport/middle_proxy/health/floor_plan.rs +++ b/src/transport/middle_proxy/health/floor_plan.rs @@ -1,7 +1,5 @@ use super::*; - - pub(super) fn adaptive_floor_class_min( pool: &Arc, endpoint_count: usize, diff --git a/src/transport/middle_proxy/health/idle_refresh.rs b/src/transport/middle_proxy/health/idle_refresh.rs index 78fddc4..5188231 100644 --- a/src/transport/middle_proxy/health/idle_refresh.rs +++ b/src/transport/middle_proxy/health/idle_refresh.rs @@ -1,6 +1,5 @@ use super::*; - pub(super) async fn maybe_refresh_idle_writer_for_dc( pool: &Arc, rng: &Arc, diff --git a/src/transport/middle_proxy/pool_refill.rs b/src/transport/middle_proxy/pool_refill.rs index ad78bfd..d2bbcc5 100644 --- a/src/transport/middle_proxy/pool_refill.rs +++ b/src/transport/middle_proxy/pool_refill.rs @@ -193,7 +193,6 @@ impl MePool { self.refill_states.lock().contains_key(&key) } - pub(super) async fn connect_endpoints_round_robin_with_generation_contour( self: &Arc, dc: i32, diff --git a/src/transport/middle_proxy/pool_reinit/reconcile.rs b/src/transport/middle_proxy/pool_reinit/reconcile.rs index e6efa13..7cab82a 100644 --- a/src/transport/middle_proxy/pool_reinit/reconcile.rs +++ b/src/transport/middle_proxy/pool_reinit/reconcile.rs @@ -169,7 +169,6 @@ impl MePool { } } - /// Reconciles exactly the revision owned by the scheduler's retry objective. pub(in crate::transport::middle_proxy) async fn reinit_endpoint_snapshot( self: &Arc, @@ -408,5 +407,4 @@ impl MePool { } true } - } diff --git a/src/transport/middle_proxy/pool_writer.rs b/src/transport/middle_proxy/pool_writer.rs index f208804..6abc36b 100644 --- a/src/transport/middle_proxy/pool_writer.rs +++ b/src/transport/middle_proxy/pool_writer.rs @@ -57,7 +57,10 @@ pub(super) enum WriterReplacementPurpose { /// Moves one idle slot from a donor above floor to a receiver below floor. #[cfg_attr( not(test), - expect(dead_code, reason = "Legacy purpose is exercised by commit-fence regression tests") + expect( + dead_code, + reason = "Legacy purpose is exercised by commit-fence regression tests" + ) )] FloorRebalance { /// Minimum writer count that must remain in the donor group. diff --git a/src/transport/middle_proxy/pool_writer/runtime.rs b/src/transport/middle_proxy/pool_writer/runtime.rs index b31fbb8..c09240f 100644 --- a/src/transport/middle_proxy/pool_writer/runtime.rs +++ b/src/transport/middle_proxy/pool_writer/runtime.rs @@ -407,7 +407,6 @@ impl MePool { removed } - /// Publishes drain metadata before the release-store that makes draining observable. pub(in crate::transport::middle_proxy) fn apply_writer_draining_state( &self, @@ -441,7 +440,6 @@ impl MePool { !already_draining } - pub(in crate::transport::middle_proxy) fn writer_accepts_new_binding( &self, writer: &MeWriter, diff --git a/src/util/secure_fs/mod.rs b/src/util/secure_fs/mod.rs index 56ce4d1..e54cbf4 100644 --- a/src/util/secure_fs/mod.rs +++ b/src/util/secure_fs/mod.rs @@ -14,8 +14,8 @@ pub(crate) use path::{ #[cfg(test)] pub(crate) use write::open_append_regular; pub(crate) use write::{ - atomic_replace, atomic_replace_async, open_append_regular_at, - read_regular_limited, read_regular_limited_async, + atomic_replace, atomic_replace_async, open_append_regular_at, read_regular_limited, + read_regular_limited_async, }; #[cfg(test)] diff --git a/src/web/bridge/behavior_tests.js b/src/web/bridge/behavior_tests.js index 467d55c..4feeb08 100644 --- a/src/web/bridge/behavior_tests.js +++ b/src/web/bridge/behavior_tests.js @@ -54,6 +54,39 @@ function streaming(env,request,next='1',length=null){ let controller;const body=new ReadableStream({start(value){controller=value}}),headers={'X-Down-Cursor':next};if(length!==null)headers['Content-Length']=String(length); request.answered=true;request.resolve({status:200,headers:new Headers(headers),body});return controller; } +test('Telemt overload hint retries at 250 ms with legacy and malformed fallback',async page=>{ + for(const android of [false,true])for(const carrier of ['https','https-lanes'])for(const [status,hint,retry,delay] of [ + [503,'250','1',250],[503,null,'1',1000],[503,'0','1',1000], + [503,'0.25','1',1000],[503,'250, 250','1',1000],[503,'250','2',2000],[429,'250','1',1000] + ]){ + const env=await session(page,carrier,4,android);env.send(frame(1,1));await flush(); + const original=env.pending('/api/v1/up')[0],headers={'Retry-After':retry}; + if(hint!==null)headers['X-Telemt-Retry-After-Ms']=hint; + env.answer(original,status,null,headers);await flush(); + await env.tick(delay-1);assert.equal(env.pending('/api/v1/up').length,0,'retry must not run early'); + await env.tick(1);const replay=env.pending('/api/v1/up')[0];assert.ok(replay,'retry at selected deadline'); + assert.equal(replay.options.headers['X-Up-Seq'],original.options.headers['X-Up-Seq']); + assert.deepEqual(new Uint8Array(replay.options.body),new Uint8Array(original.options.body)); + ack(env,replay);await flush();env.close();await flush(); + } +}); +test('repeated 250 ms overload hints retain the nine-attempt limit',async page=>{ + const env=await session(page);env.send(frame(1,1));await flush(); + for(let attempt=1;attempt<=9;attempt++){ + const request=env.pending('/api/v1/up')[0];assert.ok(request,'bounded retry attempt '+attempt); + env.answer(request,503,null,{'Retry-After':'1','X-Telemt-Retry-After-Ms':'250'});await flush(); + if(attempt<9)await env.tick(250); + } + await env.tick(250); + assert.equal(env.requests.filter(r=>r.url.endsWith('/api/v1/up')).length,9); + env.close();await flush(); +}); +test('closing the bridge cancels the 250 ms overload retry',async page=>{ + const env=await session(page);env.send(frame(1,1));await flush(); + env.answer(env.pending('/api/v1/up')[0],503,null,{'Retry-After':'1','X-Telemt-Retry-After-Ms':'250'});await flush(); + await env.tick(249);env.close();await flush();await env.tick(1000); + assert.equal(env.requests.filter(r=>r.url.endsWith('/api/v1/up')).length,1); +}); test('already queued OPEN and DATA share the zero-wait startup request',async page=>{ const env=await session(page);const body=join(frame(1,1),frame(2,1,[7,8]));env.send(body);await flush(); assert.equal(env.pending('/api/v1/up')[0].options.body.byteLength,body.byteLength);env.close();await flush(); diff --git a/src/web/bridge/request.js b/src/web/bridge/request.js index 01ddf80..edc9422 100644 --- a/src/web/bridge/request.js +++ b/src/web/bridge/request.js @@ -12,6 +12,8 @@ function create(settings){ }); function retryAfterMs(response){ const header=response.headers.get('Retry-After'); + // Only the exact Telemt overload hint overrides its legacy whole-second fallback. + if(response.status===503&&header==='1'&&response.headers.get('X-Telemt-Retry-After-Ms')==='250')return 250; if(!header)return 0; const seconds=Number(header); if(Number.isFinite(seconds)&&seconds>=0)return Math.min(seconds*1000,30000); diff --git a/src/web/http/diagnostic_tests.rs b/src/web/http/diagnostic_tests.rs index e1d680b..6084163 100644 --- a/src/web/http/diagnostic_tests.rs +++ b/src/web/http/diagnostic_tests.rs @@ -409,6 +409,10 @@ async fn diagnostic_body_pressure_does_not_claim_the_event() { response_header(split_response(&saturated).0, "retry-after"), "1" ); + assert_eq!( + response_header(split_response(&saturated).0, "x-telemt-retry-after-ms"), + "250" + ); drop(held); let accepted = request(&listener, &runtime, retry).await; diff --git a/src/web/http/response.rs b/src/web/http/response.rs index 8db4163..68c3d0b 100644 --- a/src/web/http/response.rs +++ b/src/web/http/response.rs @@ -46,6 +46,11 @@ pub(super) fn service_unavailable() -> HttpResponse { response .headers_mut() .insert(header::RETRY_AFTER, HeaderValue::from_static("1")); + // Keep whole-second HTTP guidance for clients without the Telemt extension. + response.headers_mut().insert( + HeaderName::from_static("x-telemt-retry-after-ms"), + HeaderValue::from_static("250"), + ); response } diff --git a/src/web/trace/sanitize.rs b/src/web/trace/sanitize.rs index 9c7443a..a4a5575 100644 --- a/src/web/trace/sanitize.rs +++ b/src/web/trace/sanitize.rs @@ -216,6 +216,7 @@ fn header_value_allowed(name: &hyper::header::HeaderName) -> bool { | "x-carrier-failure" | "x-carrier-mode" | "retry-after" + | "x-telemt-retry-after-ms" | "cache-control" | "etag" | "if-none-match"