From bc6d1c9e175cf6af6f530ead6ff741e520d689f9 Mon Sep 17 00:00:00 2001 From: anonsaber Date: Fri, 16 Oct 2026 12:26:59 +0800 Subject: [PATCH] fix(push): accept RFC 8620 StateChange pushes and reconcile on them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stalwart's event pushes are StateChange objects (RFC 8620 §7.1.1), which carry only @type plus a changed map of account->type->state-string. They carry no pushSubscriptionId, no accountId and no emailId, so the old handler deserialized every field to None and returned 401 from the subscription_id check. The event-push delivery path was dead from the first deploy; only the /reconcile cron could deliver new mail. Confirmed against Stalwart's own log: the diagnostic entry log showed subscription_id=None account_id=None email_id=None has_verification_code =false for request_ids fe126c05cee9d7493f4619b2727be3a5 and 4869dee4e7360803d8f9136ec1020cd5, matching Stalwart's `Push subscription error ... code = 401` lines at 01:53:24Z and 01:54:24Z. jmap_push now discriminates by content, not by @type, because Stalwart's raw body had never been observed and @type may be absent: - verificationCode present -> PushVerification (RFC 8620 §7.2), the verification branch, unchanged, still returns the code as a JSON body. - changed present -> StateChange: ack 204, then spawn push_reconcile_delta, which pulls the delta via the same /changes path /reconcile uses, under the same lock:reconcile so a push-triggered pull and the cron pull never touch reconcile:state concurrently. If the lock is held it defers: the running pull observes the same change. - neither -> 400 instead of a silent 401, so a malformed push is visible. StateChange cannot supply an emailId, so the dead dedup_exists / inline enqueue branch predicated on it is deleted, along with the four temporary diagnostic logs from ffa53b5. JmapPush keeps only its two live PushVerification fields. The stalwart:jmap queue contract {"accountId","emailId"}, worker.reconcile and the queue keys are untouched. Tests: jmap_push_requires_verification_code replaced by three focused tests -- a body with neither discriminator 400s (was: silent 401), a StateChange body 204s (was: 401 for the route's whole lifetime), and a verification body returns the code as JSON. GATE-P0 108 passed / 0 failed / 4 ignored; GATE-DOCS 6/6 clean. The one remaining `jmap_push request received` entry log is kept permanently for correlation against Stalwart's Push subscription error lines. Co-Authored-By: Claude Code --- src/notify.rs | 307 +++++++++++++++++++++++++++++--------------------- 1 file changed, 180 insertions(+), 127 deletions(-) diff --git a/src/notify.rs b/src/notify.rs index fcf8220..b8df44e 100644 --- a/src/notify.rs +++ b/src/notify.rs @@ -1290,26 +1290,45 @@ async fn jmap_push(State(app): State, body: Bytes) -> Response { if !business_enabled(&app).await { return error_response(StatusCode::SERVICE_UNAVAILABLE, "service_unavailable", true); } - let Ok(push) = serde_json::from_slice::(&body) else { + // Discern the body kind by *content*, not by `@type`: Stalwart's raw body + // has never been observed, so `@type` may be absent and keying on it alone + // would misroute a real event push to the 400 arm below. + // - `verificationCode` present → PushVerification (RFC 8620 §7.2), which + // DOES carry `pushSubscriptionId`. + // - `changed` present → StateChange (RFC 8620 §7.1.1), which + // carries NEITHER `pushSubscriptionId` NOR any emailId — only + // account→type→state-string. The emailIds must be fetched via `/changes`, + // which is what the reconcile path does. + // - otherwise → fail closed with 400. A body that is neither used to + // 401 silently as "unauthorized", hiding a malformed push behind an + // auth error; 400 makes it observable. + let Ok(value): serde_json::Result = serde_json::from_slice(&body) else { return error_response(StatusCode::BAD_REQUEST, "invalid_request", false); }; - // Diagnostic request_id minted once here so every push 401 response and - // branch log line in this handler carries the same correlation id as the - // entry log. The verification_code value is never logged (AGENTS.md §3.2). + // Correlation id minted once per push so the entry log, the branch logs and + // the 401 response all share it — that is what makes a push 401 in this log + // matchable against Stalwart's `Push subscription error` line. The + // verification_code value is never logged (AGENTS.md §3.2). let request_id = reconcile_lock_owner(); tracing::info!( request_id = %request_id, - subscription_id = ?push.subscription_id, - account_id = ?push.account_id, - email_id = ?push.email_id, - has_verification_code = push.verification_code.is_some(), + type_tag = ?value.get("@type").and_then(|v| v.as_str()), + has_verification_code = value.get("verificationCode").is_some(), + has_changed = value.get("changed").is_some(), "jmap_push request received" ); - // subscription_id is required on every request (RFC 8620 §7.2/§7.3). - let Some(subscription) = push.subscription_id.as_deref() else { - return push_401(request_id); - }; - if let Some(code) = push.verification_code.as_deref() { + if value.get("verificationCode").is_some() { + let Ok(push) = serde_json::from_value::(value).map_err(|_| ()) else { + return error_response(StatusCode::BAD_REQUEST, "invalid_request", false); + }; + // subscription_id is required on the verification request (RFC 8620 §7.2). + let Some(subscription) = push.subscription_id.as_deref() else { + return push_401(request_id); + }; + let Some(code) = push.verification_code.as_deref() else { + // Unreachable: the branch was selected on verification_code being Some. + unreachable!("verificationCode branch without a verification code") + }; // Verification request (RFC 8620 §7.2): carries verificationCode. if !matches!( app.state @@ -1364,103 +1383,114 @@ async fn jmap_push(State(app): State, body: Bytes) -> Response { "push verify API call failed; continuing (RFC 8620 §7.2 verification is via callback response)"); } } - // Diagnostic: a verification request rewrites push:subscription:{id}. - // If a concurrent event push 401s, this line tells us a verification - // raced it (key rewritten mid-window). The code value is not logged. - tracing::info!( - request_id = %request_id, - subscription = %subscription, - "jmap_push verification branch wrote key" - ); } - } else { - // Event push (RFC 8620 §7.3): no verificationCode; the subscription - // must have been verified during the initial handshake. - let push_key = format!("push:subscription:{subscription}"); - let exists = matches!(app.state.dedup_exists(&push_key).await, Ok(true)); - // Diagnostic: this is the line that decides the 401. If subscription is - // "b" and exists is false here while a CLI probe shows the key present, - // the key was deleted/rewritten in the window (see the verify-branch - // log). If subscription is not "b", Stalwart sent a different id than - // our stored push:subscription:id pointer — the contradiction's root. + // RFC 8620 §7.2: the verification response MUST be a JSON body + // containing exactly the verificationCode property (not 204). + return axum::Json(serde_json::json!({ + "verificationCode": code + })) + .into_response(); + } + if value.get("changed").is_some() { + // Event push (StateChange): acknowledge so the server stops retrying, + // then pull the delta. The spawned task takes the same `lock:reconcile` + // the /reconcile endpoint holds, so a push-triggered pull and a + // cron-triggered pull never run concurrently (they would race on + // `reconcile:state`). If the lock is already held, do nothing — the + // running reconcile will pick this change up on its own. + let worker = Arc::clone(&app.worker); + let state = Arc::clone(&app.state); tracing::info!( request_id = %request_id, - key = %push_key, - exists, - "jmap_push event dedup_exists check" + accounts_changed = value + .get("changed") + .and_then(|v| v.as_object()) + .map_or(0usize, |m| m.len()), + "jmap_push state change acked; reconciling" ); - if !exists { - return push_401(request_id); - } - } - if push.account_id.is_none() && push.email_id.is_none() { - // RFC 8620 §7.2: verification request response MUST be a JSON body - // containing exactly the verificationCode property (not 204). - if let Some(code) = push.verification_code.as_deref() { - return axum::Json(serde_json::json!({ - "verificationCode": code - })) - .into_response(); - } + tokio::spawn(push_reconcile_delta(worker, state)); return StatusCode::NO_CONTENT.into_response(); } - let (Some(account), Some(email)) = (push.account_id, push.email_id) else { - return error_response(StatusCode::BAD_REQUEST, "invalid_request", false); - }; - let key = format!("dedup:jmap:{account}:{email}"); - match app.state.claim_dedup(&key, ttl::DEDUP_JMAP_SECONDS).await { - Ok(false) => StatusCode::NO_CONTENT.into_response(), - Ok(true) => match app - .state - .enqueue("stalwart:jmap", &String::from_utf8_lossy(&body)) - .await - { - Ok(message_id) => { - // Diagnostic: the event push was accepted and queued for - // delivery. If Telegram still doesn't arrive after this line, - // the failure is downstream (worker.process / Telegram), not - // in the push 401 path. - tracing::info!( - request_id = %request_id, - %message_id, - "jmap_push event enqueued" - ); - // Claim in-flight lease so the cron-triggered worker() drain - // skips this message while we process it immediately. - let inflight_key = format!("delivery:inflight:stalwart:jmap:{message_id}"); - if matches!( - app.state - .claim_dedup(&inflight_key, ttl::DELIVERY_INFLIGHT_SECONDS) - .await, - Ok(true) - ) { - let worker = Arc::clone(&app.worker); - let state = Arc::clone(&app.state); - let payload = String::from_utf8_lossy(&body).into_owned(); - tokio::spawn(async move { - if worker.process("stalwart:jmap", &payload).await.is_ok() { - let committed = - format!("delivery:committed:stalwart:jmap:{message_id}"); - let _ = state - .claim_dedup(&committed, ttl::DELIVERY_COMMITTED_SECONDS) - .await; - let _ = state - .ack("stalwart:jmap", "stalwart-workers", &message_id) - .await; - } - let inflight = format!("delivery:inflight:stalwart:jmap:{message_id}"); - let _ = state.release_dedup(&inflight).await; - }); - } - StatusCode::NO_CONTENT.into_response() - } - Err(_) => { - let _ = app.state.release_dedup(&key).await; - error_response(StatusCode::SERVICE_UNAVAILABLE, "service_unavailable", true) - } - }, - Err(_) => error_response(StatusCode::SERVICE_UNAVAILABLE, "service_unavailable", true), + error_response(StatusCode::BAD_REQUEST, "invalid_request", false) +} + +/// Pull the JMAP delta a StateChange push triggered, on the `/reconcile` lock. +/// +/// A StateChange body carries no emailId (RFC 8620 §7.1.1 gives only +/// account→type→state-string), so the only way to learn which messages changed +/// is the `/changes` call this runs — the same path `/reconcile` uses. It +/// acquires `lock:reconcile` so a push-triggered pull and the cron pull never +/// touch `reconcile:state` concurrently. If the lock is held, this does +/// nothing: the running pull will observe the same state change. +async fn push_reconcile_delta(worker: Arc, state: Arc) { + const PUSH_RECONCILE_INITIAL_CHANGES: usize = 100; + let lock_owner = reconcile_lock_owner(); + if !matches!( + state + .acquire_lock("lock:reconcile", &lock_owner, ttl::RECONCILE_LOCK_SECONDS) + .await, + Ok(true) + ) { + tracing::info!( + "push_reconcile_delta: lock:reconcile busy; deferring to the running reconcile" + ); + return; } + let since = match state.get_reconcile_state().await { + Ok(value) => value, + Err(_) => { + // Same shape as /reconcile: a state read failure means there is no + // safe `since` to reconcile from, so release and give up. The cron + // reconcile covers the change on its next tick. + tracing::warn!("push_reconcile_delta: reconcile state read failed"); + let _ = state.release_lock("lock:reconcile", &lock_owner).await; + return; + } + }; + let heartbeat_state = Arc::clone(&state); + let heartbeat_owner = lock_owner.clone(); + let lease_lost = Arc::new(AtomicBool::new(false)); + let heartbeat_lease_lost = Arc::clone(&lease_lost); + let heartbeat = tokio::spawn(async move { + loop { + tokio::time::sleep(std::time::Duration::from_secs( + ttl::RECONCILE_HEARTBEAT_INTERVAL_SECONDS, + )) + .await; + match heartbeat_state + .renew_lock( + "lock:reconcile", + &heartbeat_owner, + ttl::RECONCILE_HEARTBEAT_SECONDS, + ) + .await + { + Ok(true) => {} + _ => { + heartbeat_lease_lost.store(true, Ordering::Release); + break; + } + } + } + }); + let result = worker + .reconcile(since.as_deref(), PUSH_RECONCILE_INITIAL_CHANGES) + .await; + heartbeat.abort(); + if lease_lost.load(Ordering::Acquire) { + tracing::warn!("push_reconcile_delta: lock lease lost mid-pull; state not advanced"); + } else { + match result { + Ok(new_state) => match state.set_reconcile_state(&new_state).await { + Ok(()) => tracing::info!("push_reconcile_delta: reconcile state advanced"), + Err(_) => tracing::error!("push_reconcile_delta: reconcile state write failed"), + }, + Err(reason) => { + tracing::warn!(reason = %reason, "push_reconcile_delta: reconcile failed") + } + } + } + let _ = state.release_lock("lock:reconcile", &lock_owner).await; } #[derive(serde::Deserialize)] @@ -1933,16 +1963,16 @@ impl TelegramUpdate { self.message.as_ref().map(|m| m.chat.id) } } +/// PushVerification body (RFC 8620 §7.2.3). A StateChange body does not fit +/// this shape at all — it carries neither `pushSubscriptionId` nor +/// `verificationCode` — so `jmap_push` discriminates on the parsed `Value` +/// before reaching for this type. #[derive(serde::Deserialize)] struct JmapPush { #[serde(rename = "pushSubscriptionId")] subscription_id: Option, #[serde(rename = "verificationCode")] verification_code: Option, - #[serde(rename = "accountId")] - account_id: Option, - #[serde(rename = "emailId")] - email_id: Option, } fn header_value_matches(headers: &HeaderMap, name: &str, expected: &str) -> bool { @@ -2538,7 +2568,10 @@ mod tests { } #[tokio::test] - async fn jmap_push_requires_verification_code() { + async fn jmap_push_rejects_bodies_without_verification_or_change() { + // Neither a verificationCode nor a `changed` map: this is not a push + // body at all (RFC 8620 §7). It used to 401 as "unauthorized", which + // hid a malformed push behind an auth error; 400 makes it observable. assert_eq!( request( Request::builder() @@ -2548,20 +2581,10 @@ mod tests { .unwrap(), ) .await, - StatusCode::UNAUTHORIZED - ); - assert_eq!( - request( - Request::builder() - .method("POST") - .uri("/push/jmap") - .header("content-type", "application/json") - .body(Body::from(r#"{"verificationCode":""}"#)) - .unwrap(), - ) - .await, - StatusCode::UNAUTHORIZED + StatusCode::BAD_REQUEST ); + // A verificationCode without a pushSubscriptionId cannot be attributed + // to a subscription, so it is refused before any state is written. assert_eq!( request( Request::builder() @@ -2574,6 +2597,14 @@ mod tests { .await, StatusCode::UNAUTHORIZED ); + } + + #[tokio::test] + async fn jmap_push_acks_state_change() { + // RFC 8620 §7.1.1 StateChange: no pushSubscriptionId, no accountId, no + // emailId — only account→type→state-string. This is what Stalwart + // actually sends for a new message, and it is the body that used to + // 401 for the lifetime of this route. assert_eq!( request( Request::builder() @@ -2581,20 +2612,42 @@ mod tests { .uri("/push/jmap") .header("content-type", "application/json") .body(Body::from( - r#"{"pushSubscriptionId":"push-1","verificationCode":"code","accountId":"a","emailId":"e"}"# + r#"{"@type":"StateChange","changed":{"a3123":{"Email":"d35ecb040aab","EmailDelivery":"428d565f2440"}}}"# )) .unwrap(), ) .await, - // Hybrid payload (verificationCode + accountId + emailId): the - // verification is recorded, then the request falls through to event - // enqueue (RFC 8620 §7.3). remember_push_subscription succeeds - // (PUSH_SUBSCRIPTION_SECONDS == 0 → no expiry), so the response is - // 204 — not 503. The pre-TTL-0 expectation of 503 was stale. StatusCode::NO_CONTENT ); } + #[tokio::test] + async fn jmap_push_verification_returns_the_code() { + // RFC 8620 §7.2: the verification response MUST be a JSON body holding + // the verificationCode property, not an empty 204. + let response = test_router() + .oneshot( + Request::builder() + .method("POST") + .uri("/push/jmap") + .header("content-type", "application/json") + .body(Body::from( + r#"{"@type":"PushVerification","pushSubscriptionId":"push-1","verificationCode":"code"}"# + )) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + assert_eq!( + serde_json::from_slice::(&body).unwrap(), + json!({"verificationCode": "code"}) + ); + } + #[tokio::test] async fn worker_requires_bearer_token() { let status = request(