diff --git a/rust-core/src/buffer.rs b/rust-core/src/buffer.rs index 36c6a2a..2465433 100644 --- a/rust-core/src/buffer.rs +++ b/rust-core/src/buffer.rs @@ -68,8 +68,15 @@ impl RingBuffer { /// 指数退避 + 随机抖动 /// /// 规则(见方案 V4.0 13.2): -/// delay(n) = min(base * 2^n, max),再叠加 ±jitter 比例的随机量 +/// delay(n) = min(base * 2^n, max),再**向下**扣掉 [0, jitter] 比例的随机量 /// 抖动的作用:即使海量设备同时断线,重连时刻也会被打散,避免"重连风暴"。 +/// +/// ⚠️ 抖动**只向下,不是 ±**(原注释写成 ±,与实现不符,已改): +/// 上限是硬承诺,正向抖会在 `min(base*2^n, max)` 之后突破 max_ms。 +/// 由此带出一个必须知道的后果:**封顶之后相邻两次延迟不保证单调不减** —— +/// 后一次可能比前一次小(实测 59009 → 59008,因为扣减量随 `now_ms()` 变化)。 +/// 调用方与测试都不要对带抖动的值断言严格单调,详见 +/// `transport::tests::capped_backoff_dips_within_jitter_band`。 pub struct Backoff { base_ms: u64, max_ms: u64, @@ -114,6 +121,28 @@ impl Backoff { pub fn attempt(&self) -> u32 { self.attempt } + + /// 起步延迟(毫秒)。 + /// + /// 对外只读:让测试与其他模块引用**策略真值**,而不必重抄 `1000` 这个字面量 + /// ("同一份数值写两处"是本仓库反复翻车的坑:指标字段、配对码契约、安装路径…)。 + pub fn base_delay_ms(&self) -> u64 { + self.base_ms + } + + /// 延迟上限(毫秒)。 + /// + /// 「重连风暴防线的硬承诺」的对外读数 —— 任何延迟都必须 ≤ 它。 + pub fn max_delay_ms(&self) -> u64 { + self.max_ms + } + + /// 抖动比例(0.0–1.0,只向下生效)。 + /// + /// 断言"抖动幅度没超设计"时需要它;不要在别处硬编码 0.3。 + pub fn jitter_ratio(&self) -> f64 { + self.jitter_ratio + } } /// 施加向下抖动:返回 [value*(1-ratio), value] 区间内的值。 diff --git a/rust-core/src/transport.rs b/rust-core/src/transport.rs index d112bbb..6590971 100644 --- a/rust-core/src/transport.rs +++ b/rust-core/src/transport.rs @@ -618,6 +618,8 @@ impl Transport { #[cfg(test)] mod tests { use super::*; + use crate::buffer::Backoff; + use crate::collector::now_ms; use serde_json::json; fn env(seq: u64) -> Envelope { @@ -854,42 +856,136 @@ mod tests { #[test] fn backoff_grows_and_is_capped_across_repeated_failures() { + // 策略值一律**从 Backoff 自己取**,不在测试里重抄 1000 / 60_000 / 0.3 + // (同一份数值写两处 = 第二个真相源,改策略时测试会"跟着一起错") + let policy = Backoff::default_policy(); + let cap = policy.max_delay_ms(); // 60_000 + let ratio = policy.jitter_ratio(); // 0.3 + let span = (cap as f64 * ratio) as u64; // 抖动最大幅度 18_000 + let mut tr = Transport::new("dev-1"); let mut now = 0u64; let mut delays = Vec::new(); + // 档位表:capped(n) = min(base * 2^n, cap) —— 与 Backoff::next_delay_ms 同源 + let mut level = policy.base_delay_ms(); - // 连续失败(期间从未连上):退避应逐级增长 + // 连续失败(期间从未连上):每个样本都必须落在**本档位**的抖动带内 for _ in 0..12 { let d = tr.on_disconnected(now); + let floor = level - (level as f64 * ratio) as u64; + assert!( + d <= level, + "退避突破本档位上限(抖动只应向下,不该超过 level={}): {}", + level, + d + ); + assert!(d >= floor, "抖动幅度超出设计(level={}): {}", level, d); delays.push(d); now += d; + level = level.saturating_mul(2).min(cap); } - // 单调不减(抖动只向下取,故允许相等) + // ── 增长性与"防退化"守卫 ──────────────────────────────────── + // 增长性由**档位表**保证(每档翻倍直到封顶),不靠相邻采样的大小关系。 + // + // ⚠️ 为什么**不能**断言"严格单调不减"(这条曾经这么写,会偶发失败): + // 抖动是向下取:delay = capped - offset,offset 由 `now_ms()` 派生。 + // 一旦封顶,capped 恒为 60s,offset 每跨 1 毫秒就 +1 —— 于是后一个样本 + // 完全可能比前一个**小 1ms**(实测 59009 → 59008,负载高时循环更容易 + // 撞上毫秒边界)。那是本函数**设计内的正常行为**,不是退避退化。 + // + // 真正要守住的不变量是:相邻样本的回落不超过一个抖动幅度。 for w in delays.windows(2) { - assert!(w[1] >= w[0], "退避应单调不减: {:?}", delays); + assert!( + w[1].saturating_add(span) >= w[0], + "相邻退避回落超过一个抖动幅度(这是退避退化,不是抖动): {:?}", + delays + ); } + // 封顶 60s —— 重连风暴防线的硬承诺 for d in &delays { - assert!(*d <= 60_000, "退避突破上限: {}", d); + assert!(*d <= cap, "退避突破上限: {}", d); } assert!( delays[delays.len() - 1] > 10_000, "退避应显著增长: {:?}", delays ); + // 末次必须真的顶进封顶带,否则"封顶"这条断言根本没被覆盖到 + assert!( + delays[delays.len() - 1] >= cap - span, + "末次退避应落在封顶带内: {:?}", + delays + ); // 一旦连上,退避档位归零:下次断线重新从 1s 起步(避免长期退避惩罚正常用户) tr.on_connected(now); assert_eq!(tr.backoff_attempt(), 0); let after_recovery = tr.on_disconnected(now); assert!( - after_recovery <= 1_000 && after_recovery > 0, + after_recovery <= policy.base_delay_ms() && after_recovery > 0, "重连成功后应回到起步退避: {}", after_recovery ); } + /// 封顶之后抖动会让相邻延迟**回落** —— 这不是退化,是设计。 + /// + /// 这个用例存在的唯一理由:上面那条用例曾把"相邻样本严格单调不减"当不变量, + /// 于是偶发失败(抖动让后一个样本小 1ms)。后人看到容差断言"太宽松"时, + /// 很可能又想改回严格单调 —— 本用例把"回落是正常行为"钉成可执行断言: + /// 在封顶区**跨毫秒**连续采样,必然观察到回落,且回落幅度不超过一个抖动幅度。 + #[test] + fn capped_backoff_dips_within_jitter_band() { + let policy = Backoff::default_policy(); + let cap = policy.max_delay_ms(); + let span = (cap as f64 * policy.jitter_ratio()) as u64; + + let mut tr = Transport::new("dev-1"); + let mut now = 0u64; + // 推到封顶档位:capped 此后恒为 cap,样本全落在 [cap-span, cap] + for _ in 0..8 { + now += tr.on_disconnected(now); + } + + let mut prev = tr.on_disconnected(now); + let mut dip: Option<(u64, u64)> = None; + for _ in 0..50 { + // 等过一次毫秒边界再取下一个样本:抖动的熵来自 `now_ms()`, + // 同一毫秒内取到的是同一个扣减量。不等就等于什么都没测却假装通过。 + let t = now_ms(); + while now_ms() == t { + std::hint::spin_loop(); + } + let cur = tr.on_disconnected(now); + assert!( + cur <= cap && cur.saturating_add(span) >= cap, + "封顶区样本必须落在抖动带 [{} , {}] 内: {}", + cap - span, + cap, + cur + ); + if cur < prev { + dip = Some((prev, cur)); + break; + } + prev = cur; + } + + let (before, after) = dip.expect( + "封顶区跨毫秒采样必须观察到回落:抖动扣减量随 now_ms() 变化,跨毫秒就该变小。\ + 这里失败说明抖动已不随时间变化(或方向被改成了向上)——请复查 apply_jitter_downward", + ); + assert!( + before - after <= span, + "回落幅度不得超过一个抖动幅度 {}: {} → {}", + span, + before, + after + ); + } + #[test] fn network_recovery_reconnects_immediately_without_waiting() { let mut tr = Transport::new("dev-1");