state: XAUTOCLAIM 空闲阈值改由运行超时配置推导,封顶 6h(01a0e75f-f9b1)

现状: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。
This commit is contained in:
anonsaber 2026-09-28 18:38:00 +08:00
parent f4cae00adc
commit 6c99ce51ab
1 changed files with 101 additions and 4 deletions

View File

@ -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];