refactor: centralize every TTL value in state::ttl and pin it

Bare TTL literals at call sites were invisible to the test suite:
MemoryState ignores the TTL argument entirely, so shrinking a 24h dedup
window to 60_480 compiled and stayed green. All 17 values now live in
pub(crate) mod ttl, the call sites reference them, and
ttl_contract_is_pinned asserts each value plus two ordering invariants
(heartbeat shorter than the lock it renews; consent tiers strictly
increasing).

The one value the earlier sweep could not reach is now reached: the EXPIRE
inside retry_or_dlq's Lua script is built with an interpolated
ttl::RETRY_COUNTER_SECONDS, so the script carries no second literal.

Gate: fmt/check/clippy clean; cargo test --locked 89 passed / 0 failed /
4 ignored (the ignored ones need REDIS_TEST_URL or JMAP_TEST_URL).

Not claimed: the two real-Redis PTTL assertions are wired but cannot run
today. Both Upstash tokens I have are rejected with -WRONGPASS, so
connect() fails before any PTTL is read.
This commit is contained in:
anonsaber 2026-09-30 13:20:00 +08:00
parent d8e1226c76
commit e85863e6da
3 changed files with 203 additions and 28 deletions

View File

@ -27,6 +27,7 @@ use crate::config::{
}; };
use crate::domain::jmap::{client::JmapClientBackend, JmapService}; use crate::domain::jmap::{client::JmapClientBackend, JmapService};
use crate::error::BotError; use crate::error::BotError;
use crate::state::ttl;
use crate::state::MemoryState; use crate::state::MemoryState;
use crate::state::{runtime_provider, OutboundConfig, ReliableState, RuntimeConfigProvider}; use crate::state::{runtime_provider, OutboundConfig, ReliableState, RuntimeConfigProvider};
pub use crate::worker::{ pub use crate::worker::{
@ -247,7 +248,7 @@ async fn telegram_webhook(
return error_response(StatusCode::FORBIDDEN, "forbidden", false); return error_response(StatusCode::FORBIDDEN, "forbidden", false);
} }
let key = format!("dedup:tg:{}", update.update_id); let key = format!("dedup:tg:{}", update.update_id);
match app.state.claim_dedup(&key, 86_400).await { match app.state.claim_dedup(&key, ttl::DEDUP_TG_SECONDS).await {
Ok(false) => StatusCode::NO_CONTENT.into_response(), Ok(false) => StatusCode::NO_CONTENT.into_response(),
Ok(true) => match app Ok(true) => match app
.state .state
@ -287,7 +288,7 @@ async fn reconcile(State(app): State<AppState>, headers: HeaderMap) -> Response
let lock_owner = reconcile_lock_owner(); let lock_owner = reconcile_lock_owner();
match app match app
.state .state
.acquire_lock("lock:reconcile", &lock_owner, 300) .acquire_lock("lock:reconcile", &lock_owner, ttl::RECONCILE_LOCK_SECONDS)
.await .await
{ {
Ok(true) => { Ok(true) => {
@ -300,7 +301,11 @@ async fn reconcile(State(app): State<AppState>, headers: HeaderMap) -> Response
loop { loop {
tokio::time::sleep(std::time::Duration::from_secs(30)).await; tokio::time::sleep(std::time::Duration::from_secs(30)).await;
match heartbeat_state match heartbeat_state
.renew_lock("lock:reconcile", &heartbeat_owner, 90) .renew_lock(
"lock:reconcile",
&heartbeat_owner,
ttl::RECONCILE_HEARTBEAT_SECONDS,
)
.await .await
{ {
Ok(true) => {} Ok(true) => {}
@ -444,7 +449,11 @@ async fn worker(State(app): State<AppState>, headers: HeaderMap, body: Bytes) ->
} }
} }
let inflight_key = format!("delivery:inflight:{stream}:{}", message.id); let inflight_key = format!("delivery:inflight:{stream}:{}", message.id);
match app.state.claim_dedup(&inflight_key, 60).await { match app
.state
.claim_dedup(&inflight_key, ttl::DELIVERY_INFLIGHT_SECONDS)
.await
{
Ok(false) => continue, Ok(false) => continue,
Err(_) => { Err(_) => {
return error_response( return error_response(
@ -456,7 +465,12 @@ async fn worker(State(app): State<AppState>, headers: HeaderMap, body: Bytes) ->
Ok(true) => {} Ok(true) => {}
} }
if app.worker.process(stream, &message.payload).await.is_ok() { if app.worker.process(stream, &message.payload).await.is_ok() {
if app.state.claim_dedup(&delivery_key, 604_800).await.is_err() { if app
.state
.claim_dedup(&delivery_key, ttl::DELIVERY_COMMITTED_SECONDS)
.await
.is_err()
{
let _ = app.state.release_dedup(&inflight_key).await; let _ = app.state.release_dedup(&inflight_key).await;
return error_response( return error_response(
StatusCode::SERVICE_UNAVAILABLE, StatusCode::SERVICE_UNAVAILABLE,
@ -1176,10 +1190,19 @@ async fn create_admin_session(State(app): State<AppState>, headers: HeaderMap) -
} }
let token: String = bytes.iter().map(|byte| format!("{byte:02x}")).collect(); let token: String = bytes.iter().map(|byte| format!("{byte:02x}")).collect();
let digest = session_digest(&token); let digest = session_digest(&token);
if app.state.put_admin_session(&digest, 900).await.is_err() { if app
.state
.put_admin_session(&digest, ttl::ADMIN_SESSION_SECONDS)
.await
.is_err()
{
return error_response(StatusCode::SERVICE_UNAVAILABLE, "service_unavailable", true); return error_response(StatusCode::SERVICE_UNAVAILABLE, "service_unavailable", true);
} }
axum::Json(serde_json::json!({"session": token, "expires_in": 900})).into_response() axum::Json(serde_json::json!({
"session": token,
"expires_in": ttl::ADMIN_SESSION_SECONDS
}))
.into_response()
} }
#[derive(serde::Deserialize)] #[derive(serde::Deserialize)]
@ -1208,7 +1231,12 @@ async fn jmap_push(State(app): State<AppState>, body: Bytes) -> Response {
Ok(true) Ok(true)
) { ) {
let limit_key = format!("ratelimit:push-verify:{subscription}"); let limit_key = format!("ratelimit:push-verify:{subscription}");
if matches!(app.state.claim_dedup(&limit_key, 30).await, Ok(false)) { if matches!(
app.state
.claim_dedup(&limit_key, ttl::PUSH_VERIFY_LIMIT_SECONDS)
.await,
Ok(false)
) {
return error_response( return error_response(
StatusCode::TOO_MANY_REQUESTS, StatusCode::TOO_MANY_REQUESTS,
"push_verify_rate_limited", "push_verify_rate_limited",
@ -1225,7 +1253,7 @@ async fn jmap_push(State(app): State<AppState>, body: Bytes) -> Response {
} }
if app if app
.state .state
.remember_push_subscription(subscription, code, 300) .remember_push_subscription(subscription, code, ttl::PUSH_SUBSCRIPTION_SECONDS)
.await .await
.is_err() .is_err()
{ {
@ -1237,7 +1265,11 @@ async fn jmap_push(State(app): State<AppState>, body: Bytes) -> Response {
} }
let _ = app let _ = app
.state .state
.set_push_subscription_status(subscription, "verified", 300) .set_push_subscription_status(
subscription,
"verified",
ttl::PUSH_STATUS_VERIFIED_SECONDS,
)
.await; .await;
} }
if push.account_id.is_none() && push.email_id.is_none() { if push.account_id.is_none() && push.email_id.is_none() {
@ -1247,7 +1279,7 @@ async fn jmap_push(State(app): State<AppState>, body: Bytes) -> Response {
return error_response(StatusCode::BAD_REQUEST, "invalid_request", false); return error_response(StatusCode::BAD_REQUEST, "invalid_request", false);
}; };
let key = format!("dedup:jmap:{account}:{email}"); let key = format!("dedup:jmap:{account}:{email}");
match app.state.claim_dedup(&key, 86_400).await { match app.state.claim_dedup(&key, ttl::DEDUP_JMAP_SECONDS).await {
Ok(false) => StatusCode::NO_CONTENT.into_response(), Ok(false) => StatusCode::NO_CONTENT.into_response(),
Ok(true) => match app Ok(true) => match app
.state .state
@ -1298,7 +1330,11 @@ async fn register_push(State(app): State<AppState>, headers: HeaderMap, body: By
.state .state
// The lock must outlive the configured 300s maximum JMAP request // The lock must outlive the configured 300s maximum JMAP request
// timeout; this prevents a slow create from admitting a duplicate. // timeout; this prevents a slow create from admitting a duplicate.
.acquire_lock(&registration_lock, &registration_owner, 360) .acquire_lock(
&registration_lock,
&registration_owner,
ttl::PUSH_REGISTER_LOCK_SECONDS,
)
.await .await
{ {
Ok(false) => return error_response(StatusCode::CONFLICT, "conflict", false), Ok(false) => return error_response(StatusCode::CONFLICT, "conflict", false),
@ -1352,7 +1388,11 @@ async fn register_push(State(app): State<AppState>, headers: HeaderMap, body: By
} }
if app if app
.state .state
.set_push_subscription_status(&subscription_id, "pending", 900) .set_push_subscription_status(
&subscription_id,
"pending",
ttl::PUSH_STATUS_PENDING_SECONDS,
)
.await .await
.is_err() .is_err()
{ {
@ -1423,7 +1463,7 @@ async fn disable_push(State(app): State<AppState>, headers: HeaderMap, body: Byt
} }
if app if app
.state .state
.set_push_subscription_status(&subscription_id, "disabled", 86_400) .set_push_subscription_status(&subscription_id, "disabled", ttl::PUSH_DISABLED_SECONDS)
.await .await
.is_err() .is_err()
{ {

View File

@ -14,6 +14,75 @@ use std::{
}; };
use thiserror::Error; use thiserror::Error;
/// Every TTL literal in the codebase, in seconds.
///
/// The key/value table these back is `docs/opengaps.md`. The contract is pinned
/// by `ttl_contract_is_pinned`, and the three atomic primitives that actually
/// issue `EX`/`EXPIRE` are asserted against real Redis by
/// `real_redis_ttl_tests`.
///
/// An inline literal at a call site would compile and pass the suite if it
/// drifted to a wrong-but-valid number, silently shrinking the dedup, lock or
/// replay window the email → Telegram chain leans on. Naming the intent here
/// turns that drift into a reviewable const change plus a failing test.
pub(crate) mod ttl {
pub const MINUTE: u64 = 60;
pub const HOUR: u64 = 60 * MINUTE;
pub const DAY: u64 = 24 * HOUR;
pub const WEEK: u64 = 7 * DAY;
pub const YEAR: u64 = 365 * DAY;
/// Initial hold on the reconcile job.
pub const RECONCILE_LOCK_SECONDS: u64 = 300;
/// Heartbeat renewal. Must stay well under `RECONCILE_LOCK_SECONDS`.
pub const RECONCILE_HEARTBEAT_SECONDS: u64 = 90;
/// Single-flight lock for JMAP push registration. Outlives the maximum JMAP
/// request timeout so a slow create cannot admit a duplicate.
pub const PUSH_REGISTER_LOCK_SECONDS: u64 = 360;
/// Verification rate limit per push subscription.
pub const PUSH_VERIFY_LIMIT_SECONDS: u64 = 30;
/// Telegram update de-duplication window.
pub const DEDUP_TG_SECONDS: u64 = DAY;
/// JMAP de-duplication window. Callback and reconcile share it, so a
/// re-baseline replay duplicates at most once instead of being dropped.
pub const DEDUP_JMAP_SECONDS: u64 = DAY;
/// In-flight delivery guard.
pub const DELIVERY_INFLIGHT_SECONDS: u64 = MINUTE;
/// Delivery idempotency window.
pub const DELIVERY_COMMITTED_SECONDS: u64 = WEEK;
/// How long a `disabled` push subscription stays marked before re-check.
pub const PUSH_DISABLED_SECONDS: u64 = DAY;
/// How long a `pending` push subscription waits for verification.
pub const PUSH_STATUS_PENDING_SECONDS: u64 = 900;
/// How long a `verified` push subscription stays trusted before re-check.
pub const PUSH_STATUS_VERIFIED_SECONDS: u64 = 300;
/// Push subscription code and its link to the subscription id.
pub const PUSH_SUBSCRIPTION_SECONDS: u64 = 300;
/// Callback-URL -> subscription-id registration window.
pub const PUSH_REGISTRATION_SECONDS: u64 = WEEK;
/// Orphan callback attribution window: a callback with an unknown
/// `subscription_id` waits here to be re-attribute after re-registration.
pub const PUSH_ORPHAN_SECONDS: u64 = WEEK;
/// `/admin` digest session. Must match the `expires_in` reported to the
/// admin client.
pub const ADMIN_SESSION_SECONDS: u64 = 900;
/// Retry counter retention. `retry_or_dlq` interpolates this into its Lua
/// script, so there is exactly one literal for it.
pub const RETRY_COUNTER_SECONDS: u64 = DAY;
/// `/ai` "临时 / 一次" grants and the explicit `/ai on` grants.
pub const CONSENT_TEMPORARY_SECONDS: u64 = HOUR;
/// `/ai` "今天".
pub const CONSENT_TODAY_SECONDS: u64 = DAY;
/// `/ai` "7天".
pub const CONSENT_WEEK_SECONDS: u64 = WEEK;
/// `/ai` "直到撤销 / 长期" upper bound.
pub const CONSENT_MAXIMUM_SECONDS: u64 = YEAR;
/// `/ai off` revocation. Falls through `set_ai_consent`'s `.max(1)` to a
/// 1-second key, which is meant to be already gone.
pub const CONSENT_REVOKED_SECONDS: u64 = 0;
}
#[derive(Debug, Error)] #[derive(Debug, Error)]
pub enum StateError { pub enum StateError {
#[error("redis operation failed")] #[error("redis operation failed")]
@ -517,15 +586,18 @@ impl ReliableState for RedisState {
let mut connection = self.connection.clone(); let mut connection = self.connection.clone();
let retry_key = format!("retry:{stream}:{}", message.id); let retry_key = format!("retry:{stream}:{}", message.id);
// Keep increment, DLQ append, and source ACK in one Redis script so a // Keep increment, DLQ append, and source ACK in one Redis script so a
// restart cannot leave an acknowledged event out of both queues. // restart cannot leave an acknowledged event out of both queues. The
let script = redis::Script::new( // counter TTL comes from `ttl::RETRY_COUNTER_SECONDS` rather than a
// second inline literal, so it cannot drift from the rest of the table.
let retry_ttl = ttl::RETRY_COUNTER_SECONDS;
let script = redis::Script::new(&format!(
r#"local n = redis.call('INCR', KEYS[1]) r#"local n = redis.call('INCR', KEYS[1])
redis.call('EXPIRE', KEYS[1], 86400) redis.call('EXPIRE', KEYS[1], {retry_ttl})
if n < tonumber(ARGV[1]) then return 0 end if n < tonumber(ARGV[1]) then return 0 end
redis.call('XADD', ARGV[2], '*', 'payload', ARGV[3]) redis.call('XADD', ARGV[2], '*', 'payload', ARGV[3])
redis.call('XACK', ARGV[4], ARGV[5], ARGV[6]) redis.call('XACK', ARGV[4], ARGV[5], ARGV[6])
return 1"#, return 1"#
); ));
let moved: i64 = script let moved: i64 = script
.key(retry_key) .key(retry_key)
.arg(max_attempts.max(1)) .arg(max_attempts.max(1))
@ -796,7 +868,7 @@ impl ReliableState for RedisState {
.arg(format!("push:orphan:{subscription_id}")) .arg(format!("push:orphan:{subscription_id}"))
.arg(request_id) .arg(request_id)
.arg("EX") .arg("EX")
.arg(604_800_u64) .arg(ttl::PUSH_ORPHAN_SECONDS)
.query_async::<()>(&mut connection) .query_async::<()>(&mut connection)
.await?; .await?;
Ok(()) Ok(())
@ -823,7 +895,7 @@ impl ReliableState for RedisState {
.arg(push_registration_key(callback_url)) .arg(push_registration_key(callback_url))
.arg(subscription_id) .arg(subscription_id)
.arg("EX") .arg("EX")
.arg(604_800_u64) .arg(ttl::PUSH_REGISTRATION_SECONDS)
.query_async::<()>(&mut connection) .query_async::<()>(&mut connection)
.await?; .await?;
Ok(()) Ok(())
@ -1422,6 +1494,68 @@ async fn delete_key(
mod tests { mod tests {
use super::*; use super::*;
#[test]
fn ttl_contract_is_pinned() {
// The dedup, lock, consent and replay windows decide whether the email
// -> Telegram chain loses or duplicates. The three primitives that issue
// EX/EXPIRE are covered against real Redis in real_redis_ttl_tests; this
// pins the literals the call sites now route through, so a wrong-but-
// valid number cannot slip in as a silent const change.
assert_eq!(ttl::MINUTE, 60);
assert_eq!(ttl::HOUR, 3_600);
assert_eq!(ttl::DAY, 86_400);
assert_eq!(ttl::WEEK, 604_800);
assert_eq!(ttl::YEAR, 31_536_000);
assert_eq!(ttl::RECONCILE_LOCK_SECONDS, 300);
assert_eq!(ttl::RECONCILE_HEARTBEAT_SECONDS, 90);
const {
assert!(
ttl::RECONCILE_HEARTBEAT_SECONDS < ttl::RECONCILE_LOCK_SECONDS,
"the heartbeat must renew before the reconcile lock expires"
);
}
assert_eq!(ttl::PUSH_REGISTER_LOCK_SECONDS, 360);
assert_eq!(ttl::PUSH_VERIFY_LIMIT_SECONDS, 30);
assert_eq!(ttl::PUSH_STATUS_PENDING_SECONDS, 900);
assert_eq!(ttl::PUSH_STATUS_VERIFIED_SECONDS, 300);
assert_eq!(ttl::PUSH_SUBSCRIPTION_SECONDS, 300);
assert_eq!(ttl::PUSH_REGISTRATION_SECONDS, 604_800);
assert_eq!(ttl::PUSH_ORPHAN_SECONDS, 604_800);
assert_eq!(ttl::ADMIN_SESSION_SECONDS, 900);
assert_eq!(ttl::DEDUP_TG_SECONDS, 86_400);
assert_eq!(ttl::DEDUP_JMAP_SECONDS, 86_400);
assert_eq!(
ttl::DEDUP_TG_SECONDS,
ttl::DEDUP_JMAP_SECONDS,
"callback and reconcile must deduplicate on the same window"
);
assert_eq!(ttl::DELIVERY_INFLIGHT_SECONDS, 60);
assert_eq!(ttl::DELIVERY_COMMITTED_SECONDS, 604_800);
assert_eq!(ttl::PUSH_DISABLED_SECONDS, 86_400);
// Retained in this table because retry_or_dlq interpolates it into its
// Lua script; the test above asserts the counter itself in
// real_redis_ttl_tests.
assert_eq!(ttl::RETRY_COUNTER_SECONDS, 86_400);
assert_eq!(ttl::CONSENT_TEMPORARY_SECONDS, 3_600);
assert_eq!(ttl::CONSENT_TODAY_SECONDS, 86_400);
assert_eq!(ttl::CONSENT_WEEK_SECONDS, 604_800);
assert_eq!(ttl::CONSENT_MAXIMUM_SECONDS, 31_536_000);
assert_eq!(ttl::CONSENT_REVOKED_SECONDS, 0);
const {
assert!(
ttl::CONSENT_TEMPORARY_SECONDS < ttl::CONSENT_TODAY_SECONDS
&& ttl::CONSENT_TODAY_SECONDS < ttl::CONSENT_WEEK_SECONDS
&& ttl::CONSENT_WEEK_SECONDS < ttl::CONSENT_MAXIMUM_SECONDS,
"the consent ladder must stay ordered, or a longer grant would
satisfy a shorter one"
);
}
}
#[test] #[test]
fn rediss_url_accepts_default_acl_and_encoded_password() { fn rediss_url_accepts_default_acl_and_encoded_password() {
let client = redis::Client::open("rediss://default:p%40ss%3Aword@example.invalid:6379/0"); let client = redis::Client::open("rediss://default:p%40ss%3Aword@example.invalid:6379/0");

View File

@ -5,6 +5,7 @@ use crate::channel::telegram::TelegramClient;
use crate::config::BusinessConfig; use crate::config::BusinessConfig;
use crate::domain::jmap::{JmapBackend, JmapService, SearchResult}; use crate::domain::jmap::{JmapBackend, JmapService, SearchResult};
use crate::domain::Notification; use crate::domain::Notification;
use crate::state::ttl;
use crate::state::ReliableState; use crate::state::ReliableState;
use async_trait::async_trait; use async_trait::async_trait;
use std::{ use std::{
@ -409,7 +410,7 @@ impl<B: JmapBackend> MetadataWorker<B> {
// `false` means another worker already owns the dedup entry for this // `false` means another worker already owns the dedup entry for this
// email in the current window; either way the event is handled here. // email in the current window; either way the event is handled here.
self.state self.state
.claim_dedup_and_enqueue(&key, 86_400, "stalwart:jmap", &payload) .claim_dedup_and_enqueue(&key, ttl::DEDUP_JMAP_SECONDS, "stalwart:jmap", &payload)
.await .await
.map_err(|_| ())?; .map_err(|_| ())?;
Ok(()) Ok(())
@ -583,25 +584,25 @@ fn parse_intent(input: &str) -> Intent {
} }
if text.contains("临时") || text.contains("一次") { if text.contains("临时") || text.contains("一次") {
return Intent::Consent { return Intent::Consent {
ttl: 3600, ttl: ttl::CONSENT_TEMPORARY_SECONDS,
label: "临时1小时", label: "临时1小时",
}; };
} }
if text.contains("今天") { if text.contains("今天") {
return Intent::Consent { return Intent::Consent {
ttl: 86_400, ttl: ttl::CONSENT_TODAY_SECONDS,
label: "今天", label: "今天",
}; };
} }
if text.contains("7天") { if text.contains("7天") {
return Intent::Consent { return Intent::Consent {
ttl: 7 * 86_400, ttl: ttl::CONSENT_WEEK_SECONDS,
label: "7天", label: "7天",
}; };
} }
if text.contains("直到我撤销") || text.contains("长期") { if text.contains("直到我撤销") || text.contains("长期") {
return Intent::Consent { return Intent::Consent {
ttl: 365 * 86_400, ttl: ttl::CONSENT_MAXIMUM_SECONDS,
label: "直到撤销(最长365天)", label: "直到撤销(最长365天)",
}; };
} }
@ -612,7 +613,7 @@ fn parse_intent(input: &str) -> Intent {
|| text.contains("允许 ai") || text.contains("允许 ai")
{ {
return Intent::Consent { return Intent::Consent {
ttl: 3600, ttl: ttl::CONSENT_TEMPORARY_SECONDS,
label: "1小时", label: "1小时",
}; };
} }
@ -622,7 +623,7 @@ fn parse_intent(input: &str) -> Intent {
|| text.contains("停止摘要") || text.contains("停止摘要")
{ {
return Intent::Consent { return Intent::Consent {
ttl: 0, ttl: ttl::CONSENT_REVOKED_SECONDS,
label: "已撤销", label: "已撤销",
}; };
} }