feat(push): retire the previous subscription when the callback URL changes

Registering with a new callback_url previously left the old Stalwart
subscription live alongside the new one, so Stalwart pushed the same mail
event to both origins and the user got duplicate notifications. The
per-URL mapping had no record of which subscription was current.

Add a push:registration:current pointer written alongside the per-URL
mapping. register_push now retires the previous subscription under the
existing registration lock before creating the new one: destroy first,
then delete the mapping, then mark the status disabled. A destroy failure
returns a retryable 503 (push_destroy_failed) rather than proceeding
silently - silently skipping is exactly the shape of the original bug.
The idempotency short-circuit still runs first, so re-registering the
same URL never touches the retire path. disable_push clears the pointer
only when the disabled URL is the current one.

Extracts destroy_push_registration, shared by disable_push and the new
retire path. Telegram needs no change: setWebhook replaces the previous
webhook atomically.

Adds 5 tests with a recording mock that captures the create/destroy
call order: idempotent no-recreate, retire-on-change, abort-without-
create when the old destroy fails, skip-destroy when the previous
mapping expired, and pointer untouched when disabling a non-current URL.

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<Response> {
+    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<AppState>, headers: HeaderMap, body: Bytes) -> Response {
@@ -1362,6 +1515,17 @@ async fn register_push(State(app): State<AppState>, headers: HeaderMap, body: By
         }))
         .into_response();
     }
+    // 更换回调地址：先注销 current 指针指向的旧订阅，再建新订阅。
+    if let Some(response) = retire_previous_push_registration(
+        &app,
+        &registration_lock,
+        &registration_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<AppState>, 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(&registration_lock, &registration_owner)
+            .await;
+        return error_response(
+            StatusCode::SERVICE_UNAVAILABLE,
+            "push_state_unavailable",
+            true,
+        );
+    }
     let _ = app
         .state
         .release_lock(&registration_lock, &registration_owner)
@@ -1518,58 +1702,13 @@ async fn disable_push(State(app): State<AppState>, 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<std::sync::atomic::AtomicUsize>,
+        destroyed: Arc<std::sync::Mutex<Vec<String>>>,
+    }
+
+    #[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<String, ()> {
+            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<dyn WorkerHandler>) -> (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::<String>::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::<String>::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::<String>::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)
+        );
     }
 }

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<Option<String>, 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<Option<String>, StateError> {
+        let mut connection = self.connection.clone();
+        let reply: Option<String> = 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<String>,
     locks: HashMap<String, String>,
     push_registrations: HashMap<String, String>,
+    /// 当前激活的 push 回调地址（current 指针）
+    push_registration_current: Option<String>,
     streams: HashMap<String, Vec<(String, String)>>,
     retries: HashMap<(String, String), u32>,
     consent: HashMap<i64, i64>,
@@ -1366,6 +1420,31 @@ impl ReliableState for MemoryState {
         Ok(())
     }
 
+    async fn get_push_registration_current(&self) -> Result<Option<String>, 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