From e9c02c981ea607272cb3d574dde5205701d404b2 Mon Sep 17 00:00:00 2001 From: PMon Dev Date: Tue, 27 Oct 2026 08:09:52 +0800 Subject: [PATCH] =?UTF-8?q?fix(=E9=80=80=E9=81=BF):=20=E4=B8=A5=E6=A0=BC?= =?UTF-8?q?=E5=8D=95=E8=B0=83=E6=96=AD=E8=A8=80=E5=81=B6=E5=8F=91=E5=A4=B1?= =?UTF-8?q?=E8=B4=A5=E2=80=94=E2=80=94=E6=98=AF=E6=96=AD=E8=A8=80=E8=BF=87?= =?UTF-8?q?=E4=B8=A5=EF=BC=8C=E4=B8=8D=E6=98=AF=20jitter=20=E6=BA=90?= =?UTF-8?q?=E7=A0=81=20bug?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 结论：断言过严（抖动实现本身正确，未改退避算法） `apply_jitter_downward` 只向下扣 [0, capped*ratio]，结果恒落在 [capped*(1-ratio), capped] 内，**绝不突破 max_ms** —— 上限这条硬承诺没被破坏， 抖动叠加方式也没有写错，不存在"源码 bug"那一路。 严格单调不成立是设计内的必然：封顶后 capped 恒为 60s，扣减量 offset = (now_ms() ^ capped.rotate_left(13)) % (span+1)，每跨 1 毫秒 offset 就 +1， 于是后一个样本比前一个正好小 1ms（实测 59009 → 59008、50901 → 50900）。 原断言要求"相邻样本单调不减"，等于要求这件不可能的事。 ## 复现（真复现，非推测） - 并行单跑该用例：48 次内命中 `退避应单调不减: [... 57955, 57954, 57954 ...]` - 变异回严格单调后，16 进程 × --test-threads=61 并发整包跑：第 1600 次命中 `[... 50901, 50901, 50901, 50900 ...]` 单跑命中率约 1/1000（12 次调用里只有封顶区那段约 1µs 窗口可能跨毫秒）， 这正是它"平时全绿、负载高才红"的原因。 ## 改动 - transport.rs 主用例改按**档位表**断言（capped(n)=min(base*2^n, cap)）： 每个样本必须落在本档位抖动带 [level*(1-ratio), level]；相邻样本回落不得超过 一个抖动幅度（这才是"防退化"该守的不变量）；末次必须真的顶进封顶带。 注释写清为什么不能要求严格单调，防后人改回去。 - 新增 capped_backoff_dips_within_jitter_band：在封顶区**跨毫秒**采样， 断言"必然观察到回落，且回落幅度 ≤ 一个抖动幅度"，把"回落是正常行为" 钉成可执行断言（而不是靠注释约定）。 - 新增 Backoff::base_delay_ms / max_delay_ms / jitter_ratio 只读访问器： 测试不再重抄 1000 / 60_000 / 0.3，避免第二个真相源。 - buffer.rs 头注释 "±jitter" 与实现（只向下）不符，一并改正。 ## 变异自证 1) 把 dips 用例的容差断言改成 `after >= before`（严格单调）→ 确定性变红 3/3， 报文 `45363 → 45362`。 2) 把主用例容差断言改回 `w[1] >= w[0]` → 上述并发压测第 1600 次变红， 特征与原报告一致；恢复后 500 次整包并发复跑全绿。 ## 测试 改前 60 passed / 0 failed；改后 61 passed / 0 failed（净增 1 条）。 仅动 rust-core/src/buffer.rs 与 rust-core/src/transport.rs，server/ tests/ client/ deploy/ 未触碰。 --- rust-core/src/buffer.rs | 31 ++++++++++- rust-core/src/transport.rs | 106 +++++++++++++++++++++++++++++++++++-- 2 files changed, 131 insertions(+), 6 deletions(-) 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");