kasl_server/webhooks/
dispatch.rs

1//! Sending what is queued: in order per destination, retried, and given up on
2//! in writing.
3//!
4//! The loop wakes every few seconds and, for each destination, sends the
5//! oldest thing still in flight - and then the next, while they keep
6//! succeeding. A failure stops that destination until its retry is due, which
7//! is what keeps "resolved" from arriving before "raised": later events wait
8//! behind the one that has not got through. Other destinations are not held
9//! up by it.
10
11use std::{sync::Arc, time::Duration};
12
13use chrono::{DateTime, TimeDelta, Utc};
14use reqwest::{StatusCode, header};
15use serde::Serialize;
16use sqlx::PgPool;
17use uuid::Uuid;
18
19use super::{
20    Destination, Event, Kind, Webhooks,
21    destination::Target,
22    render::{self, Markup},
23    sign,
24};
25use crate::error::ApiError;
26
27/// How often the dispatcher looks for work.
28///
29/// Seconds, not the sweep's minutes: an alert is raised at most every five
30/// minutes, but an acknowledgement or a test is somebody at a screen waiting
31/// to see the channel light up.
32const TICK: Duration = Duration::from_secs(5);
33
34/// How long one request may take before it counts as failed.
35const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
36
37/// The most one destination is sent in one tick. A backlog after an outage
38/// drains over a few ticks rather than holding the loop for minutes.
39const BURST: usize = 50;
40
41/// The waits between attempts. Nine tries across about a day: long enough to
42/// ride out a chat service's bad night, short enough that an alert is not
43/// delivered on Thursday about Monday.
44const RETRY_AFTER: [TimeDelta; 8] = [
45    TimeDelta::seconds(30),
46    TimeDelta::minutes(2),
47    TimeDelta::minutes(10),
48    TimeDelta::minutes(30),
49    TimeDelta::hours(1),
50    TimeDelta::hours(3),
51    TimeDelta::hours(6),
52    TimeDelta::hours(12),
53];
54
55/// The longest a receiver's `Retry-After` is taken at its word.
56const LONGEST_REQUESTED_WAIT: TimeDelta = TimeDelta::hours(1);
57
58/// How much of a refusal's body is kept. Enough for Slack's `no_service` or
59/// Telegram's `chat not found`; not enough to store a page of HTML.
60const ERROR_BODY: usize = 300;
61
62/// What one tick did.
63#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize)]
64pub struct Dispatched {
65    pub delivered: u64,
66    /// Failed, and scheduled to be tried again.
67    pub retrying: u64,
68    /// Failed for good: refused, out of retries, or sent to a destination no
69    /// longer configured.
70    pub abandoned: u64,
71}
72
73/// The HTTP client every delivery goes through.
74///
75/// TLS from rustls with the `ring` provider and the Mozilla roots compiled in,
76/// the same stack the database connection already uses, so a delivery does
77/// not depend on which certificates the host happens to have installed.
78pub fn client() -> reqwest::Client {
79    let roots = rustls::RootCertStore::from_iter(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
80    let tls = rustls::ClientConfig::builder_with_provider(Arc::new(rustls::crypto::ring::default_provider()))
81        .with_safe_default_protocol_versions()
82        .expect("ring supports the default protocol versions")
83        .with_root_certificates(roots)
84        .with_no_client_auth();
85
86    reqwest::Client::builder()
87        .use_preconfigured_tls(tls)
88        .timeout(REQUEST_TIMEOUT)
89        .user_agent(concat!("kasl-server/", env!("CARGO_PKG_VERSION")))
90        // A hook that redirects is a hook that moved, and following it would
91        // send the body somewhere nobody configured.
92        .redirect(reqwest::redirect::Policy::none())
93        .build()
94        .expect("the HTTP client builds from a fixed configuration")
95}
96
97/// One queued delivery, as the dispatcher reads it.
98#[derive(Debug, sqlx::FromRow)]
99struct Pending {
100    id: Uuid,
101    attempts: i32,
102    next_attempt_at: DateTime<Utc>,
103    payload: sqlx::types::Json<Event>,
104}
105
106/// What came of one attempt.
107#[derive(Debug, PartialEq, Eq)]
108enum Outcome {
109    Delivered,
110    /// Worth trying again: the receiver was down, slow, or asked us to wait.
111    Retry {
112        status: Option<u16>,
113        error: String,
114        wait: Option<TimeDelta>,
115    },
116    /// Not worth trying again: the receiver refused this request for a
117    /// reason sending it again will not change.
118    Refused {
119        status: u16,
120        error: String,
121    },
122}
123
124/// Sends everything that is due, once.
125pub async fn dispatch_due(pool: &PgPool, webhooks: &Webhooks, client: &reqwest::Client, now: DateTime<Utc>) -> Result<Dispatched, ApiError> {
126    let mut dispatched = Dispatched::default();
127
128    let waiting: Vec<String> = sqlx::query_scalar("SELECT DISTINCT destination FROM webhook_deliveries WHERE delivered_at IS NULL AND abandoned_at IS NULL")
129        .fetch_all(pool)
130        .await?;
131
132    for name in waiting {
133        let Some(destination) = webhooks.get(&name) else {
134            // Its variable was removed. Given up on in writing rather than
135            // left pending forever, where it would read on the screen as
136            // "still trying".
137            let dropped = sqlx::query(
138                "UPDATE webhook_deliveries SET abandoned_at = $2, last_error = $3
139                 WHERE destination = $1 AND delivered_at IS NULL AND abandoned_at IS NULL",
140            )
141            .bind(&name)
142            .bind(now)
143            .bind(format!("no destination named `{name}` is configured any more"))
144            .execute(pool)
145            .await?;
146            dispatched.abandoned += dropped.rows_affected();
147            continue;
148        };
149
150        for _ in 0..BURST {
151            let head: Option<Pending> = sqlx::query_as(
152                "SELECT id, attempts, next_attempt_at, payload FROM webhook_deliveries
153                 WHERE destination = $1 AND delivered_at IS NULL AND abandoned_at IS NULL
154                 ORDER BY created_at, id
155                 LIMIT 1",
156            )
157            .bind(&name)
158            .fetch_optional(pool)
159            .await?;
160
161            // Nothing left, or the oldest is waiting for its retry - in which
162            // case everything behind it waits too.
163            let Some(head) = head.filter(|head| head.next_attempt_at <= now) else { break };
164
165            let outcome = send(client, destination, &head.payload, now).await;
166            let attempts = head.attempts + 1;
167            match outcome {
168                Outcome::Delivered => {
169                    sqlx::query("UPDATE webhook_deliveries SET delivered_at = $2, attempts = $3, last_status = NULL, last_error = NULL WHERE id = $1")
170                        .bind(head.id)
171                        .bind(now)
172                        .bind(attempts)
173                        .execute(pool)
174                        .await?;
175                    dispatched.delivered += 1;
176                    continue;
177                }
178                Outcome::Retry { status, error, wait } => {
179                    let error = destination.redact(&error);
180                    match RETRY_AFTER.get(head.attempts as usize) {
181                        Some(scheduled) => {
182                            let wait = wait.map_or(*scheduled, |asked| asked.clamp(*scheduled, LONGEST_REQUESTED_WAIT.max(*scheduled)));
183                            sqlx::query("UPDATE webhook_deliveries SET attempts = $2, next_attempt_at = $3, last_status = $4, last_error = $5 WHERE id = $1")
184                                .bind(head.id)
185                                .bind(attempts)
186                                .bind(now + wait)
187                                .bind(status.map(i32::from))
188                                .bind(&error)
189                                .execute(pool)
190                                .await?;
191                            tracing::warn!(destination = %name, attempts, %error, "a webhook delivery failed; it will be tried again");
192                            dispatched.retrying += 1;
193                        }
194                        None => {
195                            abandon(pool, head.id, attempts, now, status, &format!("gave up after {attempts} attempts: {error}")).await?;
196                            tracing::warn!(destination = %name, attempts, %error, "gave up on a webhook delivery");
197                            dispatched.abandoned += 1;
198                        }
199                    }
200                }
201                Outcome::Refused { status, error } => {
202                    let error = destination.redact(&error);
203                    abandon(pool, head.id, attempts, now, Some(status), &error).await?;
204                    tracing::warn!(destination = %name, status, %error, "a webhook receiver refused a delivery; not trying again");
205                    dispatched.abandoned += 1;
206                }
207            }
208            break;
209        }
210    }
211
212    Ok(dispatched)
213}
214
215async fn abandon(pool: &PgPool, id: Uuid, attempts: i32, now: DateTime<Utc>, status: Option<u16>, error: &str) -> Result<(), ApiError> {
216    sqlx::query("UPDATE webhook_deliveries SET abandoned_at = $2, attempts = $3, last_status = $4, last_error = $5 WHERE id = $1")
217        .bind(id)
218        .bind(now)
219        .bind(attempts)
220        .bind(status.map(i32::from))
221        .bind(error)
222        .execute(pool)
223        .await?;
224    Ok(())
225}
226
227/// Makes one attempt.
228async fn send(client: &reqwest::Client, destination: &Destination, event: &Event, now: DateTime<Utc>) -> Outcome {
229    let request = match (&destination.target, destination.kind) {
230        (Target::Hook { url }, Kind::Slack) => client.post(url).json(&render::hook_body(render::text(event, destination, Markup::Slack))),
231        (Target::Hook { url }, _) => client.post(url).json(&render::hook_body(render::text(event, destination, Markup::Markdown))),
232        (Target::Telegram { token, chat, api }, _) => client
233            .post(format!("{api}/bot{token}/sendMessage"))
234            .json(&render::telegram_body(chat, render::text(event, destination, Markup::TelegramHtml))),
235        (Target::Json { url, secret }, _) => {
236            let body = match serde_json::to_vec(event) {
237                Ok(body) => body,
238                Err(error) => {
239                    return Outcome::Refused {
240                        status: 0,
241                        error: format!("the event could not be written as JSON: {error}"),
242                    };
243                }
244            };
245            client
246                .post(url)
247                .header(header::CONTENT_TYPE, "application/json")
248                .header("X-Kasl-Event", event.event.name())
249                .header("X-Kasl-Event-Id", event.id.to_string())
250                .header("X-Kasl-Signature", sign::signature(secret, now.timestamp(), &body))
251                .body(body)
252        }
253    };
254
255    let response = match request.send().await {
256        Ok(response) => response,
257        // `without_url`: reqwest's error names the address it was sending to,
258        // and the address is the credential.
259        Err(error) => {
260            return Outcome::Retry {
261                status: None,
262                error: describe(&error.without_url()),
263                wait: None,
264            };
265        }
266    };
267
268    let status = response.status();
269    if status.is_success() {
270        return Outcome::Delivered;
271    }
272
273    let wait = retry_after(&response);
274    let body: String = response.text().await.unwrap_or_default().chars().take(ERROR_BODY).collect();
275    let error = if body.trim().is_empty() {
276        format!("the receiver answered {status}")
277    } else {
278        format!("the receiver answered {status}: {}", body.trim())
279    };
280
281    classify(status, error, wait)
282}
283
284/// Which failures are worth another try.
285///
286/// A 4xx says the request itself is wrong - a hook that was deleted, a bot
287/// removed from the chat, a token revoked - and the same request will be
288/// wrong in an hour too. The exceptions are the two that say "not now":
289/// 408 and 429. Everything else, including every 5xx, is the receiver's bad
290/// moment and gets retried.
291fn classify(status: StatusCode, error: String, wait: Option<TimeDelta>) -> Outcome {
292    let code = status.as_u16();
293    if status.is_client_error() && code != 408 && code != 429 {
294        Outcome::Refused { status: code, error }
295    } else {
296        Outcome::Retry {
297            status: Some(code),
298            error,
299            wait,
300        }
301    }
302}
303
304/// A `Retry-After` given in seconds. The HTTP-date form is not honoured: no
305/// chat service this sends to uses it, and a wrongly parsed date is a wait of
306/// years.
307fn retry_after(response: &reqwest::Response) -> Option<TimeDelta> {
308    let seconds: i64 = response.headers().get(header::RETRY_AFTER)?.to_str().ok()?.trim().parse().ok()?;
309    (seconds >= 0).then(|| TimeDelta::seconds(seconds))
310}
311
312/// An error and its causes in one line: "error sending request: connection
313/// refused", where the top level alone says only the first half.
314fn describe(error: &reqwest::Error) -> String {
315    let mut text = error.to_string();
316    let mut source = std::error::Error::source(error);
317    while let Some(cause) = source {
318        let cause_text = cause.to_string();
319        if !text.contains(&cause_text) {
320            text.push_str(": ");
321            text.push_str(&cause_text);
322        }
323        source = cause.source();
324    }
325    text
326}
327
328/// Runs the dispatcher for as long as the server runs.
329///
330/// A tick that fails is logged and the loop goes on: the queue is in the
331/// database, so nothing is lost by a tick that did not run, and a database
332/// blip must not leave deliveries frozen until a restart.
333pub fn run_dispatcher(pool: PgPool, webhooks: Arc<Webhooks>) {
334    tokio::spawn(async move {
335        let client = client();
336        let mut ticker = tokio::time::interval(TICK);
337        ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
338        loop {
339            ticker.tick().await;
340            match dispatch_due(&pool, &webhooks, &client, Utc::now()).await {
341                Ok(done) if done != Dispatched::default() => {
342                    tracing::info!(
343                        delivered = done.delivered,
344                        retrying = done.retrying,
345                        abandoned = done.abandoned,
346                        "dispatched webhooks"
347                    );
348                }
349                Ok(_) => {}
350                Err(error) => tracing::warn!(%error, "a webhook dispatch failed; the next one picks up where it left off"),
351            }
352        }
353    });
354}
355
356#[cfg(test)]
357mod tests {
358    use super::*;
359
360    #[test]
361    fn a_refusal_is_final_and_a_bad_moment_is_not() {
362        for code in [400, 401, 403, 404, 410] {
363            assert!(
364                matches!(classify(StatusCode::from_u16(code).unwrap(), String::new(), None), Outcome::Refused { .. }),
365                "{code} will not change by being asked again"
366            );
367        }
368        for code in [408, 429, 500, 502, 503, 504] {
369            assert!(
370                matches!(classify(StatusCode::from_u16(code).unwrap(), String::new(), None), Outcome::Retry { .. }),
371                "{code} is worth another try"
372            );
373        }
374    }
375
376    #[test]
377    fn the_retries_span_about_a_day() {
378        let total: TimeDelta = RETRY_AFTER.iter().copied().sum();
379        assert!(total > TimeDelta::hours(20) && total < TimeDelta::hours(26), "{total}");
380        assert!(RETRY_AFTER.windows(2).all(|pair| pair[0] < pair[1]), "each wait longer than the last");
381    }
382}
kasl_server/webhooks/dispatch.rs

kasl_server/webhooks/
dispatch.rs
