fix(退避): 严格单调断言偶发失败——是断言过严，不是 jitter 源码 bug

## 结论：断言过严（抖动实现本身正确，未改退避算法） `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/ 未触碰。
2026-10-27 08:09:52 +08:00
1 parent 14395692e4
commit e9c02c981e
@@ -68,8 +68,15 @@ impl<T> RingBuffer<T> {
 /// 指数退避 + 随机抖动
 ///
 /// 规则（见方案 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] 区间内的值。
@@ -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");