From bf6ac57c6d5e67d8e3cf8b9660857b6fb9b4938e Mon Sep 17 00:00:00 2001 From: Hampus Date: Sun, 18 Oct 2026 02:00:27 +0200 Subject: [PATCH] fix(push): read badge counts only for recipients with a device (#3521) --- fluxer_push/src/delivery.rs | 127 +++++++++++++++++++++++++++++++----- 1 file changed, 112 insertions(+), 15 deletions(-) diff --git a/fluxer_push/src/delivery.rs b/fluxer_push/src/delivery.rs index 8fdcf886b..833954be2 100644 --- a/fluxer_push/src/delivery.rs +++ b/fluxer_push/src/delivery.rs @@ -9,7 +9,7 @@ use crate::metrics::{JobKind, JobRejection, Provider, elapsed_ms}; use crate::payload; use crate::providers::{self, SendOutcome}; use crate::retry::{self, RETRY_DEADLINE}; -use crate::rpc::RpcError; +use crate::rpc::{RpcClient, RpcError}; use crate::server::AppState; use crate::subscription::{Platform, Subscription}; use crate::unix_millis; @@ -353,13 +353,7 @@ async fn run_message_job( ) -> anyhow::Result<()> { let started_ms = now_ms(); let deadline = Instant::now() + RETRY_DEADLINE; - let (badges, subscriptions) = lookup(deadline, || async { - tokio::try_join!( - state.rpc.badge_counts(&job.user_ids), - state.rpc.push_subscriptions(&job.user_ids) - ) - }) - .await?; + let (badges, subscriptions) = lookup_recipients(&state.rpc, &job.user_ids, deadline).await?; state.metrics.record_recipients(job.user_ids.len() as u64); let envelopes = job @@ -410,13 +404,7 @@ async fn run_clear_job(state: &AppState, sends: &Semaphore, job: ClearJob) -> an let started_ms = now_ms(); let deadline = Instant::now() + RETRY_DEADLINE; let user_ids = std::slice::from_ref(&job.user_id); - let (badges, subscriptions) = lookup(deadline, || async { - tokio::try_join!( - state.rpc.badge_counts(user_ids), - state.rpc.push_subscriptions(user_ids) - ) - }) - .await?; + let (badges, subscriptions) = lookup_recipients(&state.rpc, user_ids, deadline).await?; state.metrics.record_recipients(1); let envelope = payload::web_push_clear(&job, badge_of(&badges, &job.user_id)); @@ -503,6 +491,24 @@ async fn run_ring_job(state: &AppState, sends: &Semaphore, job: RingJob) -> anyh Ok(()) } +async fn lookup_recipients( + rpc: &RpcClient, + user_ids: &[String], + deadline: Instant, +) -> Result<(HashMap, HashMap>), RpcError> { + let subscriptions = lookup(deadline, || rpc.push_subscriptions(user_ids)).await?; + let subscribed_ids: Vec = user_ids + .iter() + .filter(|user_id| subscribed(&subscriptions, user_id)) + .cloned() + .collect(); + if subscribed_ids.is_empty() { + return Ok((HashMap::new(), subscriptions)); + } + let badges = lookup(deadline, || rpc.badge_counts(&subscribed_ids)).await?; + Ok((badges, subscriptions)) +} + async fn lookup(deadline: Instant, mut call: F) -> Result where F: FnMut() -> Fut, @@ -708,4 +714,95 @@ mod tests { assert!(Audience::Standard.admits(&subscription("ios_apns", false))); assert!(Audience::Standard.admits(&subscription("ios_apns", true))); } + + async fn fake_api( + subscribed: &'static [&'static str], + ) -> (RpcClient, Arc>>) { + let calls = Arc::new(std::sync::Mutex::new(Vec::new())); + let recorded = Arc::clone(&calls); + let app = axum::Router::new().route( + "/", + axum::routing::post(move |body: axum::body::Bytes| { + let recorded = Arc::clone(&recorded); + async move { + let request: Value = serde_json::from_slice(&body).unwrap(); + recorded.lock().unwrap().push(request.clone()); + let user_ids = request["user_ids"].as_array().cloned().unwrap_or_default(); + let data = match request["type"].as_str().unwrap() { + "get_push_subscriptions" => serde_json::json!( + subscribed + .iter() + .map(|user_id| { + ( + (*user_id).to_owned(), + serde_json::json!([{ + "subscription_id": "sub", + "endpoint": "token", + "p256dh_key": null, + "auth_key": null, + "platform": "android_fcm", + "app_id": null, + "provider_environment": null + }]), + ) + }) + .collect::>() + ), + "get_badge_counts" => serde_json::json!({ + "badge_counts": user_ids + .iter() + .map(|user_id| (user_id.as_str().unwrap().to_owned(), serde_json::json!(3))) + .collect::>() + }), + other => panic!("unexpected rpc {other}"), + }; + serde_json::json!({ "data": data }).to_string() + } + }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}/", listener.local_addr().unwrap()); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + let rpc = RpcClient::new( + &crate::config::RpcConfig { + url, + auth_token: fluxer_common::secret::SecretString::new("token".to_owned()), + }, + reqwest::Client::new(), + Arc::new(crate::metrics::Metrics::new()), + ); + (rpc, calls) + } + + fn badge_requests(calls: &std::sync::Mutex>) -> Vec { + calls + .lock() + .unwrap() + .iter() + .filter(|call| call["type"] == "get_badge_counts") + .map(|call| call["user_ids"].clone()) + .collect() + } + + #[tokio::test] + async fn badge_counts_are_read_only_for_recipients_with_a_device() { + let (rpc, calls) = fake_api(&["2"]).await; + let user_ids: Vec = (1..=500).map(|id| id.to_string()).collect(); + let deadline = Instant::now() + Duration::from_secs(5); + let (badges, subscriptions) = lookup_recipients(&rpc, &user_ids, deadline).await.unwrap(); + assert_eq!(badge_requests(&calls), vec![serde_json::json!(["2"])]); + assert_eq!(badges, HashMap::from([("2".to_owned(), 3)])); + assert!(subscribed(&subscriptions, "2")); + } + + #[tokio::test] + async fn no_badge_counts_are_read_when_nobody_has_a_device() { + let (rpc, calls) = fake_api(&[]).await; + let user_ids: Vec = (1..=500).map(|id| id.to_string()).collect(); + let deadline = Instant::now() + Duration::from_secs(5); + let (badges, _subscriptions) = lookup_recipients(&rpc, &user_ids, deadline).await.unwrap(); + assert!(badge_requests(&calls).is_empty()); + assert!(badges.is_empty()); + assert_eq!(calls.lock().unwrap().len(), 1); + } }