From 309a38c503e857090adbac44986247f39f9efa78 Mon Sep 17 00:00:00 2001 From: anonsaber Date: Sun, 27 Sep 2026 12:53:05 +0800 Subject: [PATCH] fix(worker,state,jmap,config): close independent-review findings - worker/state: paginate Email/changes by widening maxChanges on the same sinceState (jmap-client 0.4.2 has no upToId) to avoid skipping a batch when advancing per page; make dedup+enqueue atomic via claim_dedup_and_enqueue (Lua SET NX EX then XADD); scale the XAUTOCLAIM idle threshold by batch size; XACK payload-less malformed stream entries out of the PEL via ack_malformed; fail closed on unknown streams (Err); tighten SET NX EX TTL to .max(1); derive the CSPRNG fallback owner-token from time+PID+counter instead of a constant; clear the push verification digest on push:disable - worker: /help menu now lists only the consent triggers the parser accepts - jmap/config: complete client and configuration validation --- src/config.rs | 68 +++- src/domain/jmap.rs | 134 ++++++++ src/domain/jmap/client.rs | 192 ++++++++++- src/state.rs | 671 +++++++++++++++++++++++++++++++++++--- src/worker.rs | 227 ++++++++++++- 5 files changed, 1226 insertions(+), 66 deletions(-) diff --git a/src/config.rs b/src/config.rs index a482a08..7ec14fb 100644 --- a/src/config.rs +++ b/src/config.rs @@ -49,7 +49,6 @@ pub struct Config { pub struct AuthSecrets { pub reconcile_token: SecretString, pub telegram_webhook_secret: SecretString, - pub jmap_push_verification: SecretString, } pub struct TelegramConfig { @@ -85,7 +84,6 @@ pub struct BusinessConfig { pub jmap_session_url: String, pub jmap_username: String, pub jmap_password: SecretString, - pub jmap_push_verification: SecretString, pub account_id: Option, pub llm_enabled: bool, pub llm_allow_net: bool, @@ -107,7 +105,6 @@ pub(crate) struct BusinessConfigWire { pub jmap_session_url: String, pub jmap_username: String, pub jmap_password: String, - pub jmap_push_verification: String, pub account_id: Option, pub llm_enabled: bool, pub llm_allow_net: bool, @@ -128,7 +125,6 @@ impl From for BusinessConfigWire { jmap_session_url: value.jmap_session_url, jmap_username: value.jmap_username, jmap_password: value.jmap_password.expose_secret().to_owned(), - jmap_push_verification: value.jmap_push_verification.expose_secret().to_owned(), account_id: value.account_id, llm_enabled: value.llm_enabled, llm_allow_net: value.llm_allow_net, @@ -150,7 +146,6 @@ impl TryFrom for BusinessConfig { validate_nonblank("JMAP_USERNAME", &value.jmap_username)?; validate_nonblank("JMAP_PASSWORD", &value.jmap_password)?; validate_nonblank("TG_WEBHOOK_SECRET", &value.telegram_webhook_secret)?; - validate_nonblank("JMAP_PUSH_VERIFICATION", &value.jmap_push_verification)?; validate_nonblank("RECONCILE_TOKEN", &value.reconcile_token)?; validate_nonblank("WORKER_TOKEN", &value.worker_token)?; if let Some(key) = value.llm_api_key.as_deref() { @@ -164,7 +159,6 @@ impl TryFrom for BusinessConfig { jmap_session_url: value.jmap_session_url, jmap_username: value.jmap_username, jmap_password: SecretString::new(value.jmap_password), - jmap_push_verification: SecretString::new(value.jmap_push_verification), account_id: value.account_id, llm_enabled: value.llm_enabled, llm_allow_net: value.llm_allow_net, @@ -179,6 +173,12 @@ impl TryFrom for BusinessConfig { pub(crate) fn validate_business_wire(wire: BusinessConfigWire) -> Result<(), BotError> { let config: BusinessConfig = wire.try_into()?; + if config.chat_allowlist.is_empty() { + // SAF-CHAT-ALLOWLIST: 必填，处理前先拒绝非白名单；空名单是非法配置而非 fail-open。 + return Err(BotError::Config( + "CHAT_ALLOWLIST must contain at least one chat id".into(), + )); + } let session = url::Url::parse(&config.jmap_session_url) .map_err(|_| BotError::Config("JMAP_SESSION_URL must be a valid URL".into()))?; if session.scheme() != "https" @@ -242,7 +242,6 @@ impl Config { auth: AuthSecrets { reconcile_token: empty(), telegram_webhook_secret: empty(), - jmap_push_verification: empty(), }, worker_token: empty(), } @@ -277,7 +276,6 @@ impl Config { auth: AuthSecrets { reconcile_token: value.reconcile_token, telegram_webhook_secret: value.telegram_webhook_secret, - jmap_push_verification: value.jmap_push_verification, }, worker_token: value.worker_token, } @@ -299,6 +297,10 @@ impl Config { Ok(Self::from_business(redis_url, wire.try_into()?)) } + /// 遗留兼容路径：从进程环境变量逐项组装完整配置（阶段0 口径）。 + /// 当前生产启动仅需 `REDIS_URL` + `CONFIG_ENCRYPTION_KEY`（见 `src/main.rs`）， + /// 业务字段改由 Redis 业务配置（`config:business`，`PUT /api/business-config` 热加载）托管； + /// 本函数仅当相关环境变量实际存在时作为引导路径使用，未声明废弃、未计划删除。 pub fn from_env() -> Result { let port = std::env::var("PORT") .unwrap_or_else(|_| "8080".into()) @@ -353,7 +355,6 @@ impl Config { auth: AuthSecrets { reconcile_token: required_secret("RECONCILE_TOKEN")?, telegram_webhook_secret: required_secret("TG_WEBHOOK_SECRET")?, - jmap_push_verification: required_secret("JMAP_PUSH_VERIFICATION")?, }, worker_token: required_secret("WORKER_TOKEN")?, }) @@ -411,6 +412,53 @@ mod tests { std::env::remove_var("S1_TEST_AUTH_SECRET"); } + #[test] + fn empty_chat_allowlist_is_rejected() { + // SAF-CHAT-ALLOWLIST: 空名单是非法配置（Web 侧同样强制至少一项）， + // 不得让绕过 SPA 的直接 PUT 静默写入"拒绝所有 chat"的配置。 + let wire = BusinessConfigWire { + bot_token: "bot".into(), + telegram_chat_id: 1, + chat_allowlist: Vec::new(), + telegram_webhook_secret: "hook".into(), + jmap_session_url: "https://example.invalid".into(), + jmap_username: "u".into(), + jmap_password: "p".into(), + account_id: None, + llm_enabled: false, + llm_allow_net: false, + llm_api_key: None, + llm_base_url: None, + llm_model: None, + reconcile_token: "r".into(), + worker_token: "w".into(), + }; + let error = validate_business_wire(wire).expect_err("empty allowlist must fail"); + assert!(error.to_string().contains("CHAT_ALLOWLIST")); + } + + #[test] + fn nonempty_chat_allowlist_passes() { + let wire = BusinessConfigWire { + bot_token: "bot".into(), + telegram_chat_id: 1, + chat_allowlist: vec![1, 2], + telegram_webhook_secret: "hook".into(), + jmap_session_url: "https://example.invalid".into(), + jmap_username: "u".into(), + jmap_password: "p".into(), + account_id: None, + llm_enabled: false, + llm_allow_net: false, + llm_api_key: None, + llm_base_url: None, + llm_model: None, + reconcile_token: "r".into(), + worker_token: "w".into(), + }; + validate_business_wire(wire).expect("non-empty allowlist must pass"); + } + #[test] fn required_config_values_reject_blank_without_exposing_values() { for name in [ @@ -437,7 +485,6 @@ mod tests { ("JMAP_PASSWORD", "password"), ("REDIS_URL", "redis://invalid"), ("TG_WEBHOOK_SECRET", "telegram"), - ("JMAP_PUSH_VERIFICATION", "verification"), ("TELEGRAM_CHAT_ID", "123"), ] { std::env::set_var(name, value); @@ -459,7 +506,6 @@ mod tests { "REDIS_URL", "RECONCILE_TOKEN", "TG_WEBHOOK_SECRET", - "JMAP_PUSH_VERIFICATION", "TELEGRAM_CHAT_ID", ] { std::env::remove_var(name); diff --git a/src/domain/jmap.rs b/src/domain/jmap.rs index 30cd6bb..1b8bb96 100644 --- a/src/domain/jmap.rs +++ b/src/domain/jmap.rs @@ -55,8 +55,33 @@ pub struct EmailContent { pub is_long: bool, } +/// A channel-neutral page of JMAP Email/changes. Deleted IDs are retained for +/// cursor accounting but are not notified because they have no metadata. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct EmailChanges { + pub new_state: String, + pub created: Vec, + pub updated: Vec, + pub destroyed: Vec, + pub has_more: bool, +} + #[async_trait] pub trait JmapBackend: Send + Sync { + async fn create_push_subscription(&self, callback_url: &str) -> Result; + async fn destroy_push_subscription(&self, subscription_id: &str) -> Result<(), JmapError>; + async fn verify_push_subscription( + &self, + subscription_id: &str, + verification_code: &str, + ) -> Result<(), JmapError>; + async fn current_state(&self) -> Result; + async fn email_changes( + &self, + account_id: &str, + since_state: &str, + max_changes: usize, + ) -> Result; async fn list_folders(&self, account_id: &str) -> Result, JmapError>; async fn list_emails( &self, @@ -64,6 +89,21 @@ pub trait JmapBackend: Send + Sync { folder_id: Option<&str>, limit: usize, ) -> Result, JmapError>; + /// Fetch one bounded page without requiring the caller to materialize the + /// entire mailbox. Implementations with a native JMAP position query should + /// override this; the fallback preserves test backends' existing contract. + async fn list_emails_page( + &self, + account_id: &str, + folder_id: Option<&str>, + position: usize, + limit: usize, + ) -> Result, JmapError> { + let all = self + .list_emails(account_id, folder_id, position.saturating_add(limit)) + .await?; + Ok(all.into_iter().skip(position).take(limit).collect()) + } async fn read_email( &self, account_id: &str, @@ -109,6 +149,49 @@ impl JmapService { } impl JmapService { + pub async fn create_push_subscription(&self, callback_url: &str) -> Result { + if !callback_url.starts_with("https://") { + return Err(JmapError::InvalidRequest("push callback URL")); + } + self.backend.create_push_subscription(callback_url).await + } + + pub async fn destroy_push_subscription(&self, subscription_id: &str) -> Result<(), JmapError> { + if subscription_id.trim().is_empty() { + return Err(JmapError::InvalidRequest("push subscription id")); + } + self.backend + .destroy_push_subscription(subscription_id) + .await + } + pub async fn verify_push_subscription( + &self, + subscription_id: &str, + verification_code: &str, + ) -> Result<(), JmapError> { + if subscription_id.trim().is_empty() || verification_code.trim().is_empty() { + return Err(JmapError::InvalidRequest("push verification")); + } + self.backend + .verify_push_subscription(subscription_id, verification_code) + .await + } + pub async fn current_state(&self) -> Result { + self.backend.current_state().await + } + + pub async fn email_changes( + &self, + since_state: &str, + max_changes: usize, + ) -> Result { + if since_state.trim().is_empty() || max_changes == 0 { + return Err(JmapError::InvalidRequest("changes")); + } + self.backend + .email_changes(&self.account_id, since_state, max_changes) + .await + } pub async fn list_folders(&self) -> Result, JmapError> { self.backend.list_folders(&self.account_id).await } @@ -126,6 +209,20 @@ impl JmapService { .await } + pub async fn list_emails_page( + &self, + folder_id: Option<&str>, + position: usize, + limit: usize, + ) -> Result, JmapError> { + if limit == 0 { + return Err(JmapError::InvalidRequest("limit")); + } + self.backend + .list_emails_page(&self.account_id, folder_id, position, limit) + .await + } + pub async fn read_email(&self, email_id: &str) -> Result, JmapError> { if email_id.trim().is_empty() { return Err(JmapError::InvalidRequest("email_id")); @@ -147,6 +244,43 @@ mod tests { #[async_trait] impl JmapBackend for MockBackend { + async fn destroy_push_subscription(&self, subscription_id: &str) -> Result<(), JmapError> { + assert_eq!(subscription_id, "push-1"); + Ok(()) + } + async fn create_push_subscription(&self, callback_url: &str) -> Result { + assert_eq!(callback_url, "https://bot.example/push/jmap"); + Ok("push-1".into()) + } + async fn verify_push_subscription( + &self, + subscription_id: &str, + verification_code: &str, + ) -> Result<(), JmapError> { + assert_eq!((subscription_id, verification_code), ("push-1", "code")); + Ok(()) + } + async fn current_state(&self) -> Result { + Ok("state-1".into()) + } + async fn email_changes( + &self, + account_id: &str, + since_state: &str, + max_changes: usize, + ) -> Result { + assert_eq!( + (account_id, since_state, max_changes), + ("account-1", "old", 10) + ); + Ok(EmailChanges { + new_state: "new".into(), + created: vec!["email-1".into()], + updated: vec![], + destroyed: vec![], + has_more: false, + }) + } async fn list_folders(&self, account_id: &str) -> Result, JmapError> { assert_eq!(account_id, "account-1"); Ok(vec![Folder { diff --git a/src/domain/jmap/client.rs b/src/domain/jmap/client.rs index fe5db0f..5250d54 100644 --- a/src/domain/jmap/client.rs +++ b/src/domain/jmap/client.rs @@ -1,9 +1,9 @@ //! Concrete read-only adapter (MOD-JMAP-CLIENT, GATE-G1-JMAP-READONLY). -use super::{EmailContent, EmailMetadata, Folder, JmapBackend, JmapError}; +use super::{EmailChanges, EmailContent, EmailMetadata, Folder, JmapBackend, JmapError}; use crate::state::{runtime_provider, OutboundConfig, RuntimeConfigProvider}; use async_trait::async_trait; -use jmap_client::{email, mailbox, Get}; +use jmap_client::{email, mailbox, DataType, Get}; pub struct JmapClientBackend { client: jmap_client::client::Client, @@ -67,6 +67,111 @@ impl JmapClientBackend { #[async_trait] impl JmapBackend for JmapClientBackend { + async fn destroy_push_subscription(&self, subscription_id: &str) -> Result<(), JmapError> { + if subscription_id.trim().is_empty() { + return Err(JmapError::InvalidRequest("push subscription id")); + } + tokio::time::timeout( + self.timeout(), + self.client.push_subscription_destroy(subscription_id), + ) + .await + .map_err(|_| JmapError::Timeout)??; + Ok(()) + } + + async fn create_push_subscription(&self, callback_url: &str) -> Result { + let parsed = url::Url::parse(callback_url) + .map_err(|_| JmapError::InvalidRequest("push callback URL"))?; + if parsed.scheme() != "https" || parsed.username() != "" || parsed.password().is_some() { + return Err(JmapError::InvalidRequest("push callback URL")); + } + let subscription = tokio::time::timeout( + self.timeout(), + self.client.push_subscription_create( + format!("message-weave-{}", self.account_id), + callback_url, + None, + ), + ) + .await + .map_err(|_| JmapError::Timeout)??; + let subscription_id = subscription + .id() + .map(ToOwned::to_owned) + .ok_or(JmapError::MissingField("push subscription id"))?; + // JMAP PushSubscription/set create has no `types` argument in + // jmap-client 0.4.2. Restrict the newly-created subscription with the + // follow-up update before exposing its id (REQ-PUSH-TYPES). + if tokio::time::timeout( + self.timeout(), + self.client.push_subscription_update_types( + &subscription_id, + Some([DataType::Email, DataType::EmailDelivery]), + ), + ) + .await + .map_err(|_| JmapError::Timeout)? + .is_err() + { + let _ = self + .client + .push_subscription_destroy(&subscription_id) + .await; + return Err(JmapError::InvalidRequest("push subscription types")); + } + Ok(subscription_id) + } + + async fn verify_push_subscription( + &self, + subscription_id: &str, + verification_code: &str, + ) -> Result<(), JmapError> { + if subscription_id.trim().is_empty() || verification_code.trim().is_empty() { + return Err(JmapError::InvalidRequest("push verification")); + } + let verified = tokio::time::timeout( + self.timeout(), + self.client + .push_subscription_verify(subscription_id, verification_code), + ) + .await + .map_err(|_| JmapError::Timeout)??; + if verified.is_some() { + Ok(()) + } else { + Err(JmapError::NotFound) + } + } + + async fn current_state(&self) -> Result { + Ok(self.client.session().state().to_owned()) + } + + async fn email_changes( + &self, + account_id: &str, + since_state: &str, + max_changes: usize, + ) -> Result { + if account_id != self.account_id || since_state.trim().is_empty() || max_changes == 0 { + return Err(JmapError::InvalidRequest("email changes")); + } + let mut request = self.client.build(); + request.changes_email(since_state).max_changes(max_changes); + let response = tokio::time::timeout(self.timeout(), request.send_changes_email()) + .await + .map_err(|_| JmapError::Timeout)??; + Ok(EmailChanges { + new_state: response.new_state().to_owned(), + created: response.created().to_owned(), + updated: response.updated().to_owned(), + destroyed: response.destroyed().to_owned(), + has_more: response.has_more_changes(), + }) + } + async fn list_folders(&self, account_id: &str) -> Result, JmapError> { if account_id != self.account_id { return Err(JmapError::InvalidRequest("account_id")); @@ -114,17 +219,32 @@ impl JmapBackend for JmapClientBackend { if account_id != self.account_id || limit == 0 { return Err(JmapError::InvalidRequest("email query")); } - let filter: Option = folder_id.map(email::query::Filter::in_mailbox); - let ids = tokio::time::timeout( - self.timeout(), - self.client.email_query( - filter, - None::>>, - ), - ) - .await - .map_err(|_| JmapError::Timeout)?? - .take_ids(); + // Query in bounded pages. A large limit (used for the cold-start + // baseline) therefore cannot silently stop at the server default page. + let page_size = limit.clamp(1, 100); + let mut position = 0_i32; + let mut ids = Vec::new(); + while ids.len() < limit { + let mut request = self.client.build(); + let query = request.query_email(); + if let Some(folder_id) = folder_id { + query.filter(email::query::Filter::in_mailbox(folder_id)); + } + query.position(position).limit(page_size); + let mut response = tokio::time::timeout(self.timeout(), request.send_query_email()) + .await + .map_err(|_| JmapError::Timeout)??; + let page = response.take_ids(); + if page.is_empty() { + break; + } + position = position.saturating_add(page.len() as i32); + ids.extend(page); + if ids.len() >= limit || position <= 0 { + break; + } + } + ids.truncate(limit); let mut out = Vec::new(); for id in ids.into_iter().take(limit) { if let Some(email) = tokio::time::timeout( @@ -140,6 +260,17 @@ impl JmapBackend for JmapClientBackend { Ok(out) } + async fn list_emails_page( + &self, + account_id: &str, + folder_id: Option<&str>, + position: usize, + limit: usize, + ) -> Result, JmapError> { + self.fetch_email_page(account_id, folder_id, position, limit) + .await + } + async fn read_email( &self, account_id: &str, @@ -182,6 +313,41 @@ impl JmapBackend for JmapClientBackend { } impl JmapClientBackend { + async fn fetch_email_page( + &self, + account_id: &str, + folder_id: Option<&str>, + position: usize, + limit: usize, + ) -> Result, JmapError> { + if account_id != self.account_id || limit == 0 { + return Err(JmapError::InvalidRequest("email query")); + } + let mut request = self.client.build(); + let query = request.query_email(); + if let Some(folder_id) = folder_id { + query.filter(email::query::Filter::in_mailbox(folder_id)); + } + query.position(position as i32).limit(limit.min(100)); + let mut response = tokio::time::timeout(self.timeout(), request.send_query_email()) + .await + .map_err(|_| JmapError::Timeout)??; + let ids = response.take_ids(); + let mut out = Vec::with_capacity(ids.len()); + for id in ids { + if let Some(email) = tokio::time::timeout( + self.timeout(), + self.client.email_get(&id, None::>), + ) + .await + .map_err(|_| JmapError::Timeout)?? + { + out.push(metadata(&email)); + } + } + Ok(out) + } + fn timeout(&self) -> std::time::Duration { std::time::Duration::from_millis( self.runtime diff --git a/src/state.rs b/src/state.rs index 10d9208..1bd5ca8 100644 --- a/src/state.rs +++ b/src/state.rs @@ -70,14 +70,39 @@ pub trait ReliableState: Send + Sync { async fn release_dedup(&self, key: &str) -> Result<(), StateError>; /// XADD enqueue; consumers perform XREADGROUP in a later worker phase. async fn enqueue(&self, stream: &str, payload: &str) -> Result; + /// Atomically claim `dedup_key` and XADD `payload` to `stream`. + /// + /// The dedup key must only be written when the append succeeds: claiming + /// first and rolling back best-effort leaves a claimed-but-not-enqueued key + /// behind on a Redis hiccup, which silently drops the event for the whole + /// dedup TTL (REQ-RECONCILE-IDEMPOTENCY). Returns false when already claimed. + async fn claim_dedup_and_enqueue( + &self, + dedup_key: &str, + ttl_seconds: u64, + stream: &str, + payload: &str, + ) -> Result; /// XACK after successful processing. async fn ack(&self, stream: &str, group: &str, message_id: &str) -> Result<(), StateError>; /// XADD to a dead-letter stream after bounded retry policy decides to stop retrying. async fn dead_letter(&self, stream: &str, payload: &str) -> Result; /// SET NX EX lock for reconcile single-flight. - async fn acquire_lock(&self, key: &str, ttl_seconds: u64) -> Result; + async fn acquire_lock( + &self, + key: &str, + owner_token: &str, + ttl_seconds: u64, + ) -> Result; /// Release a lock after the bounded reconcile attempt has completed. - async fn release_lock(&self, key: &str) -> Result<(), StateError>; + async fn release_lock(&self, key: &str, owner_token: &str) -> Result<(), StateError>; + /// Extend a lock only while the caller still owns it. + async fn renew_lock( + &self, + key: &str, + owner_token: &str, + ttl_seconds: u64, + ) -> Result; async fn read_batch( &self, stream: &str, @@ -117,6 +142,58 @@ pub trait ReliableState: Send + Sync { /// Global durable enable gate; missing/read failures are interpreted by callers as disabled. async fn is_enabled(&self) -> Result; async fn set_enabled(&self, enabled: bool) -> Result<(), StateError>; + /// Durable JMAP Email/changes cursor. It is advanced only after all page + /// events have been enqueued successfully (REQ-RECONCILE-IDEMPOTENCY). + async fn get_reconcile_state(&self) -> Result, StateError>; + async fn set_reconcile_state(&self, state: &str) -> Result<(), StateError>; + /// Short-lived PushSubscription verification state; values are stored as + /// digests and never retained as plaintext credentials. + async fn remember_push_subscription( + &self, + subscription_id: &str, + verification_code: &str, + ttl_seconds: u64, + ) -> Result<(), StateError>; + async fn push_subscription_verified( + &self, + subscription_id: &str, + verification_code: &str, + ) -> Result; + async fn remember_push_subscription_id(&self, subscription_id: &str) -> Result<(), StateError>; + async fn set_push_subscription_status( + &self, + subscription_id: &str, + status: &str, + ttl_seconds: u64, + ) -> Result<(), StateError>; + // NOTE: `push:subscription:{id}:status` is currently write-only + // ("pending"/"verified"/"disabled"). Authorization uses the verification + // digest instead; the status key stays as a lightweight ops/observability + // trail (inspectable via redis-cli), not a gate (建议-5a). + /// Drop the verification-code digest for `subscription_id`. + /// + /// Callers must invoke this when disabling a subscription: otherwise a push + /// arriving within the residual TTL of the last verify call can still pass + /// `push_subscription_verified` and be enqueued (SAF-AUTH-JMAP-PUSH). + async fn forget_push_subscription(&self, subscription_id: &str) -> Result<(), StateError>; + async fn record_push_orphan( + &self, + subscription_id: &str, + request_id: &str, + ) -> Result<(), StateError>; + async fn get_push_subscription_for_callback( + &self, + callback_url: &str, + ) -> Result, StateError>; + async fn remember_push_subscription_for_callback( + &self, + callback_url: &str, + subscription_id: &str, + ) -> Result<(), StateError>; + async fn remove_push_subscription_for_callback( + &self, + callback_url: &str, + ) -> Result<(), StateError>; } pub struct RedisState { @@ -165,9 +242,6 @@ impl ReliableState for RedisState { let mut connection = self.connection.clone(); Ok(redis::cmd("XADD") .arg(stream) - .arg("MAXLEN") - .arg("~") - .arg(10_000) .arg("*") .arg("payload") .arg(payload) @@ -175,6 +249,34 @@ impl ReliableState for RedisState { .await?) } + async fn claim_dedup_and_enqueue( + &self, + dedup_key: &str, + ttl_seconds: u64, + stream: &str, + payload: &str, + ) -> Result { + let mut connection = self.connection.clone(); + // KEYS[1] = dedup key, KEYS[2] = stream; ARGV[1] = ttl, ARGV[2] = payload. + // `SET NX` returning false means another worker owns the event; skip the + // append so the XADD is only attempted once per dedup window. + let claimed: i64 = redis::Script::new( + "if redis.call('set', KEYS[1], '1', 'NX', 'EX', ARGV[1]) then \ + redis.call('xadd', KEYS[2], '*', 'payload', ARGV[2]) \ + return 1 \ + else \ + return 0 \ + end", + ) + .key(dedup_key) + .key(stream) + .arg(ttl_seconds.max(1)) + .arg(payload) + .invoke_async(&mut connection) + .await?; + Ok(claimed != 0) + } + async fn ack(&self, stream: &str, group: &str, message_id: &str) -> Result<(), StateError> { let mut connection = self.connection.clone(); let _: i64 = redis::cmd("XACK") @@ -190,12 +292,52 @@ impl ReliableState for RedisState { self.enqueue(stream, payload).await } - async fn acquire_lock(&self, key: &str, ttl_seconds: u64) -> Result { - set_nx_ex(self.connection.clone(), key, ttl_seconds).await + async fn acquire_lock( + &self, + key: &str, + owner_token: &str, + ttl_seconds: u64, + ) -> Result { + let mut connection = self.connection.clone(); + let result: Option = redis::cmd("SET") + .arg(key) + .arg(owner_token) + .arg("NX") + .arg("EX") + .arg(ttl_seconds.max(1)) + .query_async(&mut connection) + .await?; + Ok(result.is_some()) } - async fn release_lock(&self, key: &str) -> Result<(), StateError> { - delete_key(self.connection.clone(), key).await + async fn release_lock(&self, key: &str, owner_token: &str) -> Result<(), StateError> { + let mut connection = self.connection.clone(); + redis::Script::new( + "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end", + ) + .key(key) + .arg(owner_token) + .invoke_async::(&mut connection) + .await?; + Ok(()) + } + + async fn renew_lock( + &self, + key: &str, + owner_token: &str, + ttl_seconds: u64, + ) -> Result { + let mut connection = self.connection.clone(); + let renewed: i64 = redis::Script::new( + "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('expire', KEYS[1], ARGV[2]) else return 0 end", + ) + .key(key) + .arg(owner_token) + .arg(ttl_seconds.max(1)) + .invoke_async(&mut connection) + .await?; + Ok(renewed != 0) } async fn read_batch( @@ -217,25 +359,38 @@ impl ReliableState for RedisState { .query_async(&mut connection) .await; // Recover messages left in the consumer group's PEL after a crash or - // restart before reading new entries. A bounded idle window prevents - // two short-lived workers from stealing active work. + // 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 + // 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 reclaimed: redis::streams::StreamAutoClaimReply = redis::cmd("XAUTOCLAIM") .arg(stream) .arg(group) .arg(consumer) - // A worker handles at most ten events per request; keep this - // above the worst bounded JMAP/Telegram batch duration. - .arg(300_000) + .arg(idle_threshold_ms) .arg("0-0") .arg("COUNT") .arg(count.max(1)) .query_async(&mut connection) .await?; - let reclaimed_messages = reclaimed - .claimed - .into_iter() - .filter_map(stream_id_to_message) - .collect::>(); + let mut reclaimed_messages = Vec::new(); + let mut malformed_ids = Vec::new(); + for item in reclaimed.claimed { + let id = item.id.clone(); + match stream_id_to_message(item) { + Some(message) => reclaimed_messages.push(message), + None => malformed_ids.push(id), + } + } + // A record without a `payload` field can never be processed. ACK it so + // it stops being re-claimed every cycle and squatting in the PEL + // (建议-7); a payload that is present but not valid JSON still routes + // through the normal Err -> DLQ path. + ack_malformed(&mut connection, stream, group, &malformed_ids).await; if !reclaimed_messages.is_empty() { return Ok(reclaimed_messages); } @@ -250,11 +405,21 @@ impl ReliableState for RedisState { .arg(">") .query_async(&mut connection) .await?; - Ok(reply - .keys - .into_iter() - .flat_map(|key| key.ids.into_iter().filter_map(stream_id_to_message)) - .collect()) + let mut messages = Vec::new(); + let mut malformed_ids = Vec::new(); + for key in reply.keys { + for item in key.ids { + let id = item.id.clone(); + match stream_id_to_message(item) { + Some(message) => messages.push(message), + None => malformed_ids.push(id), + } + } + } + // Same handling as the XAUTOCLAIM path: drop unprocessable records from + // the PEL instead of letting them spin forever (建议-7). + ack_malformed(&mut connection, stream, group, &malformed_ids).await; + Ok(messages) } async fn retry_or_dlq( @@ -273,7 +438,7 @@ impl ReliableState for RedisState { r#"local n = redis.call('INCR', KEYS[1]) redis.call('EXPIRE', KEYS[1], 86400) if n < tonumber(ARGV[1]) then return 0 end - redis.call('XADD', ARGV[2], 'MAXLEN', '~', 10000, '*', 'payload', ARGV[3]) + redis.call('XADD', ARGV[2], '*', 'payload', ARGV[3]) redis.call('XACK', ARGV[4], ARGV[5], ARGV[6]) return 1"#, ); @@ -452,6 +617,152 @@ impl ReliableState for RedisState { .await?; Ok(()) } + + async fn get_reconcile_state(&self) -> Result, StateError> { + let mut connection = self.connection.clone(); + Ok(redis::cmd("GET") + .arg("state:jmap:since") + .query_async(&mut connection) + .await?) + } + + async fn set_reconcile_state(&self, state: &str) -> Result<(), StateError> { + let mut connection = self.connection.clone(); + redis::cmd("SET") + .arg("state:jmap:since") + .arg(state) + .query_async::<()>(&mut connection) + .await?; + Ok(()) + } + + async fn remember_push_subscription( + &self, + subscription_id: &str, + verification_code: &str, + ttl_seconds: u64, + ) -> Result<(), StateError> { + let mut connection = self.connection.clone(); + redis::cmd("SET") + .arg(format!("push:subscription:{subscription_id}")) + .arg(crate::config::session_digest(verification_code)) + .arg("EX") + .arg(ttl_seconds.max(1)) + .query_async::<()>(&mut connection) + .await?; + Ok(()) + } + + async fn push_subscription_verified( + &self, + subscription_id: &str, + verification_code: &str, + ) -> Result { + let mut connection = self.connection.clone(); + let current: Option = redis::cmd("GET") + .arg(format!("push:subscription:{subscription_id}")) + .query_async(&mut connection) + .await?; + Ok(current.as_deref() == Some(crate::config::session_digest(verification_code).as_str())) + } + + async fn remember_push_subscription_id(&self, subscription_id: &str) -> Result<(), StateError> { + let mut connection = self.connection.clone(); + redis::cmd("SET") + .arg("push:subscription:id") + .arg(subscription_id) + .query_async::<()>(&mut connection) + .await?; + Ok(()) + } + + async fn set_push_subscription_status( + &self, + subscription_id: &str, + status: &str, + ttl_seconds: u64, + ) -> Result<(), StateError> { + let mut connection = self.connection.clone(); + redis::cmd("SET") + .arg(format!("push:subscription:{subscription_id}:status")) + .arg(status) + .arg("EX") + .arg(ttl_seconds.max(1)) + .query_async::<()>(&mut connection) + .await?; + Ok(()) + } + + async fn forget_push_subscription(&self, subscription_id: &str) -> Result<(), StateError> { + let mut connection = self.connection.clone(); + redis::cmd("DEL") + .arg(format!("push:subscription:{subscription_id}")) + .query_async::<()>(&mut connection) + .await?; + Ok(()) + } + + async fn record_push_orphan( + &self, + subscription_id: &str, + request_id: &str, + ) -> Result<(), StateError> { + let mut connection = self.connection.clone(); + redis::cmd("SET") + .arg(format!("push:orphan:{subscription_id}")) + .arg(request_id) + .arg("EX") + .arg(604_800_u64) + .query_async::<()>(&mut connection) + .await?; + Ok(()) + } + + async fn get_push_subscription_for_callback( + &self, + callback_url: &str, + ) -> Result, StateError> { + let mut connection = self.connection.clone(); + Ok(redis::cmd("GET") + .arg(push_registration_key(callback_url)) + .query_async(&mut connection) + .await?) + } + + async fn remember_push_subscription_for_callback( + &self, + callback_url: &str, + subscription_id: &str, + ) -> Result<(), StateError> { + let mut connection = self.connection.clone(); + redis::cmd("SET") + .arg(push_registration_key(callback_url)) + .arg(subscription_id) + .arg("EX") + .arg(604_800_u64) + .query_async::<()>(&mut connection) + .await?; + Ok(()) + } + + async fn remove_push_subscription_for_callback( + &self, + callback_url: &str, + ) -> Result<(), StateError> { + let mut connection = self.connection.clone(); + redis::cmd("DEL") + .arg(push_registration_key(callback_url)) + .query_async::<()>(&mut connection) + .await?; + Ok(()) + } +} + +fn push_registration_key(callback_url: &str) -> String { + format!( + "push:registration:{}", + crate::config::session_digest(callback_url) + ) } fn stream_id_to_message(item: redis::streams::StreamId) -> Option { @@ -465,6 +776,36 @@ fn stream_id_to_message(item: redis::streams::StreamId) -> Option }) } +/// XACK stream records that arrived without a parseable `payload` field. These +/// can never be processed, so acknowledging them keeps the consumer group's PEL +/// from filling up with unclaimable junk; the failure is logged for operators. +async fn ack_malformed( + connection: &mut redis::aio::MultiplexedConnection, + stream: &str, + group: &str, + ids: &[String], +) { + if ids.is_empty() { + return; + } + let ack = redis::cmd("XACK") + .arg(stream) + .arg(group) + .arg(ids) + .query_async::(connection) + .await; + if ack + .as_ref() + .map(|acked| *acked != ids.len() as i64) + .unwrap_or(true) + { + eprintln!( + "WARN malformed stream records ack incomplete (stream={stream}, group={group}, expected={}, result={ack:?})", + ids.len() + ); + } +} + async fn set_nx_ex( mut connection: redis::aio::MultiplexedConnection, key: &str, @@ -475,7 +816,9 @@ async fn set_nx_ex( .arg("1") .arg("NX") .arg("EX") - .arg(ttl_seconds) + // Guard against a ttl of 0, which Redis rejects with "invalid expire + // time" (same normalization as acquire_lock). + .arg(ttl_seconds.max(1)) .query_async(&mut connection) .await?; Ok(result.is_some()) @@ -498,7 +841,8 @@ impl MemoryState { #[derive(Default)] struct MemoryInner { dedup: HashSet, - locks: HashSet, + locks: HashMap, + push_registrations: HashMap, streams: HashMap>, retries: HashMap<(String, String), u32>, consent: HashMap, @@ -508,6 +852,11 @@ struct MemoryInner { admin_session: Option<(String, i64)>, business_revision: u64, enabled: bool, + reconcile_state: Option, + push_subscriptions: HashMap, + push_subscription_id: Option, + push_status: HashMap, + push_orphans: HashMap, } #[async_trait] @@ -534,14 +883,30 @@ impl ReliableState for MemoryState { .entry(stream.to_owned()) .or_default() .push((id.clone(), payload.to_owned())); - if let Some(messages) = inner.streams.get_mut(stream) { - let excess = messages.len().saturating_sub(10_000); - if excess > 0 { - messages.drain(..excess); - } - } Ok(id) } + async fn claim_dedup_and_enqueue( + &self, + dedup_key: &str, + _ttl_seconds: u64, + stream: &str, + payload: &str, + ) -> Result { + // Hold the lock across claim + append so MemoryState honors the same + // atomicity contract as the Lua-backed RedisState implementation. + let mut inner = self.inner.lock().map_err(|_| StateError::Poisoned)?; + if !inner.dedup.insert(dedup_key.to_owned()) { + return Ok(false); + } + inner.next_id += 1; + let id = format!("{}-0", inner.next_id); + inner + .streams + .entry(stream.to_owned()) + .or_default() + .push((id, payload.to_owned())); + Ok(true) + } async fn ack(&self, stream: &str, _group: &str, message_id: &str) -> Result<(), StateError> { let mut inner = self.inner.lock().map_err(|_| StateError::Poisoned)?; if let Some(messages) = inner.streams.get_mut(stream) { @@ -552,15 +917,43 @@ impl ReliableState for MemoryState { async fn dead_letter(&self, stream: &str, payload: &str) -> Result { self.enqueue(stream, payload).await } - async fn acquire_lock(&self, key: &str, _ttl_seconds: u64) -> Result { + async fn acquire_lock( + &self, + key: &str, + owner_token: &str, + _ttl_seconds: u64, + ) -> Result { let mut inner = self.inner.lock().map_err(|_| StateError::Poisoned)?; - Ok(inner.locks.insert(key.to_owned())) + if inner.locks.contains_key(key) { + return Ok(false); + } + inner.locks.insert(key.to_owned(), owner_token.to_owned()); + Ok(true) } - async fn release_lock(&self, key: &str) -> Result<(), StateError> { + async fn release_lock(&self, key: &str, owner_token: &str) -> Result<(), StateError> { let mut inner = self.inner.lock().map_err(|_| StateError::Poisoned)?; - inner.locks.remove(key); + if inner + .locks + .get(key) + .is_some_and(|owner| owner == owner_token) + { + inner.locks.remove(key); + } Ok(()) } + + async fn renew_lock( + &self, + key: &str, + owner_token: &str, + _ttl_seconds: u64, + ) -> Result { + let inner = self.inner.lock().map_err(|_| StateError::Poisoned)?; + Ok(inner + .locks + .get(key) + .is_some_and(|owner| owner == owner_token)) + } async fn read_batch( &self, stream: &str, @@ -714,6 +1107,147 @@ impl ReliableState for MemoryState { self.inner.lock().map_err(|_| StateError::Poisoned)?.enabled = enabled; Ok(()) } + + async fn get_reconcile_state(&self) -> Result, StateError> { + Ok(self + .inner + .lock() + .map_err(|_| StateError::Poisoned)? + .reconcile_state + .clone()) + } + + async fn set_reconcile_state(&self, state: &str) -> Result<(), StateError> { + self.inner + .lock() + .map_err(|_| StateError::Poisoned)? + .reconcile_state = Some(state.to_owned()); + Ok(()) + } + + async fn remember_push_subscription( + &self, + subscription_id: &str, + verification_code: &str, + ttl_seconds: u64, + ) -> Result<(), StateError> { + self.inner + .lock() + .map_err(|_| StateError::Poisoned)? + .push_subscriptions + .insert( + subscription_id.to_owned(), + ( + crate::config::session_digest(verification_code), + unix_now().saturating_add(ttl_seconds as i64), + ), + ); + Ok(()) + } + + async fn push_subscription_verified( + &self, + subscription_id: &str, + verification_code: &str, + ) -> Result { + let mut inner = self.inner.lock().map_err(|_| StateError::Poisoned)?; + let Some((digest, expiry)) = inner.push_subscriptions.get(subscription_id) else { + return Ok(false); + }; + if *expiry <= unix_now() { + inner.push_subscriptions.remove(subscription_id); + return Ok(false); + } + Ok(digest == &crate::config::session_digest(verification_code)) + } + + async fn remember_push_subscription_id(&self, subscription_id: &str) -> Result<(), StateError> { + self.inner + .lock() + .map_err(|_| StateError::Poisoned)? + .push_subscription_id = Some(subscription_id.to_owned()); + Ok(()) + } + + async fn get_push_subscription_for_callback( + &self, + callback_url: &str, + ) -> Result, StateError> { + Ok(self + .inner + .lock() + .map_err(|_| StateError::Poisoned)? + .push_registrations + .get(&push_registration_key(callback_url)) + .cloned()) + } + + async fn remember_push_subscription_for_callback( + &self, + callback_url: &str, + subscription_id: &str, + ) -> Result<(), StateError> { + self.inner + .lock() + .map_err(|_| StateError::Poisoned)? + .push_registrations + .insert( + push_registration_key(callback_url), + subscription_id.to_owned(), + ); + Ok(()) + } + + async fn remove_push_subscription_for_callback( + &self, + callback_url: &str, + ) -> Result<(), StateError> { + self.inner + .lock() + .map_err(|_| StateError::Poisoned)? + .push_registrations + .remove(&push_registration_key(callback_url)); + Ok(()) + } + + async fn set_push_subscription_status( + &self, + subscription_id: &str, + status: &str, + ttl_seconds: u64, + ) -> Result<(), StateError> { + self.inner + .lock() + .map_err(|_| StateError::Poisoned)? + .push_status + .insert( + subscription_id.to_owned(), + ( + status.to_owned(), + unix_now().saturating_add(ttl_seconds as i64), + ), + ); + Ok(()) + } + + async fn forget_push_subscription(&self, subscription_id: &str) -> Result<(), StateError> { + let mut inner = self.inner.lock().map_err(|_| StateError::Poisoned)?; + inner.push_subscriptions.remove(subscription_id); + Ok(()) + } + + async fn record_push_orphan( + &self, + subscription_id: &str, + request_id: &str, + ) -> Result<(), StateError> { + self.inner + .lock() + .map_err(|_| StateError::Poisoned)? + .push_orphans + .insert(subscription_id.to_owned(), request_id.to_owned()); + Ok(()) + } } fn unix_now() -> i64 { @@ -830,6 +1364,17 @@ mod tests { #[tokio::test] async fn memory_state_is_atomic_and_supports_stream_lifecycle() { let state = MemoryState::default(); + assert!(!state.is_enabled().await.unwrap()); + state.set_enabled(true).await.unwrap(); + assert!(state.is_enabled().await.unwrap()); + state.set_enabled(false).await.unwrap(); + assert!(!state.is_enabled().await.unwrap()); + assert!(state.get_reconcile_state().await.unwrap().is_none()); + state.set_reconcile_state("jmap-state-1").await.unwrap(); + assert_eq!( + state.get_reconcile_state().await.unwrap().as_deref(), + Some("jmap-state-1") + ); assert!(state.claim_dedup("tg:42", 60).await.unwrap()); assert!(!state.claim_dedup("tg:42", 60).await.unwrap()); assert!(state.dedup_exists("tg:42").await.unwrap()); @@ -857,8 +1402,56 @@ mod tests { .dedup_exists("delivery:committed:events:1-0") .await .unwrap()); - assert!(state.acquire_lock("reconcile", 30).await.unwrap()); - assert!(!state.acquire_lock("reconcile", 30).await.unwrap()); + assert!(state + .acquire_lock("reconcile", "owner-a", 30) + .await + .unwrap()); + assert!(!state + .acquire_lock("reconcile", "owner-b", 30) + .await + .unwrap()); + state.release_lock("reconcile", "owner-b").await.unwrap(); + assert!(!state + .acquire_lock("reconcile", "owner-b", 30) + .await + .unwrap()); + assert!(!state.renew_lock("reconcile", "owner-b", 90).await.unwrap()); + assert!(state.renew_lock("reconcile", "owner-a", 90).await.unwrap()); + state.release_lock("reconcile", "owner-a").await.unwrap(); + assert!(state + .acquire_lock("reconcile", "owner-b", 30) + .await + .unwrap()); + state + .remember_push_subscription_for_callback( + "https://example.test/push/jmap", + "subscription-1", + ) + .await + .unwrap(); + assert_eq!( + state + .get_push_subscription_for_callback("https://example.test/push/jmap") + .await + .unwrap() + .as_deref(), + Some("subscription-1") + ); + state + .set_push_subscription_status("subscription-1", "disabled", 60) + .await + .unwrap(); + // `pending/verified/disabled` 状态键是只写的运维轨迹（建议-5a）： + // 授权语义由验证码摘要承担，故这里不对其做读取断言。 + state + .remove_push_subscription_for_callback("https://example.test/push/jmap") + .await + .unwrap(); + assert!(state + .get_push_subscription_for_callback("https://example.test/push/jmap") + .await + .unwrap() + .is_none()); assert!(state.ai_consent_until(7).await.unwrap().is_none()); state.set_ai_consent(7, 3600).await.unwrap(); assert!(state.ai_consent_until(7).await.unwrap().is_some()); diff --git a/src/worker.rs b/src/worker.rs index ec2aecc..4e2e673 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -15,6 +15,32 @@ use std::{ #[async_trait] pub trait WorkerHandler: Send + Sync { async fn process(&self, stream: &str, payload: &str) -> Result<(), ()>; + + async fn create_push_subscription(&self, _callback_url: &str) -> Result { + Err(()) + } + + async fn destroy_push_subscription(&self, _subscription_id: &str) -> Result<(), ()> { + Err(()) + } + + async fn verify_push_subscription( + &self, + _subscription_id: &str, + _verification_code: &str, + ) -> Result<(), ()> { + Err(()) + } + + /// Performs one bounded JMAP reconciliation pass. Implementations must not + /// persist the cursor; the HTTP coordinator commits it after enqueueing. + async fn reconcile( + &self, + _since_state: Option<&str>, + _max_changes: usize, + ) -> Result { + Err(()) + } } /// Atomically replaceable worker handle. In-flight requests keep their old worker; @@ -81,10 +107,36 @@ impl ReloadCoordinator { #[async_trait] impl WorkerHandler for WorkerHandle { + async fn destroy_push_subscription(&self, subscription_id: &str) -> Result<(), ()> { + let worker = self.current.read().map_err(|_| ())?.clone(); + worker.destroy_push_subscription(subscription_id).await + } + + async fn create_push_subscription(&self, callback_url: &str) -> Result { + let worker = self.current.read().map_err(|_| ())?.clone(); + worker.create_push_subscription(callback_url).await + } + + async fn verify_push_subscription( + &self, + subscription_id: &str, + verification_code: &str, + ) -> Result<(), ()> { + let worker = self.current.read().map_err(|_| ())?.clone(); + worker + .verify_push_subscription(subscription_id, verification_code) + .await + } + async fn process(&self, stream: &str, payload: &str) -> Result<(), ()> { let worker = self.current.read().map_err(|_| ())?.clone(); worker.process(stream, payload).await } + + async fn reconcile(&self, since_state: Option<&str>, max_changes: usize) -> Result { + let worker = self.current.read().map_err(|_| ())?.clone(); + worker.reconcile(since_state, max_changes).await + } } pub struct NoopWorker; @@ -124,12 +176,41 @@ impl MetadataWorker { #[async_trait] impl WorkerHandler for MetadataWorker { + async fn destroy_push_subscription(&self, subscription_id: &str) -> Result<(), ()> { + self.jmap + .destroy_push_subscription(subscription_id) + .await + .map_err(|_| ()) + } + + async fn create_push_subscription(&self, callback_url: &str) -> Result { + self.jmap + .create_push_subscription(callback_url) + .await + .map_err(|_| ()) + } + + async fn verify_push_subscription( + &self, + subscription_id: &str, + verification_code: &str, + ) -> Result<(), ()> { + self.jmap + .verify_push_subscription(subscription_id, verification_code) + .await + .map_err(|_| ()) + } + async fn process(&self, stream: &str, payload: &str) -> Result<(), ()> { if stream == "stalwart:telegram" { return self.process_telegram(payload).await; } if stream != "stalwart:jmap" { - return Ok(()); + // Unknown stream: fail closed. Returning Ok would let the coordinator + // write `delivery:committed` + XACK and silently drop an event we do + // not understand. Falling through to Err routes it to retry/DLQ so it + // stays observable instead of vanishing. + return Err(()); } let event: WorkerEvent = serde_json::from_str(payload).map_err(|_| ())?; let (Some(account), Some(email)) = (event.account_id.as_deref(), event.email_id.as_deref()) @@ -161,9 +242,140 @@ impl WorkerHandler for MetadataWorker { .await .map_err(|_| ()) } + + async fn reconcile(&self, since_state: Option<&str>, max_changes: usize) -> Result { + let max_changes = max_changes.max(1); + const MAX_BASELINE_PAGES: usize = 100; + const MAX_BASELINE_EMAILS: usize = 10_000; + const RECONCILE_BUDGET: std::time::Duration = std::time::Duration::from_secs(20); + let started = tokio::time::Instant::now(); + let (mut cursor, mut baseline_position) = if let Some(state) = since_state { + if let Some((baseline, position)) = decode_baseline_cursor(state) { + (baseline, position) + } else { + (state.to_owned(), usize::MAX) + } + } else { + // Capture a stable baseline before paging. Changes occurring while + // the listing is in flight are consumed below before committing. + (self.jmap.current_state().await.map_err(|_| ())?, 0) + }; + + if baseline_position != usize::MAX { + let mut pages = 0_usize; + let mut emails = 0_usize; + loop { + if pages >= MAX_BASELINE_PAGES + || emails >= MAX_BASELINE_EMAILS + || started.elapsed() >= RECONCILE_BUDGET + { + return if pages == 0 { + Err(()) + } else { + Ok(encode_baseline_cursor(&cursor, baseline_position)) + }; + } + let page = self + .jmap + .list_emails_page(None, baseline_position, 100) + .await + .map_err(|_| ())?; + if page.is_empty() { + break; + } + pages += 1; + emails = emails.saturating_add(page.len()); + baseline_position = baseline_position.saturating_add(page.len()); + for email in page { + self.enqueue_reconcile_event(&email.id).await?; + } + } + } + + // RFC 8620 lets a server answer `/changes` with `hasMoreChanges=true` + // when more events remain than `maxChanges` allowed. The spec-blessed + // continuation is `sinceState = newState`; but a server that resolves + // `newState` to "after *all* pending changes" would then hand us a + // cursor that silently skips the unreturned batch. We cannot pass + // `upToId` here (jmap-client 0.4.2 does not expose it), so while more + // changes remain we keep the *same* sinceState and widen the window + // instead. The 24h dedup key makes re-reading the superset idempotent, + // so this is at-most-duplicate, never-loss. Only when the window stops + // growing (server-side cap) do we fall back to advancing to `new_state` + // to guarantee forward progress. + const CHANGE_WINDOW_CAP: usize = 4_096; + let mut window = max_changes; + let mut pages = 0_usize; + loop { + if pages >= 100 || started.elapsed() >= RECONCILE_BUDGET { + // Every event from the last completed page is already enqueued. + // Returning that cursor lets the next invocation continue from + // it instead of retrying the same bounded window forever. + return if pages == 0 { Err(()) } else { Ok(cursor) }; + } + pages += 1; + let changes = self + .jmap + .email_changes(&cursor, window) + .await + .map_err(|_| ())?; + for email_id in changes.created.iter().chain(changes.updated.iter()) { + self.enqueue_reconcile_event(email_id).await?; + } + if !changes.has_more { + return Ok(changes.new_state); + } + if window < CHANGE_WINDOW_CAP { + window = window.saturating_mul(2).min(CHANGE_WINDOW_CAP); + continue; + } + // Window is already at the cap and the server still reports more: + // advancing is the only way to make progress. + cursor = changes.new_state; + } + } +} + +fn encode_baseline_cursor(state: &str, position: usize) -> String { + let encoded = state + .as_bytes() + .iter() + .map(|byte| format!("{byte:02x}")) + .collect::(); + format!("baseline:{encoded}:{position}") +} + +fn decode_baseline_cursor(value: &str) -> Option<(String, usize)> { + let rest = value.strip_prefix("baseline:")?; + let (encoded, position) = rest.rsplit_once(':')?; + let bytes = (0..encoded.len()) + .step_by(2) + .map(|index| u8::from_str_radix(encoded.get(index..index + 2)?, 16).ok()) + .collect::>>()?; + Some((String::from_utf8(bytes).ok()?, position.parse().ok()?)) } impl MetadataWorker { + async fn enqueue_reconcile_event(&self, email_id: &str) -> Result<(), ()> { + let key = format!("dedup:jmap:{}:{}", self.jmap.account_id(), email_id); + let payload = serde_json::json!({ + "accountId": self.jmap.account_id(), + "emailId": email_id, + }) + .to_string(); + // Claim the dedup key and XADD in a single atomic step: a Redis hiccup + // mid-way must not leave the dedup key claimed while the payload is + // absent from the stream (which would drop the event for the whole + // 24h dedup window). Returning false means another worker owns it. + // `false` means another worker already owns the dedup entry for this + // email in the current window; either way the event is handled here. + self.state + .claim_dedup_and_enqueue(&key, 86_400, "stalwart:jmap", &payload) + .await + .map_err(|_| ())?; + Ok(()) + } + async fn process_telegram(&self, payload: &str) -> Result<(), ()> { let update: TelegramUpdate = serde_json::from_str(payload).map_err(|_| ())?; let Some(message) = update.message else { @@ -354,7 +566,7 @@ fn parse_intent(input: &str) -> Intent { } fn help_message() -> &'static str { - "使用说明：\n邮件到达后会先发送发件人、主题和时间等元数据通知。\n\n授权示例：/ai on（1小时）、临时一次、1小时、今天、7天、直到我撤销。\n/ai off：立即撤销 AI 正文授权。\n/summary ：请求指定邮件摘要；未授权或授权到期时只返回元数据并提示重新授权。\n/help（或 help）：显示本说明。\n\n隐私：AI 默认关闭，只有你明确开启后才会发送正文；授权有期限且不会自动续期；正文和 AI 结果不会持久化，也不会写入日志。\n如果未配置 AI 或 AI 调用失败，将回退为本地截取摘要。" + "使用说明：\n邮件到达后会先发送发件人、主题和时间等元数据通知。\n\n授权示例（请按原文中文输入，不支持英文别名）：\n1小时：/ai on、临时一次、临时、一次\n今天：今天\n7天：7天\n直到我撤销（最长365天）：直到我撤销、长期\n撤销授权：/ai off、关闭 ai、撤销授权、停止摘要\n/summary ：请求指定邮件摘要；未授权或授权到期时只返回元数据并提示重新授权。\n/help（或 help）：显示本说明。\n\n隐私：AI 默认关闭，只有你明确开启后才会发送正文；授权有期限且不会自动续期；正文和 AI 结果不会持久化，也不会写入日志。\n如果未配置 AI 或 AI 调用失败，将回退为本地截取摘要。" } #[derive(serde::Deserialize)] @@ -444,7 +656,6 @@ mod reload_tests { jmap_session_url: "https://mail.example.test".into(), jmap_username: "user".into(), jmap_password: SecretString::new("pass".into()), - jmap_push_verification: SecretString::new("verify".into()), account_id: None, llm_enabled: false, llm_allow_net: false, @@ -482,6 +693,16 @@ mod reload_tests { assert!(handle.process("x", "y").await.is_ok()); } + #[test] + fn baseline_cursor_roundtrips_state_and_position() { + let encoded = encode_baseline_cursor("state:with:punctuation", 1234); + assert_eq!( + decode_baseline_cursor(&encoded), + Some(("state:with:punctuation".to_owned(), 1234)) + ); + assert!(decode_baseline_cursor("baseline:odd:hex").is_none()); + } + struct OkWorker; #[async_trait] impl WorkerHandler for OkWorker {