From 6c99ce51abe4256a10d9f745f5f90f281e3c23c9 Mon Sep 17 00:00:00 2001 From: anonsaber Date: Mon, 28 Sep 2026 18:38:00 +0800 Subject: [PATCH] =?UTF-8?q?state:=20XAUTOCLAIM=20=E7=A9=BA=E9=97=B2?= =?UTF-8?q?=E9=98=88=E5=80=BC=E6=94=B9=E7=94=B1=E8=BF=90=E8=A1=8C=E8=B6=85?= =?UTF-8?q?=E6=97=B6=E9=85=8D=E7=BD=AE=E6=8E=A8=E5=AF=BC=EF=BC=8C=E5=B0=81?= =?UTF-8?q?=E9=A1=B6=206h=EF=BC=8801a0e75f-f9b1=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 现状：read_batch 里 const SINGLE_EVENT_CEILING_MS = 300_000 是硬编码， 而真实单事件上限 = (max_retries+1) x (jmap+telegram+llm 三个 timeout)， 运维把任一超时调大后实际耗时可能远超 300s → 空闲阈值欠保护， 多实例会把还在处理的条目提前认领，造成重复投递。 改动（ROADMAP 缺口③「多实例重复窗口」）： - single_event_ceiling_ms(config) 纯函数：(max_retries+1) x (jmap+telegram+llm) x 2 (SAFETY)，并用 .max(SINGLE_EVENT_CEILING_FLOOR_MS=300_000) 兜底， 零值配置不会把窗口缩到 legacy 常量之下。 - reclaim_idle_threshold_ms(config, count) 纯函数：count.max(1) x 单事件上限， 再 .min(RECLAIM_IDLE_THRESHOLD_CAP_MS = 6h) 防误调把恢复拖到天级。 - read_batch 调用 self.get_outbound_config()（trait 方法，同实现内已有，未改签名）， Ok 走推导值，Err 回退 count x 300_000（不破坏投递主流程、不传播 Redis 错误）。 既有闸门不动：notify.rs 投递前查 delivery:committed、60s delivery:inflight 在途租约、 retry_or_dlq 3 次。本次只修阈值本身，性质保持「永不丢、只可能重复；单实例不受影响」。 :356-364 那段注释改写为描述真实机制（旧文「~300s 上限」已不准确）， 未写入任何未经测试证明的断言。 测试：3 个纯函数单测覆盖 —— 默认配置 (15+10+30)x4x2=440s >= 300s floor、 大超时配置阈值更大、零值配置不缩到 floor 之下、荒唐超时被封顶 6h、count=0 用 1 事件兜底。 GATE-P0 全绿：fmt --check、check --locked、clippy --locked --all-targets -D warnings、 test --locked → 55 passed / 0 failed / 1 ignored。 --- src/state.rs | 105 +++++++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 101 insertions(+), 4 deletions(-) diff --git a/src/state.rs b/src/state.rs index 1bd5ca8..47ad6f3 100644 --- a/src/state.rs +++ b/src/state.rs @@ -60,6 +60,47 @@ impl Default for OutboundConfig { } } +/// Legacy single-event ceiling. It is the floor the derived ceiling must clear +/// and the fallback when the outbound config cannot be read. +const SINGLE_EVENT_CEILING_FLOOR_MS: u64 = 300_000; + +/// Safety factor on the derived ceiling. Retries insert sleeps between +/// attempts, and the ceiling bounds a window that must not be exceeded while +/// the owning consumer is still alive. +const SINGLE_EVENT_CEILING_SAFETY: u64 = 2; + +/// Upper bound on the XAUTOCLAIM idle window: a misconfigured timeout must not +/// push multi-instance recovery out by days. Six hours keeps a crashed batch +/// recoverable within the same working day. +const RECLAIM_IDLE_THRESHOLD_CAP_MS: u64 = 6 * 60 * 60 * 1000; + +/// Worst-case wall clock for one event: every outbound call can burn its full +/// timeout once per attempt and `send_text` retries up to `max_retries` more +/// times, so the ceiling is +/// `(max_retries + 1) * (jmap + telegram + llm) * SAFETY`, floored at the +/// legacy constant so a zeroed-out config cannot shrink the window. +fn single_event_ceiling_ms(config: &OutboundConfig) -> u64 { + let attempts = config.max_retries as u64 + 1; + let per_attempt = config + .jmap_timeout_ms + .saturating_add(config.telegram_timeout_ms) + .saturating_add(config.llm_timeout_ms); + per_attempt + .saturating_mul(attempts) + .saturating_mul(SINGLE_EVENT_CEILING_SAFETY) + .max(SINGLE_EVENT_CEILING_FLOOR_MS) +} + +/// Idle window handed to `XAUTOCLAIM`: the worst-case *batch* duration, since a +/// worker processes up to `count` events sequentially. Floored through +/// `single_event_ceiling_ms` and capped so a bad config cannot stretch +/// recovery into days. Only duplicate-on-multi-instance is at stake, never loss. +fn reclaim_idle_threshold_ms(config: &OutboundConfig, count: usize) -> u64 { + (count.max(1) as u64) + .saturating_mul(single_event_ceiling_ms(config)) + .min(RECLAIM_IDLE_THRESHOLD_CAP_MS) +} + #[async_trait] pub trait ReliableState: Send + Sync { /// Atomic SET NX EX claim for Telegram update or `(account,email)` dedup key. @@ -360,13 +401,18 @@ impl ReliableState for RedisState { .await; // Recover messages left in the consumer group's PEL after a crash or // restart before reading new entries. The idle window has to clear the - // worst-case *batch* duration (a worker processes up to `count` events - // sequentially, each bounded by the ~300s single-event ceiling), else a + // worst-case *batch* duration: a worker processes up to `count` events + // sequentially, each bounded by the single-event ceiling derived from + // the live outbound config. If the window is shorter than that, a // second instance reclaims events that are still being processed and // delivers them twice (建议-6). Single-instance deployments are // unaffected; only duplicate-on-multi-instance is at stake, never loss. - const SINGLE_EVENT_CEILING_MS: u64 = 300_000; - let idle_threshold_ms = count.max(1) as u64 * SINGLE_EVENT_CEILING_MS; + let idle_threshold_ms = match self.get_outbound_config().await { + Ok(config) => reclaim_idle_threshold_ms(&config, count), + // A Redis error here must not break the delivery loop: fall back to + // the legacy constant instead of propagating the failure. + Err(_) => count.max(1) as u64 * SINGLE_EVENT_CEILING_FLOOR_MS, + }; let reclaimed: redis::streams::StreamAutoClaimReply = redis::cmd("XAUTOCLAIM") .arg(stream) .arg(group) @@ -1346,6 +1392,57 @@ mod tests { assert!(client.is_ok()); } + #[test] + fn reclaim_idle_threshold_is_derived_from_runtime_config() { + // (15s + 10s + 30s) * (3 + 1) attempts * 2 safety = 440s per event. + let default = OutboundConfig::default(); + assert_eq!(single_event_ceiling_ms(&default), 440_000); + assert!(reclaim_idle_threshold_ms(&default, 1) >= SINGLE_EVENT_CEILING_FLOOR_MS); + assert!(reclaim_idle_threshold_ms(&default, 16) > reclaim_idle_threshold_ms(&default, 8)); + + let slow = OutboundConfig { + jmap_timeout_ms: 60_000, + telegram_timeout_ms: 60_000, + llm_timeout_ms: 60_000, + max_retries: 4, + }; + assert!(reclaim_idle_threshold_ms(&slow, 1) > reclaim_idle_threshold_ms(&default, 1)); + } + + #[test] + fn reclaim_idle_threshold_is_floored_and_capped() { + let zeroed = OutboundConfig { + jmap_timeout_ms: 0, + telegram_timeout_ms: 0, + llm_timeout_ms: 0, + max_retries: 0, + }; + assert_eq!( + single_event_ceiling_ms(&zeroed), + SINGLE_EVENT_CEILING_FLOOR_MS + ); + + let absurd = OutboundConfig { + jmap_timeout_ms: 3_600_000, + telegram_timeout_ms: 3_600_000, + llm_timeout_ms: 3_600_000, + max_retries: 5, + }; + assert_eq!( + reclaim_idle_threshold_ms(&absurd, 64), + RECLAIM_IDLE_THRESHOLD_CAP_MS + ); + } + + #[test] + fn reclaim_idle_threshold_ignores_empty_batches() { + let default = OutboundConfig::default(); + assert_eq!( + reclaim_idle_threshold_ms(&default, 0), + single_event_ceiling_ms(&default) + ); + } + #[test] fn encrypted_config_roundtrip_and_tamper_fail_closed() { let key = [7_u8; 32];