diff --git a/src/notify.rs b/src/notify.rs index b6811d1..77c5acb 100644 --- a/src/notify.rs +++ b/src/notify.rs @@ -1310,6 +1310,159 @@ struct PushDisable { callback_url: String, } +/// `destroy_push_registration` 之后本地清理失败时的响应。Redis 写入失败与 +/// 订阅销毁失败是两种不同故障：前者与订阅无关（不重试），后者意味着订阅仍 +/// 存活（可重试，带追踪 request_id）。 +enum PushCleanupFailure { + /// Redis 写入失败。 + StateUnavailable, + /// Worker 拒绝销毁：订阅仍然存活，已登记为孤儿；携带追踪用 request_id。 + DestroyFailed(String), +} + +impl PushCleanupFailure { + fn response(self) -> Response { + match self { + PushCleanupFailure::StateUnavailable => error_response( + StatusCode::SERVICE_UNAVAILABLE, + "push_state_unavailable", + true, + ), + PushCleanupFailure::DestroyFailed(request_id) => error_response_with_id( + StatusCode::SERVICE_UNAVAILABLE, + "push_destroy_failed", + request_id, + ), + } + } +} + +/// 销毁订阅并清理它的全部本地状态（per-URL 映射、status、verification 摘要）。 +/// +/// 销毁必须先于删映射：JMAP 瞬时失败时应保留映射让调用方重试，而不是静默丢 +/// 登记。返回 `Err` 时订阅的状态不确定，调用方必须中止并原样上报。 +async fn destroy_push_registration( + app: &AppState, + callback_url: &str, + subscription_id: &str, +) -> Result<(), PushCleanupFailure> { + if app + .worker + .destroy_push_subscription(subscription_id) + .await + .is_err() + { + return Err(PushCleanupFailure::DestroyFailed( + cleanup_push_orphan(app, subscription_id).await, + )); + } + if app + .state + .set_push_subscription_status(subscription_id, "disabled", ttl::PUSH_DISABLED_SECONDS) + .await + .is_err() + { + return Err(PushCleanupFailure::StateUnavailable); + } + // 一并删掉 verification 摘要：否则订阅销毁后、旧 TTL 窗口内到达的 push + // 仍能通过 `push_subscription_verified` 并进入队列。 + if app + .state + .forget_push_subscription(subscription_id) + .await + .is_err() + { + return Err(PushCleanupFailure::StateUnavailable); + } + if app + .state + .remove_push_subscription_for_callback(callback_url) + .await + .is_err() + { + return Err(PushCleanupFailure::StateUnavailable); + } + Ok(()) +} + +/// 注销掉的是 current 地址时清掉 current 指针，否则它会指向一个已 disabled +/// 的订阅。读取失败不影响注销结果：指针陈旧只会让下一次 register 多做一次 +/// 落空的查找。 +async fn clear_push_registration_current_if(app: &AppState, callback_url: &str) { + let Ok(current) = app.state.get_push_registration_current().await else { + return; + }; + if current.as_deref() != Some(callback_url) { + return; + } + let _ = app.state.clear_push_registration_current().await; +} + +/// 更换回调地址前注销 current 指针指向的旧订阅。 +/// +/// 返回 `Some(response)` 表示本次注册必须中止并把该响应返回给调用方； +/// `None` 表示没有需要注销的旧订阅。返回 503 而不是静默跳过是刻意的：跳过 +/// 会留下一个仍在收推送的旧订阅，Stalwart 就会对同一邮件事件向新旧两个 +/// 地址各推一次（重复通知）。 +async fn retire_previous_push_registration( + app: &AppState, + registration_lock: &str, + registration_owner: &str, + callback_url: &str, +) -> Option { + let current = match app.state.get_push_registration_current().await { + Ok(current) => current, + Err(_) => { + let _ = app + .state + .release_lock(registration_lock, registration_owner) + .await; + return Some(error_response( + StatusCode::SERVICE_UNAVAILABLE, + "service_unavailable", + true, + )); + } + }; + let previous_url = current?; + if previous_url == callback_url { + return None; + } + let Ok(existing) = app + .state + .get_push_subscription_for_callback(&previous_url) + .await + else { + let _ = app + .state + .release_lock(registration_lock, registration_owner) + .await; + return Some(error_response( + StatusCode::SERVICE_UNAVAILABLE, + "service_unavailable", + true, + )); + }; + let Some(subscription_id) = existing else { + // 防御分支：current 指针还在，但 per-URL 映射已经丢了。拿不到旧订阅 id + // 就无从销毁，也不能假定它已消失：JMAP push subscription 没有服务端 + // TTL，只在回调不可达时由服务端尝试删除，与本地这份映射的 TTL 无关。 + // 指针和映射同用 PUSH_REGISTRATION_SECONDS 且总是相邻写入，正常会一起 + // 过期，所以这里接近不可达。退化为按「无旧订阅」继续注册：拿不到 id + // 就没有可重试的对象，改成 503 毫无意义。 + return None; + }; + let failure = match destroy_push_registration(app, &previous_url, &subscription_id).await { + Ok(()) => return None, + Err(failure) => failure, + }; + let _ = app + .state + .release_lock(registration_lock, registration_owner) + .await; + Some(failure.response()) +} + /// Registers the callback through JMAP; the verification code is generated by /// Stalwart and is never accepted from SPA configuration. async fn register_push(State(app): State, headers: HeaderMap, body: Bytes) -> Response { @@ -1362,6 +1515,17 @@ async fn register_push(State(app): State, headers: HeaderMap, body: By })) .into_response(); } + // 更换回调地址：先注销 current 指针指向的旧订阅，再建新订阅。 + if let Some(response) = retire_previous_push_registration( + &app, + ®istration_lock, + ®istration_owner, + &request.callback_url, + ) + .await + { + return response; + } let Ok(subscription_id) = app .worker .create_push_subscription(&request.callback_url) @@ -1428,6 +1592,26 @@ async fn register_push(State(app): State, headers: HeaderMap, body: By request_id, ); } + // 写 current 指针：下一次更换地址时靠它找到这份注册去注销。不写会重新 + // 出现重复通知。写失败时**不**销毁已生效的订阅——映射可用、回调已在推送 + // 了，销毁会打断一个正常工作的注册；返回 503 让调用方重试（重试命中 + // 幂等分支返回 200）。 + if app + .state + .set_push_registration_current(&request.callback_url) + .await + .is_err() + { + let _ = app + .state + .release_lock(®istration_lock, ®istration_owner) + .await; + return error_response( + StatusCode::SERVICE_UNAVAILABLE, + "push_state_unavailable", + true, + ); + } let _ = app .state .release_lock(®istration_lock, ®istration_owner) @@ -1518,58 +1702,13 @@ async fn disable_push(State(app): State, headers: HeaderMap, body: Byt else { return error_response(StatusCode::NOT_FOUND, "push_subscription_not_found", false); }; - if app - .worker - .destroy_push_subscription(&subscription_id) - .await - .is_err() + if let Err(failure) = + destroy_push_registration(&app, &request.callback_url, &subscription_id).await { - let request_id = cleanup_push_orphan(&app, &subscription_id).await; - return error_response_with_id( - StatusCode::SERVICE_UNAVAILABLE, - "push_destroy_failed", - request_id, - ); - } - if app - .state - .set_push_subscription_status(&subscription_id, "disabled", ttl::PUSH_DISABLED_SECONDS) - .await - .is_err() - { - return error_response( - StatusCode::SERVICE_UNAVAILABLE, - "push_state_unavailable", - true, - ); - } - // Drop the verification digest as well: otherwise a push arriving within the - // residual TTL of the last verify call still passes - // `push_subscription_verified` and gets enqueued after disable. - if app - .state - .forget_push_subscription(&subscription_id) - .await - .is_err() - { - return error_response( - StatusCode::SERVICE_UNAVAILABLE, - "push_state_unavailable", - true, - ); - } - if app - .state - .remove_push_subscription_for_callback(&request.callback_url) - .await - .is_err() - { - return error_response( - StatusCode::SERVICE_UNAVAILABLE, - "push_state_unavailable", - true, - ); + return failure.response(); } + // 注销掉的是 current 地址时清掉指针，否则它会指向已 disabled 的订阅。 + clear_push_registration_current_if(&app, &request.callback_url).await; axum::Json(serde_json::json!({"disabled": true})).into_response() } @@ -2999,7 +3138,51 @@ mod tests { } } + /// 记录调用次数的 Worker：判断"未重建"与"旧订阅已销毁"需要调用证据， + /// 光看 Redis 状态无法区分。 + struct RecordingPushMockWorker { + /// 按调用次序返回的 subscription id；耗尽后复用最后一个。 + create_ids: Vec<&'static str>, + destroy_ok: bool, + create_calls: Arc, + destroyed: Arc>>, + } + + #[async_trait::async_trait] + impl WorkerHandler for RecordingPushMockWorker { + async fn process(&self, _stream: &str, _payload: &str) -> Result<(), ()> { + Err(()) + } + + async fn create_push_subscription(&self, _callback_url: &str) -> Result { + let call = self.create_calls.fetch_add(1, Ordering::SeqCst); + let Some(id) = self + .create_ids + .get(call.min(self.create_ids.len().saturating_sub(1))) + else { + return Err(()); + }; + Ok((*id).to_string()) + } + + async fn destroy_push_subscription(&self, subscription_id: &str) -> Result<(), ()> { + self.destroyed + .lock() + .unwrap() + .push(subscription_id.to_owned()); + if self.destroy_ok { + Ok(()) + } else { + Err(()) + } + } + } + fn push_router(worker: PushMockWorker) -> (Router, MemoryState) { + push_router_with_handler(Arc::new(worker)) + } + + fn push_router_with_handler(handler: Arc) -> (Router, MemoryState) { let state = MemoryState::enabled_for_tests(); let router = router_with_worker_state( AuthSecrets { @@ -3009,7 +3192,7 @@ mod tests { SecretString::new("worker-secret".into()), state.clone(), HashSet::new(), - Arc::new(worker), + handler, ); (router, state) } @@ -3155,6 +3338,247 @@ mod tests { assert_eq!(second_json["idempotent"], true); } + #[tokio::test] + async fn push_register_is_idempotent_without_recreating_the_subscription() { + let create_calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let destroyed = Arc::new(std::sync::Mutex::new(Vec::::new())); + let (router, state) = push_router_with_handler(Arc::new(RecordingPushMockWorker { + create_ids: vec!["sub-abc"], + destroy_ok: true, + create_calls: create_calls.clone(), + destroyed: destroyed.clone(), + })); + let url = "https://example.com/push"; + let body = format!(r#"{{"callback_url":"{url}"}}"#); + for call in 0..3 { + let response = router + .clone() + .oneshot(push_req("/api/push/register", &body, Some("worker-secret"))) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let json = json_body(response).await; + assert_eq!(json["push_subscription_id"], "sub-abc"); + // 首次注册没有 `idempotent` 字段；再次注册同一地址必须报 `true`。 + assert_eq!( + json["idempotent"], + if call == 0 { + serde_json::Value::Null + } else { + serde_json::Value::Bool(true) + } + ); + } + // 幂等的前提是"不重建"：三次注册只建了一次订阅，也没销毁过任何东西。 + assert_eq!(create_calls.load(Ordering::SeqCst), 1); + assert!(destroyed.lock().unwrap().is_empty()); + // current 指针随注册建立，下一次更换地址时靠它找到这份注册去注销。 + assert_eq!( + state + .get_push_registration_current() + .await + .unwrap() + .as_deref(), + Some(url) + ); + } + + #[tokio::test] + async fn push_register_destroys_previous_subscription_when_the_url_changes() { + let create_calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let destroyed = Arc::new(std::sync::Mutex::new(Vec::::new())); + let (router, state) = push_router_with_handler(Arc::new(RecordingPushMockWorker { + create_ids: vec!["sub-old", "sub-new"], + destroy_ok: true, + create_calls: create_calls.clone(), + destroyed: destroyed.clone(), + })); + let old_url = "https://example.com/push/old"; + let new_url = "https://example.com/push/new"; + let old_body = format!(r#"{{"callback_url":"{old_url}"}}"#); + let first = router + .clone() + .oneshot(push_req( + "/api/push/register", + &old_body, + Some("worker-secret"), + )) + .await + .unwrap(); + assert_eq!(first.status(), StatusCode::OK); + assert_eq!( + state + .get_push_subscription_for_callback(old_url) + .await + .unwrap() + .as_deref(), + Some("sub-old") + ); + assert_eq!( + state + .get_push_registration_current() + .await + .unwrap() + .as_deref(), + Some(old_url) + ); + + let new_body = format!(r#"{{"callback_url":"{new_url}"}}"#); + let second = router + .clone() + .oneshot(push_req( + "/api/push/register", + &new_body, + Some("worker-secret"), + )) + .await + .unwrap(); + assert_eq!(second.status(), StatusCode::OK); + let second_json = json_body(second).await; + assert_eq!(second_json["push_subscription_id"], "sub-new"); + assert_ne!( + second_json.get("idempotent"), + Some(&serde_json::Value::Bool(true)) + ); + + // 旧订阅被销毁——否则 Stalwart 会对同一邮件事件向两个地址各推一次。 + assert_eq!( + destroyed.lock().unwrap().as_slice(), + &[String::from("sub-old")] + ); + assert_eq!(create_calls.load(Ordering::SeqCst), 2); + assert!(state + .get_push_subscription_for_callback(old_url) + .await + .unwrap() + .is_none()); + assert_eq!( + state + .get_push_subscription_for_callback(new_url) + .await + .unwrap() + .as_deref(), + Some("sub-new") + ); + assert_eq!( + state + .get_push_registration_current() + .await + .unwrap() + .as_deref(), + Some(new_url) + ); + } + + #[tokio::test] + async fn push_register_aborts_without_creating_when_the_old_destroy_fails() { + let create_calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let destroyed = Arc::new(std::sync::Mutex::new(Vec::::new())); + let (router, state) = push_router_with_handler(Arc::new(RecordingPushMockWorker { + create_ids: vec!["sub-old", "sub-new"], + destroy_ok: false, + create_calls: create_calls.clone(), + destroyed: destroyed.clone(), + })); + let old_url = "https://example.com/push/old"; + let new_url = "https://example.com/push/new"; + let first = router + .clone() + .oneshot(push_req( + "/api/push/register", + &format!(r#"{{"callback_url":"{old_url}"}}"#), + Some("worker-secret"), + )) + .await + .unwrap(); + assert_eq!(first.status(), StatusCode::OK); + + let second = router + .clone() + .oneshot(push_req( + "/api/push/register", + &format!(r#"{{"callback_url":"{new_url}"}}"#), + Some("worker-secret"), + )) + .await + .unwrap(); + assert_eq!(second.status(), StatusCode::SERVICE_UNAVAILABLE); + assert_eq!( + second + .headers() + .get("retry-after") + .and_then(|v| v.to_str().ok()), + Some("30") + ); + let second_json = json_body(second).await; + assert_eq!(second_json["error"], "push_destroy_failed"); + assert!(second_json["request_id"].is_string()); + + // 旧订阅销毁失败时**不得**创建新订阅：静默继续正是重复通知的形态。 + assert_eq!(create_calls.load(Ordering::SeqCst), 1); + assert!(state + .get_push_subscription_for_callback(new_url) + .await + .unwrap() + .is_none()); + // 旧登记原样保留，指针也未移动，失败可重试。 + assert_eq!( + state + .get_push_subscription_for_callback(old_url) + .await + .unwrap() + .as_deref(), + Some("sub-old") + ); + assert_eq!( + state + .get_push_registration_current() + .await + .unwrap() + .as_deref(), + Some(old_url) + ); + // 销毁被尝试了两次（首次 + `cleanup_push_orphan` 的兜底重试）并两次失败。 + assert_eq!( + destroyed.lock().unwrap().as_slice(), + &[String::from("sub-old"), String::from("sub-old")] + ); + } + + #[tokio::test] + async fn push_register_skips_destroy_when_the_previous_mapping_expired() { + let (router, state) = push_router(PushMockWorker { + create_id: Some("sub-new"), + destroy_ok: true, + }); + // current 指针仍在，但它指向的映射已过期（7 天 TTL）。 + state + .set_push_registration_current("https://example.com/push/expired") + .await + .unwrap(); + let new_url = "https://example.com/push/new"; + let response = router + .oneshot(push_req( + "/api/push/register", + &format!(r#"{{"callback_url":"{new_url}"}}"#), + Some("worker-secret"), + )) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let json = json_body(response).await; + assert_eq!(json["push_subscription_id"], "sub-new"); + assert_ne!(json.get("idempotent"), Some(&serde_json::Value::Bool(true))); + assert_eq!( + state + .get_push_registration_current() + .await + .unwrap() + .as_deref(), + Some(new_url) + ); + } + #[tokio::test] async fn push_register_returns_503_and_releases_lock_when_worker_fails() { let (router, state) = push_router(PushMockWorker { @@ -3240,6 +3664,47 @@ mod tests { .push_subscription_verified("sub-1", "verify-code") .await .unwrap()); + // 注销掉的是 current 地址时指针被清掉，否则它会指向已 disabled 的订阅。 + assert!(state + .get_push_registration_current() + .await + .unwrap() + .is_none()); + } + + #[tokio::test] + async fn push_disable_keeps_current_pointer_when_the_disabled_url_is_not_current() { + let (router, state) = push_router(PushMockWorker { + create_id: Some("sub-1"), + destroy_ok: true, + }); + let url = "https://example.com/push/other"; + state + .remember_push_subscription_for_callback(url, "sub-1") + .await + .unwrap(); + state + .set_push_registration_current("https://example.com/push/current") + .await + .unwrap(); + let response = router + .oneshot(push_req( + "/api/push/disable", + &format!(r#"{{"callback_url":"{url}"}}"#), + Some("worker-secret"), + )) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + // 注销的不是 current 地址：指针必须保持不动。 + assert_eq!( + state + .get_push_registration_current() + .await + .unwrap() + .as_deref(), + Some("https://example.com/push/current") + ); } #[tokio::test] @@ -3253,6 +3718,7 @@ mod tests { .remember_push_subscription_for_callback(url, "sub-1") .await .unwrap(); + state.set_push_registration_current(url).await.unwrap(); let body = format!(r#"{{"callback_url":"{url}"}}"#); let response = router .oneshot(push_req("/api/push/disable", &body, Some("worker-secret"))) @@ -3271,5 +3737,14 @@ mod tests { .as_deref(), Some("sub-1") ); + // 同样保留 current 指针：订阅仍存活，它仍然指向生效的地址。 + assert_eq!( + state + .get_push_registration_current() + .await + .unwrap() + .as_deref(), + Some(url) + ); } } diff --git a/src/state.rs b/src/state.rs index 70b9cea..d8fa6bf 100644 --- a/src/state.rs +++ b/src/state.rs @@ -305,6 +305,21 @@ pub trait ReliableState: Send + Sync { &self, callback_url: &str, ) -> Result<(), StateError>; + + /// 读取当前激活的 push 回调地址（current 指针），无注册时为 None。 + /// + /// per-URL 映射是分散的、无法枚举出"现在生效的是哪一个"；这个指针让 + /// /api/push/register 在更换地址时能找到旧订阅去注销，否则 Stalwart 会 + /// 对同一邮件事件向新旧两个地址各推一次（重复通知）。 + async fn get_push_registration_current(&self) -> Result, StateError>; + + /// 记录当前激活的 push 回调地址。仅在 /api/push/register 真正建好订阅并 + /// 写入 per-URL 映射之后调用；幂等命中（未创建订阅）不写。 + async fn set_push_registration_current(&self, callback_url: &str) -> Result<(), StateError>; + + /// 清除 current 指针。注销掉的是 current 地址时调用，否则指针会指向 + /// 一个已 disabled 的订阅。 + async fn clear_push_registration_current(&self) -> Result<(), StateError>; } /// Bound on a single Redis connect attempt, for both the initial connection @@ -912,8 +927,45 @@ impl ReliableState for RedisState { .await?; Ok(()) } + + async fn get_push_registration_current(&self) -> Result, StateError> { + let mut connection = self.connection.clone(); + let reply: Option = redis::cmd("GET") + .arg(PUSH_REGISTRATION_CURRENT_KEY) + .query_async(&mut connection) + .await?; + Ok(reply) + } + + async fn set_push_registration_current(&self, callback_url: &str) -> Result<(), StateError> { + let mut connection = self.connection.clone(); + // 与 per-URL 映射同 TTL：指针指向的注册过期后，指针本身也应一起消失。 + redis::cmd("SET") + .arg(PUSH_REGISTRATION_CURRENT_KEY) + .arg(callback_url) + .arg("EX") + .arg(ttl::PUSH_REGISTRATION_SECONDS) + .query_async::<()>(&mut connection) + .await?; + Ok(()) + } + + async fn clear_push_registration_current(&self) -> Result<(), StateError> { + let mut connection = self.connection.clone(); + redis::cmd("DEL") + .arg(PUSH_REGISTRATION_CURRENT_KEY) + .query_async::<()>(&mut connection) + .await?; + Ok(()) + } } +/// current 指针的键：当前激活的 push 回调地址原文。 +/// +/// 与 per-URL 映射共用前缀，但字面量 `current` 含非十六进制字符，永远不会 +/// 与 `session_digest` 的十六进制摘要相撞。 +const PUSH_REGISTRATION_CURRENT_KEY: &str = "push:registration:current"; + fn push_registration_key(callback_url: &str) -> String { format!( "push:registration:{}", @@ -999,6 +1051,8 @@ struct MemoryInner { dedup: HashSet, locks: HashMap, push_registrations: HashMap, + /// 当前激活的 push 回调地址（current 指针） + push_registration_current: Option, streams: HashMap>, retries: HashMap<(String, String), u32>, consent: HashMap, @@ -1366,6 +1420,31 @@ impl ReliableState for MemoryState { Ok(()) } + async fn get_push_registration_current(&self) -> Result, StateError> { + Ok(self + .inner + .lock() + .map_err(|_| StateError::Poisoned)? + .push_registration_current + .clone()) + } + + async fn set_push_registration_current(&self, callback_url: &str) -> Result<(), StateError> { + self.inner + .lock() + .map_err(|_| StateError::Poisoned)? + .push_registration_current = Some(callback_url.to_owned()); + Ok(()) + } + + async fn clear_push_registration_current(&self) -> Result<(), StateError> { + self.inner + .lock() + .map_err(|_| StateError::Poisoned)? + .push_registration_current = None; + Ok(()) + } + async fn set_push_subscription_status( &self, subscription_id: &str, @@ -1882,6 +1961,37 @@ mod real_redis_ttl_tests { .await .unwrap(); + // `set_push_registration_current` reuses the registration TTL: the + // pointer is only useful while the registration it points at is still + // valid, so both must expire together. The key name is fixed and this + // may target a shared database, so a populated key means a real + // registration exists -- skip it rather than clobber it. + let previous = state.get_push_registration_current().await.unwrap(); + if previous.is_some() { + eprintln!("skipped pointer TTL check: push:registration:current is already populated"); + return; + } + state + .set_push_registration_current("ttlcheck://example.invalid/push") + .await + .unwrap(); + assert_eq!( + state + .get_push_registration_current() + .await + .unwrap() + .as_deref(), + Some("ttlcheck://example.invalid/push") + ); + let pttl: i64 = redis::cmd("PTTL") + .arg(PUSH_REGISTRATION_CURRENT_KEY) + .query_async(&mut conn) + .await + .unwrap(); + assert_within(pttl, ttl::PUSH_REGISTRATION_SECONDS, 2_000); + assert!(state.clear_push_registration_current().await.is_ok()); + assert_eq!(state.get_push_registration_current().await.unwrap(), None); + // The retry counter's 86400s lives in the Lua script, not at a call // site, so it is asserted through `retry_or_dlq` rather than read back // from a constant. max_attempts=99 keeps n below the cap, so the script