mirror of
https://github.com/telemt/telemt.git
synced 2026-11-01 21:31:58 +03:00
Retry Jitter
This commit is contained in:
@@ -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.
|
// Effective process policy must reject patterns until the explicit restart has happened.
|
||||||
#[path = "wildcard_tests.rs"]
|
#[path = "wildcard_tests.rs"]
|
||||||
mod wildcard_tests;
|
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);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -260,7 +260,10 @@ const LISTENER_CONFIG_KEYS: &[&str] = &[
|
|||||||
"web_trusted_proxy_cidrs",
|
"web_trusted_proxy_cidrs",
|
||||||
];
|
];
|
||||||
|
|
||||||
|
const WEB_RETRY_JITTER_KEYS: &[&str] = &["enabled", "min_ms", "max_ms", "distribution", "peak_ms"];
|
||||||
|
|
||||||
const WEB_CONFIG_KEYS: &[&str] = &[
|
const WEB_CONFIG_KEYS: &[&str] = &[
|
||||||
|
"retry_jitter",
|
||||||
"allow_wildcard_hosts",
|
"allow_wildcard_hosts",
|
||||||
"enabled",
|
"enabled",
|
||||||
"carrier",
|
"carrier",
|
||||||
|
|||||||
@@ -36,6 +36,7 @@ fn known_config_keys_for_suggestion() -> Vec<&'static str> {
|
|||||||
CONNTRACK_CONTROL_CONFIG_KEYS,
|
CONNTRACK_CONTROL_CONFIG_KEYS,
|
||||||
LISTENER_CONFIG_KEYS,
|
LISTENER_CONFIG_KEYS,
|
||||||
WEB_CONFIG_KEYS,
|
WEB_CONFIG_KEYS,
|
||||||
|
WEB_RETRY_JITTER_KEYS,
|
||||||
WEB_LIMITS_CONFIG_KEYS,
|
WEB_LIMITS_CONFIG_KEYS,
|
||||||
WEB_WILDCARD_LIMITS_CONFIG_KEYS,
|
WEB_WILDCARD_LIMITS_CONFIG_KEYS,
|
||||||
WEB_DEBUG_CONFIG_KEYS,
|
WEB_DEBUG_CONFIG_KEYS,
|
||||||
@@ -236,6 +237,13 @@ pub(super) fn collect_unknown_config_keys(parsed_toml: &toml::Value) -> Vec<Unkn
|
|||||||
&["web"],
|
&["web"],
|
||||||
WEB_CONFIG_KEYS,
|
WEB_CONFIG_KEYS,
|
||||||
);
|
);
|
||||||
|
check_known_table(
|
||||||
|
parsed_toml,
|
||||||
|
&mut unknown,
|
||||||
|
&known_for_suggestion,
|
||||||
|
&["web", "retry_jitter"],
|
||||||
|
WEB_RETRY_JITTER_KEYS,
|
||||||
|
);
|
||||||
check_known_table(
|
check_known_table(
|
||||||
parsed_toml,
|
parsed_toml,
|
||||||
&mut unknown,
|
&mut unknown,
|
||||||
|
|||||||
@@ -75,6 +75,11 @@ pub(super) fn validate_source(config: &mut ProxyConfig) -> Result<()> {
|
|||||||
));
|
));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
config
|
||||||
|
.web
|
||||||
|
.retry_jitter
|
||||||
|
.validate()
|
||||||
|
.map_err(|message| ProxyError::Config(message.into()))?;
|
||||||
validate_limits(&config.web.limits, config.web.allow_wildcard_hosts)?;
|
validate_limits(&config.web.limits, config.web.allow_wildcard_hosts)?;
|
||||||
debug::validate(&config.web.debug, &config.web.limits)?;
|
debug::validate(&config.web.debug, &config.web.limits)?;
|
||||||
let carriers = negotiation::validate(&config.web)?;
|
let carriers = negotiation::validate(&config.web)?;
|
||||||
|
|||||||
+2
-1
@@ -58,7 +58,8 @@ pub use server::{
|
|||||||
pub use web::{
|
pub use web::{
|
||||||
WebCarrierNegotiationAggressiveness, WebConfig, WebDecoyConfig, WebDecoyFastTrackMode,
|
WebCarrierNegotiationAggressiveness, WebConfig, WebDecoyConfig, WebDecoyFastTrackMode,
|
||||||
WebDecoyResolve, WebHttpConnectionCapacityAction, WebLimitsConfig, WebProfileConfig,
|
WebDecoyResolve, WebHttpConnectionCapacityAction, WebLimitsConfig, WebProfileConfig,
|
||||||
WebSecretMode, WebTimeoutsConfig, WebVhostConfig, WebWildcardLimitsConfig,
|
WebRetryJitterConfig, WebRetryJitterDistribution, WebSecretMode, WebTimeoutsConfig,
|
||||||
|
WebVhostConfig, WebWildcardLimitsConfig,
|
||||||
};
|
};
|
||||||
pub(crate) use web::{
|
pub(crate) use web::{
|
||||||
WebDecoyDnsSnapshot, WebRuntimeConfig, WebRuntimeDecoy, WebRuntimeProfile,
|
WebDecoyDnsSnapshot, WebRuntimeConfig, WebRuntimeDecoy, WebRuntimeProfile,
|
||||||
|
|||||||
@@ -14,6 +14,9 @@ mod wildcard;
|
|||||||
pub use wildcard::WebWildcardLimitsConfig;
|
pub use wildcard::WebWildcardLimitsConfig;
|
||||||
|
|
||||||
// Serialized WEB defaults remain separate from the runtime data model.
|
// Serialized WEB defaults remain separate from the runtime data model.
|
||||||
|
mod retry;
|
||||||
|
pub use retry::{WebRetryJitterConfig, WebRetryJitterDistribution};
|
||||||
|
|
||||||
mod defaults;
|
mod defaults;
|
||||||
use defaults::*;
|
use defaults::*;
|
||||||
// Decoy fast-track policy remains isolated from the bulky WEB data model.
|
// Decoy fast-track policy remains isolated from the bulky WEB data model.
|
||||||
@@ -469,6 +472,9 @@ pub struct WebConfig {
|
|||||||
/// WEB lifecycle deadlines.
|
/// WEB lifecycle deadlines.
|
||||||
#[serde(default)]
|
#[serde(default)]
|
||||||
pub timeouts: WebTimeoutsConfig,
|
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.
|
/// Exact hostname and opted-in wildcard groups served by WEB listeners.
|
||||||
#[serde(default)]
|
#[serde(default)]
|
||||||
pub vhosts: Vec<WebVhostConfig>,
|
pub vhosts: Vec<WebVhostConfig>,
|
||||||
@@ -517,6 +523,7 @@ impl Default for WebConfig {
|
|||||||
limits: WebLimitsConfig::default(),
|
limits: WebLimitsConfig::default(),
|
||||||
debug: WebDebugConfig::default(),
|
debug: WebDebugConfig::default(),
|
||||||
timeouts: WebTimeoutsConfig::default(),
|
timeouts: WebTimeoutsConfig::default(),
|
||||||
|
retry_jitter: WebRetryJitterConfig::default(),
|
||||||
vhosts: Vec::new(),
|
vhosts: Vec::new(),
|
||||||
runtime: None,
|
runtime: None,
|
||||||
decoy_dns: Arc::default(),
|
decoy_dns: Arc::default(),
|
||||||
|
|||||||
@@ -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<u32>,
|
||||||
|
}
|
||||||
|
|
||||||
|
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::<WebRetryJitterConfig>(&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::<WebRetryJitterConfig>(json).is_err());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -38,6 +38,7 @@ pub(crate) fn render(
|
|||||||
carrier_probe_coalesce_ms: u64,
|
carrier_probe_coalesce_ms: u64,
|
||||||
bridge_diagnostics_enabled: bool,
|
bridge_diagnostics_enabled: bool,
|
||||||
carrier_method: WebCarrierMethod,
|
carrier_method: WebCarrierMethod,
|
||||||
|
retry_jitter: &crate::config::WebRetryJitterConfig,
|
||||||
rng: &SecureRandom,
|
rng: &SecureRandom,
|
||||||
) -> BridgePage {
|
) -> BridgePage {
|
||||||
let mut nonce = [0u8; 18];
|
let mut nonce = [0u8; 18];
|
||||||
@@ -47,6 +48,7 @@ pub(crate) fn render(
|
|||||||
let body = template(bridge_diagnostics_enabled)
|
let body = template(bridge_diagnostics_enabled)
|
||||||
.replace("__NONCE__", &nonce)
|
.replace("__NONCE__", &nonce)
|
||||||
.replace("__HOST__", host)
|
.replace("__HOST__", host)
|
||||||
|
.replace("__RETRY_JITTER__", &retry_jitter.bridge_policy())
|
||||||
.replace("__BASE_PREFIX__", base_prefix)
|
.replace("__BASE_PREFIX__", base_prefix)
|
||||||
.replace("__BOOTSTRAP__", bootstrap)
|
.replace("__BOOTSTRAP__", bootstrap)
|
||||||
.replace("__CARRIER_METHOD__", carrier_method.as_str())
|
.replace("__CARRIER_METHOD__", carrier_method.as_str())
|
||||||
|
|||||||
@@ -11,7 +11,7 @@ function renderedPage(){
|
|||||||
page=page.replace('__DIAGNOSTIC_RUNTIME__','');
|
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'));
|
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'));
|
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]??''));
|
return page.replace(/__([A-Z_]+)__/g,(all,key)=>key.startsWith('DIAGNOSTIC_')?'':String(values[key]??''));
|
||||||
}
|
}
|
||||||
function frame(type,id=0,payload=[]){
|
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);
|
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);
|
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=>{
|
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();
|
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();
|
assert.equal(env.pending('/api/v1/up')[0].options.body.byteLength,body.byteLength);env.close();await flush();
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ use std::process::{Command, Stdio};
|
|||||||
/// Exercises the actual generated page with deterministic network and native-boundary doubles.
|
/// Exercises the actual generated page with deterministic network and native-boundary doubles.
|
||||||
pub(super) fn run(page: &super::BridgePage) {
|
pub(super) fn run(page: &super::BridgePage) {
|
||||||
let mut child = Command::new("node")
|
let mut child = Command::new("node")
|
||||||
|
.current_dir(concat!(env!("CARGO_MANIFEST_DIR"), "/src/web/bridge"))
|
||||||
.args(["-e", include_str!("behavior_tests.js"), "--", "--stdin"])
|
.args(["-e", include_str!("behavior_tests.js"), "--", "--stdin"])
|
||||||
.stdin(Stdio::piped())
|
.stdin(Stdio::piped())
|
||||||
.stdout(Stdio::piped())
|
.stdout(Stdio::piped())
|
||||||
|
|||||||
@@ -23,6 +23,7 @@ pub(super) fn render(
|
|||||||
carrier_probe_coalesce_ms: u64,
|
carrier_probe_coalesce_ms: u64,
|
||||||
bridge_diagnostics_enabled: bool,
|
bridge_diagnostics_enabled: bool,
|
||||||
carrier_method: WebCarrierMethod,
|
carrier_method: WebCarrierMethod,
|
||||||
|
retry_jitter: &crate::config::WebRetryJitterConfig,
|
||||||
nonce: &str,
|
nonce: &str,
|
||||||
) -> BridgePage {
|
) -> BridgePage {
|
||||||
let diagnostic_script = if bridge_diagnostics_enabled {
|
let diagnostic_script = if bridge_diagnostics_enabled {
|
||||||
@@ -110,6 +111,7 @@ pub(super) fn render(
|
|||||||
)
|
)
|
||||||
.replace("__NONCE__", &nonce)
|
.replace("__NONCE__", &nonce)
|
||||||
.replace("__HOST__", host)
|
.replace("__HOST__", host)
|
||||||
|
.replace("__RETRY_JITTER__", &retry_jitter.bridge_policy())
|
||||||
.replace("__BASE_PREFIX__", base_prefix)
|
.replace("__BASE_PREFIX__", base_prefix)
|
||||||
.replace("__BOOTSTRAP__", bootstrap)
|
.replace("__BOOTSTRAP__", bootstrap)
|
||||||
.replace("__CARRIER_METHOD__", carrier_method.as_str())
|
.replace("__CARRIER_METHOD__", carrier_method.as_str())
|
||||||
|
|||||||
@@ -55,11 +55,13 @@ struct Case {
|
|||||||
negotiation: bool,
|
negotiation: bool,
|
||||||
candidates: usize,
|
candidates: usize,
|
||||||
value: usize,
|
value: usize,
|
||||||
|
jitter: crate::config::WebRetryJitterConfig,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Case {
|
impl Case {
|
||||||
fn new(index: usize) -> Self {
|
fn new(index: usize) -> Self {
|
||||||
Self {
|
Self {
|
||||||
|
jitter: crate::config::WebRetryJitterConfig::default(),
|
||||||
host: format!("proxy-{index}.example.com"),
|
host: format!("proxy-{index}.example.com"),
|
||||||
base: "/telegram/web/".to_string(),
|
base: "/telegram/web/".to_string(),
|
||||||
bootstrap: format!("{index:043}"),
|
bootstrap: format!("{index:043}"),
|
||||||
@@ -96,6 +98,7 @@ impl Case {
|
|||||||
self.value as u64 - 1,
|
self.value as u64 - 1,
|
||||||
self.diagnostics,
|
self.diagnostics,
|
||||||
self.method,
|
self.method,
|
||||||
|
&self.jitter,
|
||||||
rng,
|
rng,
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
@@ -121,6 +124,7 @@ impl Case {
|
|||||||
self.value as u64 - 1,
|
self.value as u64 - 1,
|
||||||
self.diagnostics,
|
self.diagnostics,
|
||||||
self.method,
|
self.method,
|
||||||
|
&self.jitter,
|
||||||
nonce,
|
nonce,
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
@@ -223,3 +227,25 @@ fn concurrent_responses_keep_nonces_and_configuration_response_owned() {
|
|||||||
assert!(template(diagnostics).contains("__BASE_PREFIX__"));
|
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);
|
||||||
|
}
|
||||||
|
|||||||
@@ -1,5 +1,21 @@
|
|||||||
(()=>{'use strict';
|
(()=>{'use strict';
|
||||||
function create(settings){
|
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)=>{
|
const pause=(milliseconds,signal)=>new Promise((resolve,reject)=>{
|
||||||
if(signal&&signal.aborted){reject(new Error('request aborted'));return}
|
if(signal&&signal.aborted){reject(new Error('request aborted'));return}
|
||||||
const timer=setTimeout(done,milliseconds);function done(){if(signal)signal.removeEventListener('abort',abort);resolve()}
|
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.
|
// 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(response.status===503&&header==='1'&&response.headers.get('X-Telemt-Retry-After-Ms')==='250')return 250;
|
||||||
if(!header)return 0;
|
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);
|
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);
|
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;
|
return 0;
|
||||||
}
|
}
|
||||||
function retryableStatus(status){return status===408||status===429||status===502||status===503||status===504}
|
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)}
|
}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;
|
const after=Math.min(deadline-Date.now(),remainingBudget?remainingBudget():Infinity);if(attempt>=attempts()||after<=0)break;
|
||||||
settings.retrying();
|
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);
|
await pause(Math.min(backoff,after),external);delay=Math.min(delay*2,2000);
|
||||||
}
|
}
|
||||||
throw settings.failure(lastReason,'carrier retry limit reached');
|
throw settings.failure(lastReason,'carrier retry limit reached');
|
||||||
|
|||||||
@@ -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);
|
||||||
|
};
|
||||||
@@ -30,6 +30,7 @@ const failureReason=(error,fallback)=>error&&canonicalFailures.includes(error.te
|
|||||||
const status=__STATUS_FUNCTION__;
|
const status=__STATUS_FUNCTION__;
|
||||||
const socketURL=()=>relayBase.replace(/^https:/,'wss:')+'/api/v1/ws';
|
const socketURL=()=>relayBase.replace(/^https:/,'wss:')+'/api/v1/ws';
|
||||||
const requestClient=requestSupport.create({
|
const requestClient=requestSupport.create({
|
||||||
|
jitter:Object.freeze(__RETRY_JITTER__),
|
||||||
base:()=>relayBase,closed:()=>closed,retryMs:()=>bridgeRetryMs,longPollMs:()=>longPollMs,requestMs:()=>bridgeRequestMs,
|
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,
|
batchLimit:()=>batchLimit,read:(response,limit,exact,signal)=>responseBody.read(response,limit,exact,signal),cancel:responseBody.cancel,
|
||||||
failure,reason:failureReason,retrying:()=>status('reconnecting')
|
failure,reason:failureReason,retrying:()=>status('reconnecting')
|
||||||
|
|||||||
@@ -23,6 +23,7 @@ fn carrier_method_is_page_owned_and_used_by_every_https_request() {
|
|||||||
0,
|
0,
|
||||||
true,
|
true,
|
||||||
method,
|
method,
|
||||||
|
&crate::config::WebRetryJitterConfig::default(),
|
||||||
&SecureRandom::new(),
|
&SecureRandom::new(),
|
||||||
);
|
);
|
||||||
assert!(!page.body.contains("__"));
|
assert!(!page.body.contains("__"));
|
||||||
@@ -80,6 +81,7 @@ fn render_page(bootstrap: &str, candidate_count: usize) -> BridgePage {
|
|||||||
0,
|
0,
|
||||||
false,
|
false,
|
||||||
WebCarrierMethod::Post,
|
WebCarrierMethod::Post,
|
||||||
|
&crate::config::WebRetryJitterConfig::default(),
|
||||||
&SecureRandom::new(),
|
&SecureRandom::new(),
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
@@ -105,6 +107,7 @@ fn render_diagnostic_page(bootstrap: &str) -> BridgePage {
|
|||||||
0,
|
0,
|
||||||
true,
|
true,
|
||||||
WebCarrierMethod::Post,
|
WebCarrierMethod::Post,
|
||||||
|
&crate::config::WebRetryJitterConfig::default(),
|
||||||
&SecureRandom::new(),
|
&SecureRandom::new(),
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
@@ -157,6 +160,7 @@ fn rendered_page_resolves_carriers_against_the_exact_base_path() {
|
|||||||
0,
|
0,
|
||||||
true,
|
true,
|
||||||
WebCarrierMethod::Post,
|
WebCarrierMethod::Post,
|
||||||
|
&crate::config::WebRetryJitterConfig::default(),
|
||||||
&SecureRandom::new(),
|
&SecureRandom::new(),
|
||||||
);
|
);
|
||||||
|
|
||||||
@@ -211,6 +215,7 @@ fn rendered_page_embeds_the_configured_bridge_timing_policy() {
|
|||||||
4,
|
4,
|
||||||
false,
|
false,
|
||||||
WebCarrierMethod::Post,
|
WebCarrierMethod::Post,
|
||||||
|
&crate::config::WebRetryJitterConfig::default(),
|
||||||
&SecureRandom::new(),
|
&SecureRandom::new(),
|
||||||
);
|
);
|
||||||
|
|
||||||
@@ -265,6 +270,7 @@ fn disabled_negotiation_does_not_arm_a_carrier_deadline() {
|
|||||||
0,
|
0,
|
||||||
false,
|
false,
|
||||||
WebCarrierMethod::Post,
|
WebCarrierMethod::Post,
|
||||||
|
&crate::config::WebRetryJitterConfig::default(),
|
||||||
&SecureRandom::new(),
|
&SecureRandom::new(),
|
||||||
);
|
);
|
||||||
assert!(page.body.contains(
|
assert!(page.body.contains(
|
||||||
|
|||||||
@@ -432,6 +432,7 @@ async fn handle_root(
|
|||||||
config.web.timeouts.carrier_probe_coalesce_ms,
|
config.web.timeouts.carrier_probe_coalesce_ms,
|
||||||
config.web.debug.bridge_diagnostics_enabled(),
|
config.web.debug.bridge_diagnostics_enabled(),
|
||||||
config.web.carrier_method,
|
config.web.carrier_method,
|
||||||
|
&config.web.retry_jitter,
|
||||||
&generation.rng,
|
&generation.rng,
|
||||||
);
|
);
|
||||||
let mut response = full_response(StatusCode::OK, Bytes::from(page.body));
|
let mut response = full_response(StatusCode::OK, Bytes::from(page.body));
|
||||||
|
|||||||
Reference in New Issue
Block a user