From 3df59cfa28f7d12186548126c2b6e417831eb218 Mon Sep 17 00:00:00 2001 From: Alexey <247128645+axkurcom@users.noreply.github.com> Date: Sat, 31 Oct 2026 14:23:01 +0300 Subject: [PATCH] Retry Jitter --- src/api/config_edit/host_list_tests.rs | 47 +++++++++ src/config/load/strict_keys.rs | 3 + src/config/load/strict_keys/check.rs | 8 ++ src/config/load/validate_web.rs | 5 + src/config/types.rs | 3 +- src/config/types/web.rs | 7 ++ src/config/types/web/retry.rs | 128 +++++++++++++++++++++++++ src/web/bridge.rs | 2 + src/web/bridge/behavior_tests.js | 11 ++- src/web/bridge/behavior_tests.rs | 1 + src/web/bridge/render_reference.rs | 2 + src/web/bridge/render_tests.rs | 26 +++++ src/web/bridge/request.js | 24 ++++- src/web/bridge/retry_jitter_tests.js | 72 ++++++++++++++ src/web/bridge/runtime.js | 1 + src/web/bridge/tests.rs | 6 ++ src/web/http.rs | 1 + 17 files changed, 342 insertions(+), 5 deletions(-) create mode 100644 src/config/types/web/retry.rs create mode 100644 src/web/bridge/retry_jitter_tests.js diff --git a/src/api/config_edit/host_list_tests.rs b/src/api/config_edit/host_list_tests.rs index 1ac2d08..f0f9450 100644 --- a/src/api/config_edit/host_list_tests.rs +++ b/src/api/config_edit/host_list_tests.rs @@ -141,3 +141,50 @@ async fn grouped_hosts_cannot_bypass_frozen_process_limits() { // Effective process policy must reject patterns until the explicit restart has happened. #[path = "wildcard_tests.rs"] mod wildcard_tests; + +#[tokio::test] +async fn retry_jitter_patch_validates_and_reloads_without_restart() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("config.toml"); + std::fs::write(&path, source()).unwrap(); + let initial = ProxyConfig::load(&path).unwrap(); + let (_, revision) = read_managed_config(&path).await.unwrap(); + let patch = serde_json::json!({"web":{"retry_jitter":{ + "enabled":true,"min_ms":10,"max_ms":300,"distribution":"triangular","peak_ms":100 + }}}); + let mut response = apply_patch_to_path(&path, &patch, Some(revision)) + .await + .unwrap(); + let desired = ProxyConfig::load(&path).unwrap(); + let resolved = reconcile_runtime_effect(&mut response, &initial, &desired).unwrap(); + assert!(resolved.runtime_changed); + assert!(!response.process_restart_required); + assert_eq!( + resolved.effective.web.retry_jitter, + desired.web.retry_jitter + ); + assert!(!initial.web.retry_jitter.enabled); + let (managed, revision) = read_managed_config(&path).await.unwrap(); + assert_eq!( + managed["web"]["retry_jitter"]["peak_ms"].as_integer(), + Some(100) + ); + let before = std::fs::read(&path).unwrap(); + for invalid in [ + serde_json::json!({"max_ms":30001}), + serde_json::json!({"peak_ms":301}), + serde_json::json!({"distribution":"uniform"}), + serde_json::json!({"typo":1}), + ] { + assert!( + apply_patch_to_path( + &path, + &serde_json::json!({"web":{"retry_jitter":invalid}}), + Some(revision.clone()) + ) + .await + .is_err() + ); + assert_eq!(std::fs::read(&path).unwrap(), before); + } +} diff --git a/src/config/load/strict_keys.rs b/src/config/load/strict_keys.rs index eda1a8a..8199365 100644 --- a/src/config/load/strict_keys.rs +++ b/src/config/load/strict_keys.rs @@ -260,7 +260,10 @@ const LISTENER_CONFIG_KEYS: &[&str] = &[ "web_trusted_proxy_cidrs", ]; +const WEB_RETRY_JITTER_KEYS: &[&str] = &["enabled", "min_ms", "max_ms", "distribution", "peak_ms"]; + const WEB_CONFIG_KEYS: &[&str] = &[ + "retry_jitter", "allow_wildcard_hosts", "enabled", "carrier", diff --git a/src/config/load/strict_keys/check.rs b/src/config/load/strict_keys/check.rs index 1fc6357..71b2ac5 100644 --- a/src/config/load/strict_keys/check.rs +++ b/src/config/load/strict_keys/check.rs @@ -36,6 +36,7 @@ fn known_config_keys_for_suggestion() -> Vec<&'static str> { CONNTRACK_CONTROL_CONFIG_KEYS, LISTENER_CONFIG_KEYS, WEB_CONFIG_KEYS, + WEB_RETRY_JITTER_KEYS, WEB_LIMITS_CONFIG_KEYS, WEB_WILDCARD_LIMITS_CONFIG_KEYS, WEB_DEBUG_CONFIG_KEYS, @@ -236,6 +237,13 @@ pub(super) fn collect_unknown_config_keys(parsed_toml: &toml::Value) -> Vec Result<()> { )); } + config + .web + .retry_jitter + .validate() + .map_err(|message| ProxyError::Config(message.into()))?; validate_limits(&config.web.limits, config.web.allow_wildcard_hosts)?; debug::validate(&config.web.debug, &config.web.limits)?; let carriers = negotiation::validate(&config.web)?; diff --git a/src/config/types.rs b/src/config/types.rs index 51d8efd..185d845 100644 --- a/src/config/types.rs +++ b/src/config/types.rs @@ -58,7 +58,8 @@ pub use server::{ pub use web::{ WebCarrierNegotiationAggressiveness, WebConfig, WebDecoyConfig, WebDecoyFastTrackMode, WebDecoyResolve, WebHttpConnectionCapacityAction, WebLimitsConfig, WebProfileConfig, - WebSecretMode, WebTimeoutsConfig, WebVhostConfig, WebWildcardLimitsConfig, + WebRetryJitterConfig, WebRetryJitterDistribution, WebSecretMode, WebTimeoutsConfig, + WebVhostConfig, WebWildcardLimitsConfig, }; pub(crate) use web::{ WebDecoyDnsSnapshot, WebRuntimeConfig, WebRuntimeDecoy, WebRuntimeProfile, diff --git a/src/config/types/web.rs b/src/config/types/web.rs index cbc34c9..8c86353 100644 --- a/src/config/types/web.rs +++ b/src/config/types/web.rs @@ -14,6 +14,9 @@ mod wildcard; pub use wildcard::WebWildcardLimitsConfig; // Serialized WEB defaults remain separate from the runtime data model. +mod retry; +pub use retry::{WebRetryJitterConfig, WebRetryJitterDistribution}; + mod defaults; use defaults::*; // Decoy fast-track policy remains isolated from the bulky WEB data model. @@ -469,6 +472,9 @@ pub struct WebConfig { /// WEB lifecycle deadlines. #[serde(default)] pub timeouts: WebTimeoutsConfig, + /// Additive HTTP retry jitter frozen into each generated bridge page. + #[serde(default)] + pub retry_jitter: WebRetryJitterConfig, /// Exact hostname and opted-in wildcard groups served by WEB listeners. #[serde(default)] pub vhosts: Vec, @@ -517,6 +523,7 @@ impl Default for WebConfig { limits: WebLimitsConfig::default(), debug: WebDebugConfig::default(), timeouts: WebTimeoutsConfig::default(), + retry_jitter: WebRetryJitterConfig::default(), vhosts: Vec::new(), runtime: None, decoy_dns: Arc::default(), diff --git a/src/config/types/web/retry.rs b/src/config/types/web/retry.rs new file mode 100644 index 0000000..b92cc41 --- /dev/null +++ b/src/config/types/web/retry.rs @@ -0,0 +1,128 @@ +use serde::{Deserialize, Serialize}; + +/// Shape of the optional additive HTTP retry delay. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum WebRetryJitterDistribution { + /// Equal probability for each integer delay in the configured interval. + #[default] + Uniform, + /// Continuous triangular density rounded to milliseconds. + Triangular, +} + +/// Page-owned retry policy; configuration reload affects newly rendered bridges. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(default, deny_unknown_fields)] +pub struct WebRetryJitterConfig { + /// Replaces the legacy jitter while retaining the underlying retry delay. + pub enabled: bool, + /// Inclusive minimum additive delay in milliseconds. + pub min_ms: u32, + /// Inclusive maximum additive delay, bounded to 30000 milliseconds. + pub max_ms: u32, + /// Probability distribution used independently for each retry. + pub distribution: WebRetryJitterDistribution, + /// Triangular mode; omission selects the midpoint of the interval. + #[serde(skip_serializing_if = "Option::is_none")] + pub peak_ms: Option, +} + +impl Default for WebRetryJitterConfig { + fn default() -> Self { + Self { + enabled: false, + min_ms: 0, + max_ms: 250, + distribution: WebRetryJitterDistribution::Uniform, + peak_ms: None, + } + } +} + +impl WebRetryJitterConfig { + /// Validates inactive policies too, so later activation cannot expose invalid values. + pub(crate) fn validate(&self) -> Result<(), &'static str> { + if self.min_ms > self.max_ms || self.max_ms > 30000 { + return Err("web.retry_jitter requires 0 <= min_ms <= max_ms <= 30000"); + } + if let Some(peak) = self.peak_ms { + if self.distribution != WebRetryJitterDistribution::Triangular + || peak < self.min_ms + || peak > self.max_ms + { + return Err( + "web.retry_jitter.peak_ms requires triangular distribution and an in-range peak", + ); + } + } + Ok(()) + } + + /// Emits only typed numeric and boolean literals, without user-controlled script text. + pub(crate) fn bridge_policy(&self) -> String { + let peak = self + .peak_ms + .map(f64::from) + .unwrap_or((f64::from(self.min_ms) + f64::from(self.max_ms)) / 2.0); + format!( + "{{enabled:{},min:{},max:{},peak:{},triangular:{}}}", + self.enabled, + self.min_ms, + self.max_ms, + peak, + self.distribution == WebRetryJitterDistribution::Triangular, + ) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn retry_jitter_validation_and_roundtrip() { + let default = WebRetryJitterConfig::default(); + assert!(!default.enabled); + assert_eq!(default.validate(), Ok(())); + for (min, max, peak, triangular, valid) in [ + (0, 250, None, false, true), + (250, 250, Some(250), true, true), + (0, 30000, Some(0), true, true), + (0, 250, Some(250), true, true), + (251, 250, None, false, false), + (0, 30001, None, false, false), + (0, 250, Some(125), false, false), + (100, 250, Some(99), true, false), + (0, 250, Some(251), true, false), + ] { + for enabled in [false, true] { + let policy = WebRetryJitterConfig { + enabled, + min_ms: min, + max_ms: max, + peak_ms: peak, + distribution: if triangular { + WebRetryJitterDistribution::Triangular + } else { + WebRetryJitterDistribution::Uniform + }, + }; + assert_eq!(policy.validate().is_ok(), valid); + let json = serde_json::to_string(&policy).unwrap(); + assert_eq!( + serde_json::from_str::(&json).unwrap(), + policy + ); + } + } + for json in [ + r#"{"min_ms":-1}"#, + r#"{"max_ms":1.5}"#, + r#"{"distribution":"normal"}"#, + r#"{"unknown":1}"#, + ] { + assert!(serde_json::from_str::(json).is_err()); + } + } +} diff --git a/src/web/bridge.rs b/src/web/bridge.rs index b057072..efe65f2 100644 --- a/src/web/bridge.rs +++ b/src/web/bridge.rs @@ -38,6 +38,7 @@ pub(crate) fn render( carrier_probe_coalesce_ms: u64, bridge_diagnostics_enabled: bool, carrier_method: WebCarrierMethod, + retry_jitter: &crate::config::WebRetryJitterConfig, rng: &SecureRandom, ) -> BridgePage { let mut nonce = [0u8; 18]; @@ -47,6 +48,7 @@ pub(crate) fn render( let body = template(bridge_diagnostics_enabled) .replace("__NONCE__", &nonce) .replace("__HOST__", host) + .replace("__RETRY_JITTER__", &retry_jitter.bridge_policy()) .replace("__BASE_PREFIX__", base_prefix) .replace("__BOOTSTRAP__", bootstrap) .replace("__CARRIER_METHOD__", carrier_method.as_str()) diff --git a/src/web/bridge/behavior_tests.js b/src/web/bridge/behavior_tests.js index 4feeb08..4744793 100644 --- a/src/web/bridge/behavior_tests.js +++ b/src/web/bridge/behavior_tests.js @@ -11,7 +11,7 @@ function renderedPage(){ page=page.replace('__DIAGNOSTIC_RUNTIME__',''); for(const [key,file] of Object.entries(modules))if(page.includes('__'+key+'_RUNTIME__'))page=page.replace('__'+key+'_RUNTIME__',()=>fs.readFileSync(path.join(__dirname,file+'.js'),'utf8')); page=page.replace('__RUNTIME__',()=>fs.readFileSync(path.join(__dirname,'runtime.js'),'utf8')); - const values={BOOTSTRAP:'A'.repeat(43),HOST:'proxy.example.com',BASE_PREFIX:'',CARRIER_METHOD:'POST',NEGOTIATION_ENABLED:'true',CANDIDATE_COUNT:4,CARRIER_DEADLINES:'3,5,8,12',LONG_POLL_SECS:25,BRIDGE_REQUEST_SECS:10,BRIDGE_RETRY_SECS:90,BRIDGE_RECOVERY_SECS:15,WEBSOCKET_OPEN_SECS:15,RECONNECT_GRACE_SECS:120,CARRIER_PROBE_COALESCE_MS:0,BATCH_LIMIT:2097152,QUEUE_LIMIT:33554432,QUEUE_ITEMS:16384,MAX_STREAMS:1024,STATUS_FUNCTION:"state=>{if(port&&!closed)port.postMessage({t:'status',state})}",HELLO_TIMEOUT_CALLBACK:"()=>fail('timeout')",PAGEHIDE_CALLBACK:'()=>close(true)'}; + const values={RETRY_JITTER:'{enabled:false,min:0,max:250,peak:125,triangular:false}',BOOTSTRAP:'A'.repeat(43),HOST:'proxy.example.com',BASE_PREFIX:'',CARRIER_METHOD:'POST',NEGOTIATION_ENABLED:'true',CANDIDATE_COUNT:4,CARRIER_DEADLINES:'3,5,8,12',LONG_POLL_SECS:25,BRIDGE_REQUEST_SECS:10,BRIDGE_RETRY_SECS:90,BRIDGE_RECOVERY_SECS:15,WEBSOCKET_OPEN_SECS:15,RECONNECT_GRACE_SECS:120,CARRIER_PROBE_COALESCE_MS:0,BATCH_LIMIT:2097152,QUEUE_LIMIT:33554432,QUEUE_ITEMS:16384,MAX_STREAMS:1024,STATUS_FUNCTION:"state=>{if(port&&!closed)port.postMessage({t:'status',state})}",HELLO_TIMEOUT_CALLBACK:"()=>fail('timeout')",PAGEHIDE_CALLBACK:'()=>close(true)'}; return page.replace(/__([A-Z_]+)__/g,(all,key)=>key.startsWith('DIAGNOSTIC_')?'':String(values[key]??'')); } function frame(type,id=0,payload=[]){ @@ -87,6 +87,15 @@ test('closing the bridge cancels the 250 ms overload retry',async page=>{ 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('retry jitter distributions, fallback, deadlines and page ownership',require('./retry_jitter_tests')); +test('configured jitter is added once to the Telemt delay',async page=>{ + page=page.replace('{enabled:false,min:0,max:250,peak:125,triangular:false}','{enabled:true,min:100,max:100,peak:100,triangular:false}'); + 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(349);assert.equal(env.pending('/api/v1/up').length,0); + await env.tick(1);assert.equal(env.pending('/api/v1/up').length,1); + env.close();await flush(); +}); 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/behavior_tests.rs b/src/web/bridge/behavior_tests.rs index 093d791..e797841 100644 --- a/src/web/bridge/behavior_tests.rs +++ b/src/web/bridge/behavior_tests.rs @@ -4,6 +4,7 @@ use std::process::{Command, Stdio}; /// Exercises the actual generated page with deterministic network and native-boundary doubles. pub(super) fn run(page: &super::BridgePage) { let mut child = Command::new("node") + .current_dir(concat!(env!("CARGO_MANIFEST_DIR"), "/src/web/bridge")) .args(["-e", include_str!("behavior_tests.js"), "--", "--stdin"]) .stdin(Stdio::piped()) .stdout(Stdio::piped()) diff --git a/src/web/bridge/render_reference.rs b/src/web/bridge/render_reference.rs index 2629100..3f5ee0c 100644 --- a/src/web/bridge/render_reference.rs +++ b/src/web/bridge/render_reference.rs @@ -23,6 +23,7 @@ pub(super) fn render( carrier_probe_coalesce_ms: u64, bridge_diagnostics_enabled: bool, carrier_method: WebCarrierMethod, + retry_jitter: &crate::config::WebRetryJitterConfig, nonce: &str, ) -> BridgePage { let diagnostic_script = if bridge_diagnostics_enabled { @@ -110,6 +111,7 @@ pub(super) fn render( ) .replace("__NONCE__", &nonce) .replace("__HOST__", host) + .replace("__RETRY_JITTER__", &retry_jitter.bridge_policy()) .replace("__BASE_PREFIX__", base_prefix) .replace("__BOOTSTRAP__", bootstrap) .replace("__CARRIER_METHOD__", carrier_method.as_str()) diff --git a/src/web/bridge/render_tests.rs b/src/web/bridge/render_tests.rs index f04d976..b7d6fc8 100644 --- a/src/web/bridge/render_tests.rs +++ b/src/web/bridge/render_tests.rs @@ -55,11 +55,13 @@ struct Case { negotiation: bool, candidates: usize, value: usize, + jitter: crate::config::WebRetryJitterConfig, } impl Case { fn new(index: usize) -> Self { Self { + jitter: crate::config::WebRetryJitterConfig::default(), host: format!("proxy-{index}.example.com"), base: "/telegram/web/".to_string(), bootstrap: format!("{index:043}"), @@ -96,6 +98,7 @@ impl Case { self.value as u64 - 1, self.diagnostics, self.method, + &self.jitter, rng, ) } @@ -121,6 +124,7 @@ impl Case { self.value as u64 - 1, self.diagnostics, self.method, + &self.jitter, nonce, ) } @@ -223,3 +227,25 @@ fn concurrent_responses_keep_nonces_and_configuration_response_owned() { assert!(template(diagnostics).contains("__BASE_PREFIX__")); } } + +#[test] +fn retry_jitter_is_rendered_per_page_without_changing_existing_pages() { + let rng = SecureRandom::new(); + let mut case = Case::new(1); + let old = case.render(&rng); + case.jitter.enabled = true; + case.jitter.min_ms = 10; + case.jitter.max_ms = 301; + case.jitter.distribution = crate::config::WebRetryJitterDistribution::Triangular; + let new = case.render(&rng); + assert!( + old.body + .contains("{enabled:false,min:0,max:250,peak:125,triangular:false}") + ); + assert!( + new.body + .contains("{enabled:true,min:10,max:301,peak:155.5,triangular:true}") + ); + let nonce = case.assert_page(&new); + assert_eq!(new.body, case.reference(&nonce).body); +} diff --git a/src/web/bridge/request.js b/src/web/bridge/request.js index edc9422..192b4cd 100644 --- a/src/web/bridge/request.js +++ b/src/web/bridge/request.js @@ -1,5 +1,21 @@ (()=>{'use strict'; function create(settings){ + // Snapshot policy per page, including during session recovery. + const jitter=Object.freeze(Object.assign({},settings.jitter||{enabled:false})); + const randomWord=jitter.enabled?new Uint32Array(1):null; + function randomUnit(){ + try{if(globalThis.crypto&&typeof globalThis.crypto.getRandomValues==='function'){ + globalThis.crypto.getRandomValues(randomWord);return randomWord[0]/4294967296; + }}catch(error){} + return Math.random(); + } + function jitterMs(){ + const a=jitter.min,b=jitter.max;if(a===b)return a; + const u=randomUnit(); + if(!jitter.triangular)return a+Math.floor(u*(b-a+1)); + const c=jitter.peak,span=b-a; + return Math.round(u<(c-a)/span?a+Math.sqrt(u*span*(c-a)):b-Math.sqrt((1-u)*span*(b-c))); + } const pause=(milliseconds,signal)=>new Promise((resolve,reject)=>{ if(signal&&signal.aborted){reject(new Error('request aborted'));return} const timer=setTimeout(done,milliseconds);function done(){if(signal)signal.removeEventListener('abort',abort);resolve()} @@ -15,10 +31,12 @@ function create(settings){ // 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; + // Enabled policies must never shorten a valid server minimum to the legacy 30 s cap. + if(jitter.enabled&&/^\d+$/.test(header))return Number(header)*1000; const seconds=Number(header); - if(Number.isFinite(seconds)&&seconds>=0)return Math.min(seconds*1000,30000); + if(Number.isFinite(seconds)&&seconds>=0)return jitter.enabled?seconds*1000:Math.min(seconds*1000,30000); const when=Date.parse(header); - if(Number.isFinite(when)){const delta=when-Date.now();return delta>0?Math.min(delta,30000):0} + if(Number.isFinite(when)){const delta=when-Date.now();return delta>0?(jitter.enabled?delta:Math.min(delta,30000)):0} return 0; } function retryableStatus(status){return status===408||status===429||status===502||status===503||status===504} @@ -65,7 +83,7 @@ function create(settings){ }finally{clearTimeout(timer);if(external)external.removeEventListener('abort',abort)} const after=Math.min(deadline-Date.now(),remainingBudget?remainingBudget():Infinity);if(attempt>=attempts()||after<=0)break; settings.retrying(); - const backoff=wait||delay+Math.floor(Math.random()*Math.max(1,delay/4)); + const backoff=jitter.enabled?(wait||delay)+jitterMs():wait||delay+Math.floor(Math.random()*Math.max(1,delay/4)); await pause(Math.min(backoff,after),external);delay=Math.min(delay*2,2000); } throw settings.failure(lastReason,'carrier retry limit reached'); diff --git a/src/web/bridge/retry_jitter_tests.js b/src/web/bridge/retry_jitter_tests.js new file mode 100644 index 0000000..63a4840 --- /dev/null +++ b/src/web/bridge/retry_jitter_tests.js @@ -0,0 +1,72 @@ +'use strict'; +const assert=require('node:assert/strict'); +const fs=require('node:fs'); +const vm=require('node:vm'); +const path=require('node:path'); +const source=fs.readFileSync(path.join(__dirname,'request.js'),'utf8'); +const flush=async()=>{for(let i=0;i<20;i++)await Promise.resolve()}; +function harness(policy,unit=0.5,cryptoMode='ok',budget=90000){ + let now=0,next=1,calls=0,draws=0;const timers=new Map(); + const context={Uint32Array,AbortController,Date:class extends Date{static now(){return now}}, + Math:Object.assign(Object.create(Math),{random:()=>unit}), + setTimeout:(fn,ms)=>{const id=next++;timers.set(id,{fn,ms});return id},clearTimeout:id=>timers.delete(id), + fetch:async()=>{calls++;return {status:503,headers:new Headers({'Retry-After':'1','X-Telemt-Retry-After-Ms':'250'})}} + }; + if(cryptoMode!=='absent')context.crypto={getRandomValues(word){draws++;if(cryptoMode==='throw')throw Error('unavailable');word[0]=Math.floor(unit*4294967296);return word}}; + vm.createContext(context);vm.runInContext(source,context); + const settings={jitter:policy,base:()=>'',closed:()=>false,retryMs:()=>budget,longPollMs:()=>0,requestMs:()=>10000, + batchLimit:()=>8,cancel(){},retrying(){},failure:(_,message)=>new Error(message),reason:()=>'',read:async()=>new ArrayBuffer(0)}; + const client=context.TelemtBridgeRequest.create(settings); + return {context,timers,client,get calls(){return calls},get draws(){return draws}, + async sample(status=503,headers={'Retry-After':'1','X-Telemt-Retry-After-Ms':'250'}){ + context.fetch=async()=>{calls++;if(status===0)throw Error('network');return {status,headers:new Headers(headers)}}; + const controller=new AbortController();const promise=client.send('/api/v1/up',{signal:controller.signal},null).catch(()=>{}); + await flush();const timer=[...timers.values()][0];assert.ok(timer);const delay=timer.ms; + controller.abort();await promise;assert.equal(timers.size,0);return delay; + },async expire(){const item=[...timers.entries()][0];assert.ok(item);timers.delete(item[0]);now+=item[1].ms;item[1].fn();await flush()} + }; +} +module.exports=async()=>{ + const config={enabled:true,min:0,max:250,peak:125,triangular:false}; + for(const mode of ['ok','absent','throw']){ + const h=harness(config,0.5,mode);assert.equal(await h.sample(),375); + } + for(const triangular of [false,true]){ + for(const peak of [0,25,100]){ + let sum=0; + for(let i=0;i<256;i++){ + const h=harness({...config,min:0,max:100,peak,triangular},(i+0.5)/256); + const value=await h.sample()-250;assert.ok(value>=0&&value<=100);sum+=value; + } + assert.ok(Math.abs(sum/256-(triangular?(100+peak)/3:50))<1); + } + } + for(const u of [0,0.25,0.5,0.75,1-1/4294967296]){ + const h=harness({...config,min:0,max:100,peak:50,triangular:true},u); + const expected=[0,35,50,65,100][[0,0.25,0.5,0.75,1-1/4294967296].indexOf(u)]; + assert.equal(await h.sample()-250,expected); + } + const frozen={...config,min:100,max:100};const h=harness(frozen);frozen.min=0;frozen.max=0; + assert.equal(await h.sample(),350);assert.equal(h.draws,0); + for(const status of [0,408,429,502,503,504])assert.equal(await h.sample(status,{}),350); + assert.equal(await h.sample(429,{'Retry-After':'2'}),2100); + assert.equal(await h.sample(503,{'Retry-After':'40'}),40100); + assert.equal(await h.sample(503,{'Retry-After':new Date(40000).toUTCString()}),40100); + assert.equal(await h.sample(503,{'Retry-After':'bad'}),350); + // Budget exhaustion must not turn a long minimum into an early retry. + const bounded=harness(config,0.5,'ok',1000); + bounded.context.fetch=async()=>({status:503,headers:new Headers({'Retry-After':'99999999999999999999999999999999999999999'})}); + const pending=bounded.client.send('/api/v1/up',{},null).then(()=>assert.fail('unexpected success'),()=>{}); + await flush();await bounded.expire();await pending;assert.equal(bounded.timers.size,0); + // A new draw is made per retry while the exponential base and attempt cap remain. + const retries=harness(config,0.5);let attempts=0; + retries.context.fetch=async()=>{attempts++;throw Error('network')}; + const failed=retries.client.send('/api/v1/up',{},null).catch(()=>{});await flush(); + for(const delay of [375,625,1125,2125,2125,2125,2125,2125]){ + assert.equal([...retries.timers.values()][0].ms,delay);await retries.expire(); + } + await failed;assert.equal(attempts,9);assert.equal(retries.draws,8);assert.equal(retries.timers.size,0); + // Successful responses do not draw entropy or wait for jitter. + const success=harness(config);success.context.fetch=async()=>({status:204,headers:new Headers()}); + await success.client.send('/api/v1/up',{},null);assert.equal(success.draws,0);assert.equal(success.timers.size,0); +}; diff --git a/src/web/bridge/runtime.js b/src/web/bridge/runtime.js index 1271bee..35b2a3d 100644 --- a/src/web/bridge/runtime.js +++ b/src/web/bridge/runtime.js @@ -30,6 +30,7 @@ const failureReason=(error,fallback)=>error&&canonicalFailures.includes(error.te const status=__STATUS_FUNCTION__; const socketURL=()=>relayBase.replace(/^https:/,'wss:')+'/api/v1/ws'; const requestClient=requestSupport.create({ + jitter:Object.freeze(__RETRY_JITTER__), base:()=>relayBase,closed:()=>closed,retryMs:()=>bridgeRetryMs,longPollMs:()=>longPollMs,requestMs:()=>bridgeRequestMs, batchLimit:()=>batchLimit,read:(response,limit,exact,signal)=>responseBody.read(response,limit,exact,signal),cancel:responseBody.cancel, failure,reason:failureReason,retrying:()=>status('reconnecting') diff --git a/src/web/bridge/tests.rs b/src/web/bridge/tests.rs index 7eca039..1e6db1c 100644 --- a/src/web/bridge/tests.rs +++ b/src/web/bridge/tests.rs @@ -23,6 +23,7 @@ fn carrier_method_is_page_owned_and_used_by_every_https_request() { 0, true, method, + &crate::config::WebRetryJitterConfig::default(), &SecureRandom::new(), ); assert!(!page.body.contains("__")); @@ -80,6 +81,7 @@ fn render_page(bootstrap: &str, candidate_count: usize) -> BridgePage { 0, false, WebCarrierMethod::Post, + &crate::config::WebRetryJitterConfig::default(), &SecureRandom::new(), ) } @@ -105,6 +107,7 @@ fn render_diagnostic_page(bootstrap: &str) -> BridgePage { 0, true, WebCarrierMethod::Post, + &crate::config::WebRetryJitterConfig::default(), &SecureRandom::new(), ) } @@ -157,6 +160,7 @@ fn rendered_page_resolves_carriers_against_the_exact_base_path() { 0, true, WebCarrierMethod::Post, + &crate::config::WebRetryJitterConfig::default(), &SecureRandom::new(), ); @@ -211,6 +215,7 @@ fn rendered_page_embeds_the_configured_bridge_timing_policy() { 4, false, WebCarrierMethod::Post, + &crate::config::WebRetryJitterConfig::default(), &SecureRandom::new(), ); @@ -265,6 +270,7 @@ fn disabled_negotiation_does_not_arm_a_carrier_deadline() { 0, false, WebCarrierMethod::Post, + &crate::config::WebRetryJitterConfig::default(), &SecureRandom::new(), ); assert!(page.body.contains( diff --git a/src/web/http.rs b/src/web/http.rs index c8e726c..9e76efb 100644 --- a/src/web/http.rs +++ b/src/web/http.rs @@ -432,6 +432,7 @@ async fn handle_root( config.web.timeouts.carrier_probe_coalesce_ms, config.web.debug.bridge_diagnostics_enabled(), config.web.carrier_method, + &config.web.retry_jitter, &generation.rng, ); let mut response = full_response(StatusCode::OK, Bytes::from(page.body));