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:
parent
f4cae00adc
commit
6c99ce51ab
105
src/state.rs
105
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];
|
||||
|
|
|
|||
Loading…
Reference in New Issue