Webhook responses read up to 64 KiB; at most 1000 waiting per hook c98935537c57

/// The most deliveries one hook may have waiting (#220): beyond it, new

/// events for that hook are dropped (and logged) until it catches up.

pub const MAX_WAITING: i64 = 1000;

/// Queue a delivery; `Ok(None)` when the hook already has

/// [`MAX_WAITING`] deliveries waiting, so nothing was queued.

pub async fn queue_delivery(pool: &Pool, hook_id: i64, delivery: &str, event: &str, payload: &str) -> sqlx::Result<Option<i64>> {

    let waiting: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM deliveries WHERE hook_id = ? AND delivered_ms IS NULL AND next_ms < ?")

        .bind(hook_id)

        .bind(i64::MAX / 4)

        .fetch_one(pool)

        .await?;

    if waiting >= MAX_WAITING {

        return Ok(None);

    }

    .map(Some)

        match hdb::queue_delivery(&app.db, h.id, &delivery, event, &payload).await {

            Ok(Some(_)) => {}

            Ok(None) => tracing::warn!("webhook {} has {} deliveries waiting: {event} not queued", h.id, hdb::MAX_WAITING),

            Err(e) => tracing::warn!("queueing delivery: {e}"),

                (Some(status), capped_text(r, RESPONSE_LIMIT).await)

/// How much of a webhook's response is read (#220): enough to show why a

/// delivery failed, never so much that an endpoint can fill memory.

const RESPONSE_LIMIT: usize = 64 * 1024;

/// A response's body as text, read in chunks and stopped at `limit` bytes.

async fn capped_text(mut r: reqwest::Response, limit: usize) -> String {

    let mut body = Vec::new();

    while let Ok(Some(chunk)) = r.chunk().await {

        let room = limit - body.len();

        body.extend_from_slice(&chunk[..chunk.len().min(room)]);

        if body.len() >= limit {

            break;

        }

    }

    String::from_utf8_lossy(&body).into_owned()

}

    #[tokio::test]

    async fn a_response_is_read_only_up_to_the_limit() {

        // #220: an endpoint answering 100 KiB is read to 64 KiB, no further.

        let big = reqwest::Response::from(axum::http::Response::new("x".repeat(100 * 1024)));

        assert_eq!(capped_text(big, RESPONSE_LIMIT).await.len(), RESPONSE_LIMIT);

        let small = reqwest::Response::from(axum::http::Response::new("ok".to_string()));

        assert_eq!(capped_text(small, RESPONSE_LIMIT).await, "ok");

    }

    // #220: a hook with MAX_WAITING deliveries waiting takes no more until it catches up.

    {

        use oxygenhub::db::hooks::{queue_delivery, MAX_WAITING};

        let before: i64 =

            sqlx::query_scalar("SELECT COUNT(*) FROM deliveries WHERE hook_id = ?").bind(hook.id).fetch_one(&app.db).await.unwrap();

        for i in 0..MAX_WAITING {

            sqlx::query(

                "INSERT INTO deliveries (hook_id, delivery, event, payload, created_ms, next_ms) VALUES (?, ?, 'push', '{}', 0, 0)",

            )

            .bind(hook.id)

            .bind(format!("flood-{i}"))

            .execute(&app.db)

            .await

            .unwrap();

        }

        assert_eq!(queue_delivery(&app.db, hook.id, "one-more", "push", "{}").await.unwrap(), None, "the queue is full");

        sqlx::query("DELETE FROM deliveries WHERE delivery LIKE 'flood-%'").execute(&app.db).await.unwrap();

        assert!(queue_delivery(&app.db, hook.id, "now-room", "push", "{}").await.unwrap().is_some());

        sqlx::query("DELETE FROM deliveries WHERE delivery = 'now-room'").execute(&app.db).await.unwrap();

        let after: i64 =

            sqlx::query_scalar("SELECT COUNT(*) FROM deliveries WHERE hook_id = ?").bind(hook.id).fetch_one(&app.db).await.unwrap();

        assert_eq!(before, after);

    }