worker.rs (7197B)
1 /* 2 This file is part of TALER 3 Copyright (C) 2026 Taler Systems SA 4 5 TALER is free software; you can redistribute it and/or modify it under the 6 terms of the GNU Affero General Public License as published by the Free Software 7 Foundation; either version 3, or (at your option) any later version. 8 9 TALER is distributed in the hope that it will be useful, but WITHOUT ANY 10 WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR 11 A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details. 12 13 You should have received a copy of the GNU Affero General Public License along with 14 TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/> 15 */ 16 17 use std::time::Duration; 18 19 use jiff::Timestamp; 20 use sqlx::PgPool; 21 use taler_common::{ExpoBackoffDecorr, config::Config}; 22 use tracing::{error, info, trace, warn}; 23 24 use crate::{ 25 apns::{ApnsError, Client, Reason}, 26 config::WorkerCfg, 27 db, 28 }; 29 30 #[derive(Debug, thiserror::Error)] 31 pub enum WorkerError { 32 #[error(transparent)] 33 Db(#[from] sqlx::Error), 34 #[error(transparent)] 35 Apns(#[from] ApnsError), 36 } 37 38 const NB_RETRY: usize = 5; 39 40 async fn wakeup(pool: &PgPool, client: &mut Client) -> Result<(), WorkerError> { 41 let tokens = db::all_registrations(pool).await?; 42 info!(target: "worker", "send notification to {} devices", tokens.len()); 43 44 // Wait at least 15 minutes before retrying API calls as per: https://developer.apple.com/documentation/usernotifications/sending-notification-requests-to-apns#Follow-best-practices-while-sending-push-notifications-with-APNs 45 // After 15 minutes, you can retry JSON payloads that receive response status codes that begin with 5XX. 46 // While retrying, you may employ a back-off technique. Most notifications with the status code 4XX can 47 // be retried after you fix the error noted in the reason field. Don’t retry notification responses with 48 // the error code BadDeviceToken, DeviceTokenNotForTopic, Forbidden, ExpiredToken, Unregistered, or PayloadTooLarge. 49 // You can retry with a delay, if you get the error code TooManyRequests. 50 let mut jitter = ExpoBackoffDecorr::new(Duration::from_mins(15), Duration::from_hours(1), 2.5); 51 52 // TODO paginate tokens 53 // TODO send in batches 54 55 'devices: for token in tokens { 56 trace!(target: "worker", "send background notification to {token}"); 57 let mut attempt = 0; 58 while let Err(e) = client.send(&token).await { 59 trace!(target: "worker", "notification to {token} failed {e}"); 60 if let ApnsError::Err { reason, timestamp } = e { 61 match reason { 62 // Fatal error 63 Reason::BadCollapseId 64 | Reason::BadExpirationDate 65 | Reason::BadMessageId 66 | Reason::BadPriority 67 | Reason::BadTopic 68 | Reason::DuplicateHeaders 69 | Reason::InvalidPushType 70 | Reason::MissingDeviceToken 71 | Reason::MissingTopic 72 | Reason::PayloadEmpty 73 | Reason::BadPath 74 | Reason::MethodNotAllowed => { 75 error!(target: "worker", "fatal error the service is broken: {e}"); 76 std::process::exit(9); 77 } 78 // Config error 79 Reason::TopicDisallowed 80 | Reason::BadCertificate 81 | Reason::BadCertificateEnvironment 82 | Reason::Forbidden 83 | Reason::InvalidProviderToken 84 | Reason::MissingProviderToken 85 | Reason::UnrelatedKeyIdInToken 86 | Reason::BadEnvironmentKeyIdInToken 87 | Reason::PayloadTooLarge => { 88 error!(target: "worker", "config error, check the configuration: {e}"); 89 std::process::exit(6); 90 } 91 // The client already retried once with a fresh provider token. 92 Reason::ExpiredProviderToken => return Err(e.into()), 93 // Unregister 94 Reason::BadDeviceToken 95 | Reason::DeviceTokenNotForTopic 96 | Reason::ExpiredToken 97 | Reason::Unregistered => { 98 db::unregister( 99 pool, 100 &token, 101 ×tamp 102 .and_then(|s| i64::try_from(s).ok()) 103 .and_then(|s| Timestamp::from_millisecond(s).ok()) 104 .unwrap_or_else(Timestamp::now), 105 ) 106 .await?; 107 continue 'devices; 108 } 109 // Retry this device on the next scheduled pass. 110 Reason::TooManyRequests => { 111 warn!(target: "worker", "notification to {token} rate-limited, deferring to the next pass"); 112 continue 'devices; 113 } 114 // Wait before retry 115 Reason::IdleTimeout 116 | Reason::TooManyProviderTokenUpdates 117 | Reason::InternalServerError 118 | Reason::ServiceUnavailable 119 | Reason::Shutdown => {} 120 } 121 } 122 123 attempt += 1; 124 125 if attempt >= NB_RETRY { 126 error!(target: "worker", "notification to {token} failed more than {NB_RETRY} times, stopping worker after back-off"); 127 tokio::time::sleep(jitter.backoff()).await; 128 return Err(WorkerError::Apns(e)); 129 } else { 130 warn!(target: "worker", "notification to {token} failed {e}, retrying after back-off"); 131 tokio::time::sleep(jitter.backoff()).await; 132 } 133 } 134 info!(target: "trace", "notification to {token} succeeded"); 135 } 136 137 Ok(()) 138 } 139 140 pub async fn run(cfg: &Config, pool: &PgPool, transient: bool) -> anyhow::Result<()> { 141 let cfg = WorkerCfg::parse(cfg)?; 142 let mut client = Client::new(&cfg.apns)?; 143 144 if transient { 145 wakeup(pool, &mut client).await?; 146 return Ok(()); 147 } 148 149 info!(target: "worker", "running at initialisation"); 150 let mut jitter = ExpoBackoffDecorr::default(); 151 loop { 152 match wakeup(pool, &mut client).await { 153 Ok(()) => jitter.reset(), 154 Err(WorkerError::Db(err)) if !matches!(err, sqlx::Error::PoolClosed) => { 155 let delay = jitter.backoff(); 156 error!(target: "worker", ?delay, "database operation failed: {err}"); 157 tokio::time::sleep(delay).await; 158 continue; 159 } 160 // APNs errors reach here only after their bounded operation retries. 161 Err(err) => return Err(err.into()), 162 } 163 164 // TODO take sending time into account 165 tokio::time::sleep(cfg.frequency).await; 166 info!(target: "worker", "running at frequency"); 167 } 168 }