diff --git a/.env.example b/.env.example index c6e9a11..f10c7cc 100644 --- a/.env.example +++ b/.env.example @@ -37,6 +37,23 @@ WEBHOOK_RETRY_DELAY_MS=5000 # Each retry is bounded independently. Default: 10s. WEBHOOK_TIMEOUT_SECS=10 +# Redrive worker — re-attempts deliveries left pending/failed by a crash or +# an exhausted retry loop, independently of the inline retries above. +# How often (seconds) the worker scans for stuck deliveries. +WEBHOOK_REDRIVE_INTERVAL_SECS=30 +# Maximum redrive HTTP attempts in flight at once. +WEBHOOK_REDRIVE_CONCURRENCY=4 +# Total attempts (inline + redrive) before a delivery is left failed permanently. +WEBHOOK_REDRIVE_MAX_ATTEMPTS=8 +# How long (seconds) a delivery must sit idle since its last attempt before +# the redrive worker will touch it, so it never races a still-in-flight +# inline delivery for the same row. Also the floor under the backoff below. +WEBHOOK_REDRIVE_GRACE_SECS=60 +# Exponential backoff (seconds) applied to redrive attempts once a delivery +# has failed at least once: initial * 2^(attempts-1), capped at max. +WEBHOOK_REDRIVE_BACKOFF_INITIAL_SECS=30 +WEBHOOK_REDRIVE_BACKOFF_MAX_SECS=900 + # CORS — comma-separated list of allowed origins. # Required when STELLAR_NETWORK=public; omit only for local dev (testnet). # Example: CORS_ALLOWED_ORIGINS=https://app.example.com,https://www.example.com diff --git a/Cargo.toml b/Cargo.toml index 0928a88..fd2dfe5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -36,10 +36,10 @@ dotenvy = "0.15" tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } anyhow = "1" +time = { version = "0.3", features = ["parsing"] } [dev-dependencies] axum-test = "15" tokio = { version = "1", features = ["full"] } -time = { version = "0.3", features = ["parsing"] } wiremock = "0.6.5" diff --git a/README.md b/README.md index d25567e..fac9004 100644 --- a/README.md +++ b/README.md @@ -80,7 +80,9 @@ cp .env.example .env | `WEBHOOK_REDRIVE_INTERVAL_SECS` | How often the background redrive worker scans for stuck webhook deliveries (rows left `pending`/`failed` by a process that exited mid-delivery). Its first pass runs immediately on startup, so a restart redrives without waiting a full interval. | `30` | | `WEBHOOK_REDRIVE_CONCURRENCY` | Maximum redrive HTTP attempts in flight at once. | `4` | | `WEBHOOK_REDRIVE_MAX_ATTEMPTS` | Total attempts (inline + redrive) before a delivery is left `failed` permanently. | `8` | -| `WEBHOOK_REDRIVE_GRACE_SECS` | How long (seconds) a delivery must sit idle since its last attempt before the redrive worker will touch it, so it never races a still-in-flight inline delivery for the same row. | `60` | +| `WEBHOOK_REDRIVE_GRACE_SECS` | How long (seconds) a delivery must sit idle since its last attempt before the redrive worker will touch it, so it never races a still-in-flight inline delivery for the same row. Also the floor under the backoff below. | `60` | +| `WEBHOOK_REDRIVE_BACKOFF_INITIAL_SECS` | Starting delay (seconds) of the exponential backoff applied to redrive attempts once a delivery has failed at least once (`initial * 2^(attempts-1)`, capped by `WEBHOOK_REDRIVE_BACKOFF_MAX_SECS`). A row never attempted (crash before its first send) is exempt and gated by `WEBHOOK_REDRIVE_GRACE_SECS` alone. Set to `0` to disable growth. | `30` | +| `WEBHOOK_REDRIVE_BACKOFF_MAX_SECS` | Upper bound (seconds) on the backoff above. Must be `>= WEBHOOK_REDRIVE_BACKOFF_INITIAL_SECS`. | `900` | | `WEBHOOK_ALLOW_PRIVATE_TARGETS` | Bypasses the SSRF guard's loopback/link-local/private/reserved IP check on `webhook_url` (still requires http(s) and a resolvable host). For local development and tests only — never enable in production. | `false` | | `CORS_ALLOWED_ORIGINS` | Comma-separated allowed CORS origins (e.g. `https://app.example.com`). Required on `public` network; omitting on testnet falls back to permissive with a warning. | _(unset — permissive on testnet)_ | | `RATE_LIMIT_REQUESTS_PER_SEC` | Rate limit for `POST /payments` and `POST /merchants` (requests per second per IP, tracked independently per route) | `10` | diff --git a/TODO.md b/TODO.md index fa2f86a..2f07b28 100644 --- a/TODO.md +++ b/TODO.md @@ -38,7 +38,9 @@ - redrive interval (`WEBHOOK_REDRIVE_INTERVAL_SECS`) - concurrency (`WEBHOOK_REDRIVE_CONCURRENCY`) - max attempts (`WEBHOOK_REDRIVE_MAX_ATTEMPTS`) - - backoff initial/max (`WEBHOOK_REDRIVE_GRACE_SECS` — the idle window before a stuck row is retried) + - grace window (`WEBHOOK_REDRIVE_GRACE_SECS` — the idle floor before a stuck row is retried) + - backoff initial/max (`WEBHOOK_REDRIVE_BACKOFF_INITIAL_SECS` / `WEBHOOK_REDRIVE_BACKOFF_MAX_SECS`, + issue #144 — exponential backoff for rows that have already failed at least once) ## Step 7: Tests - [x] Add test for “settle/poller not blocked by slow webhook” diff --git a/src/api/mod.rs b/src/api/mod.rs index b8bedae..47b7c18 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -1,6 +1,6 @@ use crate::{db, AppState}; use axum::{ - extract::{ConnectInfo, Request, State}, + extract::{ConnectInfo, MatchedPath, Request, State}, http::{header, HeaderValue, StatusCode}, middleware::{self, Next}, response::IntoResponse, @@ -12,7 +12,7 @@ use std::collections::HashMap; use std::net::SocketAddr; use std::num::NonZeroU32; use std::sync::{Arc, Mutex}; -use std::time::Duration; +use std::time::{Duration, Instant}; use tower_http::{ cors::CorsLayer, limit::RequestBodyLimitLayer, @@ -100,14 +100,30 @@ pub fn router(state: Arc) -> axum::Router { StatusCode::REQUEST_TIMEOUT, request_timeout, )) + // Outermost so it captures the final status (including a 408 from the + // timeout layer above) and the full request latency (issue #133). + .layer(middleware::from_fn_with_state( + state.clone(), + request_metrics_middleware, + )) .with_state(state) } +/// Authenticates via the `Authorization: Bearer ` header, injecting +/// [`AuthenticatedMerchant`] into request extensions on success. +/// +/// Every outcome is both logged (`source_ip` + `reason`, at a level matched +/// to severity — failures visible by default, success at `debug` to avoid +/// flooding logs) and counted in `AuthMetrics` (issue #139), so +/// credential-stuffing or a misconfigured client shows up in logs/metrics +/// instead of silently returning 401s. async fn auth_middleware( State(state): State>, mut req: Request, next: Next, ) -> axum::response::Response { + let source_ip = client_ip_key(&req); + let raw_key = req .headers() .get(axum::http::header::AUTHORIZATION) @@ -118,6 +134,12 @@ async fn auth_middleware( .map(str::to_string); let Some(key) = raw_key else { + tracing::warn!( + %source_ip, + reason = "missing_key", + "auth denied: missing or malformed Authorization header" + ); + state.auth_metrics.record_failure_missing_key(); return ( StatusCode::UNAUTHORIZED, Json(json!({ "error": "missing or invalid Authorization header", "code": "unauthorized" })), @@ -127,20 +149,34 @@ async fn auth_middleware( match db::find_merchant_by_key(&state.pool, &key).await { Ok(Some(merchant_id)) => { + tracing::debug!(%source_ip, %merchant_id, "auth succeeded"); + state.auth_metrics.record_success(); req.extensions_mut() .insert(AuthenticatedMerchant(merchant_id)); next.run(req).await } - Ok(None) => ( - StatusCode::UNAUTHORIZED, - Json(json!({ "error": "invalid API key", "code": "unauthorized" })), - ) - .into_response(), - Err(_) => ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(json!({ "error": "internal server error", "code": "internal_error" })), - ) - .into_response(), + Ok(None) => { + tracing::warn!( + %source_ip, + reason = "invalid_key", + "auth denied: API key did not match any merchant" + ); + state.auth_metrics.record_failure_invalid_key(); + ( + StatusCode::UNAUTHORIZED, + Json(json!({ "error": "invalid API key", "code": "unauthorized" })), + ) + .into_response() + } + Err(e) => { + tracing::error!(%source_ip, error = %e, "auth errored: merchant key lookup failed"); + state.auth_metrics.record_failure_internal_error(); + ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({ "error": "internal server error", "code": "internal_error" })), + ) + .into_response() + } } } @@ -195,6 +231,39 @@ async fn provision_merchant( )) } +/// Records every request's `(method, matched route, status)` and its +/// end-to-end latency into `AppState::request_metrics`, exposed via +/// `GET /metrics` (issue #133). +/// +/// Uses the route *template* from [`MatchedPath`] rather than the raw path so +/// `/payments/:id` style routes don't blow up the label cardinality with one +/// series per id. Requests that matched no route (e.g. a 404 on an arbitrary +/// path) are counted under the fixed `"unmatched"` route for the same reason. +async fn request_metrics_middleware( + State(state): State>, + req: Request, + next: Next, +) -> axum::response::Response { + let method = req.method().to_string(); + let route = req + .extensions() + .get::() + .map(|p| p.as_str().to_string()) + .unwrap_or_else(|| "unmatched".to_string()); + + let start = Instant::now(); + let resp = next.run(req).await; + + state + .request_metrics + .record(&method, &route, resp.status().as_u16()); + state + .request_metrics + .record_latency_ms(start.elapsed().as_millis() as u64); + + resp +} + async fn rate_limit_middleware( State(rate_limit): State, req: Request, @@ -365,7 +434,12 @@ async fn check_horizon_ready(state: &Arc) -> Result<(), String> { /// `GET /metrics` — Prometheus-compatible plain-text metrics snapshot. async fn metrics_handler(State(state): State>) -> impl IntoResponse { - let body = crate::metrics::render(&state.webhook_metrics); + let body = crate::metrics::render( + &state.webhook_metrics, + &state.auth_metrics, + &state.request_metrics, + &state.settlement_metrics, + ); ( StatusCode::OK, [( diff --git a/src/config.rs b/src/config.rs index 7ecd5b9..8da0f1c 100644 --- a/src/config.rs +++ b/src/config.rs @@ -114,8 +114,22 @@ pub struct Config { /// exceed the worst-case inline delivery time /// (`webhook_retry_attempts * (webhook_timeout_secs + webhook_retry_delay_ms)`) /// so the worker never races a `dispatch()` call that is still in flight - /// for the same row. + /// for the same row. Acts as a hard floor under the exponential backoff + /// below — a row is never touched sooner than this, even on its very + /// first redrive attempt. pub webhook_redrive_grace_secs: i64, + /// Starting delay (seconds) of the exponential backoff applied to a + /// delivery's *redrive* attempts after it has failed at least once + /// (`initial * 2^(attempts-1)`, capped by `webhook_redrive_backoff_max_secs`). + /// A row that has never been attempted (`attempts == 0`, left behind by a + /// crash between insert and its first send) is exempt from this backoff + /// and is only gated by `webhook_redrive_grace_secs`. Set to `0` to + /// disable growth and redrive purely on the fixed grace window. + pub webhook_redrive_backoff_initial_secs: i64, + /// Upper bound (seconds) on the exponential backoff above, so a delivery + /// that has failed many times still gets retried at a bounded cadence + /// rather than being pushed further and further out. + pub webhook_redrive_backoff_max_secs: i64, pub poll_interval_secs: u64, /// How long a payment intent stays `pending` before the expiry sweeper /// transitions it to `expired`. Counted from the intent's `created_at`. @@ -211,6 +225,11 @@ impl Config { webhook_redrive_concurrency: parse_env("WEBHOOK_REDRIVE_CONCURRENCY", 4)?, webhook_redrive_max_attempts: parse_env("WEBHOOK_REDRIVE_MAX_ATTEMPTS", 8)?, webhook_redrive_grace_secs: parse_env("WEBHOOK_REDRIVE_GRACE_SECS", 60)?, + webhook_redrive_backoff_initial_secs: parse_env( + "WEBHOOK_REDRIVE_BACKOFF_INITIAL_SECS", + 30, + )?, + webhook_redrive_backoff_max_secs: parse_env("WEBHOOK_REDRIVE_BACKOFF_MAX_SECS", 900)?, poll_interval_secs: parse_env("POLL_INTERVAL_SECS", 10)?, payment_ttl_secs: parse_env("PAYMENT_TTL_SECS", 3600)?, rate_limit_requests_per_sec: parse_env("RATE_LIMIT_REQUESTS_PER_SEC", 10)?, @@ -273,6 +292,9 @@ impl Config { /// - `WEBHOOK_RETRY_DELAY_MS == 0` with retries > 1 → retries hammer the /// target endpoint with no back-off /// - `REQUEST_TIMEOUT_SECS == 0` → every request is aborted immediately + /// - `WEBHOOK_REDRIVE_BACKOFF_MAX_SECS < WEBHOOK_REDRIVE_BACKOFF_INITIAL_SECS` + /// → the cap would silently override the starting delay, so backoff + /// never actually grows fn validate_timing(&self) -> Result<()> { if self.poll_interval_secs == 0 { return Err(anyhow::anyhow!( @@ -320,6 +342,16 @@ impl Config { )); } + if self.webhook_redrive_backoff_max_secs < self.webhook_redrive_backoff_initial_secs { + return Err(anyhow::anyhow!( + "WEBHOOK_REDRIVE_BACKOFF_MAX_SECS ({}) must be >= WEBHOOK_REDRIVE_BACKOFF_INITIAL_SECS ({}). \ + With the current settings the cap would override the starting delay and backoff \ + would never actually grow.", + self.webhook_redrive_backoff_max_secs, + self.webhook_redrive_backoff_initial_secs + )); + } + Ok(()) } @@ -435,6 +467,14 @@ impl std::fmt::Debug for Config { "webhook_redrive_grace_secs", &self.webhook_redrive_grace_secs, ) + .field( + "webhook_redrive_backoff_initial_secs", + &self.webhook_redrive_backoff_initial_secs, + ) + .field( + "webhook_redrive_backoff_max_secs", + &self.webhook_redrive_backoff_max_secs, + ) .field("poll_interval_secs", &self.poll_interval_secs) .field("payment_ttl_secs", &self.payment_ttl_secs) .field( @@ -504,6 +544,8 @@ mod tests { webhook_redrive_concurrency: 4, webhook_redrive_max_attempts: 8, webhook_redrive_grace_secs: 60, + webhook_redrive_backoff_initial_secs: 30, + webhook_redrive_backoff_max_secs: 900, poll_interval_secs: 10, payment_ttl_secs: 3600, rate_limit_requests_per_sec: 10, @@ -578,6 +620,8 @@ mod tests { webhook_redrive_concurrency: 4, webhook_redrive_max_attempts: 8, webhook_redrive_grace_secs: 60, + webhook_redrive_backoff_initial_secs: 30, + webhook_redrive_backoff_max_secs: 900, poll_interval_secs: 10, payment_ttl_secs: 3600, rate_limit_requests_per_sec: 10, @@ -844,6 +888,35 @@ mod tests { assert!(cfg.validate_timing().is_ok()); } + #[test] + fn timing_rejects_backoff_max_below_initial() { + let mut cfg = timing_config(); + cfg.webhook_redrive_backoff_initial_secs = 300; + cfg.webhook_redrive_backoff_max_secs = 30; + let err = cfg.validate_timing().unwrap_err().to_string(); + assert!( + err.contains("WEBHOOK_REDRIVE_BACKOFF_MAX_SECS") + && err.contains("WEBHOOK_REDRIVE_BACKOFF_INITIAL_SECS"), + "got: {err}" + ); + } + + #[test] + fn timing_allows_backoff_max_equal_to_initial() { + let mut cfg = timing_config(); + cfg.webhook_redrive_backoff_initial_secs = 30; + cfg.webhook_redrive_backoff_max_secs = 30; + assert!(cfg.validate_timing().is_ok()); + } + + #[test] + fn timing_allows_zero_backoff_initial_to_disable_growth() { + let mut cfg = timing_config(); + cfg.webhook_redrive_backoff_initial_secs = 0; + cfg.webhook_redrive_backoff_max_secs = 0; + assert!(cfg.validate_timing().is_ok()); + } + #[test] fn startup_fails_on_ttl_shorter_than_poll_interval_via_env() { run_with_env( diff --git a/src/db.rs b/src/db.rs index c171bb1..3829639 100644 --- a/src/db.rs +++ b/src/db.rs @@ -763,23 +763,40 @@ fn row_to_webhook_delivery(row: &sqlx::sqlite::SqliteRow) -> WebhookDelivery { /// row never attempted) that a row must clear before being considered stuck /// rather than merely in flight — callers must size it comfortably above the /// worst-case inline delivery time so this worker never races a live -/// `dispatch()` for the same row. +/// `dispatch()` for the same row. It is also the hard floor under the +/// exponential backoff below, so a row is never touched sooner than this +/// regardless of `attempts`. +/// +/// A row that has failed at least once (`attempts > 0`) additionally has to +/// clear an exponential backoff — `backoff_initial_secs * 2^(attempts-1)`, +/// capped at `backoff_max_secs` — before it is considered eligible again. +/// A row with `attempts == 0` (left behind by a crash between insert and its +/// first send, not a delivery failure) is exempt from this backoff and is +/// gated by `grace_secs` alone. pub async fn list_redrivable_deliveries( pool: &Db, max_attempts: i64, grace_secs: i64, + backoff_initial_secs: i64, + backoff_max_secs: i64, ) -> Result> { - let grace_modifier = format!("-{grace_secs} seconds"); let rows = sqlx::query( "SELECT id, payment_id, url, payload, event_type, status, attempts, last_attempt, created_at FROM webhook_deliveries WHERE status IN ('pending', 'failed') AND attempts < ? - AND COALESCE(last_attempt, created_at) <= strftime('%Y-%m-%dT%H:%M:%SZ', 'now', ?) + AND datetime(COALESCE(last_attempt, created_at), '+' || ( + CASE WHEN attempts = 0 THEN ? + ELSE MAX(?, MIN(? * (1 << MIN(attempts - 1, 32)), ?)) + END + ) || ' seconds') <= datetime('now') ORDER BY created_at ASC", ) .bind(max_attempts) - .bind(&grace_modifier) + .bind(grace_secs) + .bind(grace_secs) + .bind(backoff_initial_secs) + .bind(backoff_max_secs) .fetch_all(pool) .await?; @@ -1024,7 +1041,7 @@ mod tests { .await .unwrap(); - let candidates = list_redrivable_deliveries(&pool, 8, 0).await.unwrap(); + let candidates = list_redrivable_deliveries(&pool, 8, 0, 0, 0).await.unwrap(); let ids: Vec<&str> = candidates.iter().map(|d| d.id.as_str()).collect(); assert_eq!( ids, @@ -1108,14 +1125,78 @@ mod tests { .unwrap(); // Freshly inserted, so a large grace window makes it ineligible... - assert!(list_redrivable_deliveries(&pool, 8, 3600) + assert!(list_redrivable_deliveries(&pool, 8, 3600, 0, 0) .await .unwrap() .is_empty()); // ...while a zero grace window makes it immediately eligible. assert_eq!( - list_redrivable_deliveries(&pool, 8, 0).await.unwrap().len(), + list_redrivable_deliveries(&pool, 8, 0, 0, 0) + .await + .unwrap() + .len(), 1 ); } + + #[tokio::test] + async fn list_redrivable_deliveries_exempts_never_attempted_rows_from_backoff() { + let pool = memory_db().await; + create_payment(&pool, new_payment("p1", "MEMOR3", 3600)) + .await + .unwrap(); + save_webhook_delivery( + &pool, + "crashed", + "p1", + "http://x", + "{}", + "payment.completed", + ) + .await + .unwrap(); + + // attempts == 0 (never sent) is gated by grace_secs alone, not the + // exponential backoff, even with a huge backoff floor configured. + assert_eq!( + list_redrivable_deliveries(&pool, 8, 0, 3600, 3600) + .await + .unwrap() + .len(), + 1 + ); + } + + #[tokio::test] + async fn list_redrivable_deliveries_backs_off_exponentially_after_a_failure() { + let pool = memory_db().await; + create_payment(&pool, new_payment("p1", "MEMOR4", 3600)) + .await + .unwrap(); + save_webhook_delivery(&pool, "flaky", "p1", "http://x", "{}", "payment.completed") + .await + .unwrap(); + update_webhook_delivery(&pool, "flaky", "failed", 1) + .await + .unwrap(); + + // One failed attempt (attempts=1): backoff = initial * 2^0 = initial. + // A huge initial delay makes it ineligible even with grace_secs=0. + assert!( + list_redrivable_deliveries(&pool, 8, 0, 3600, 3600) + .await + .unwrap() + .is_empty(), + "a row with a recent failure must wait out the backoff delay" + ); + // grace_secs is a floor under the backoff: even with backoff disabled + // (initial=max=0), a large grace_secs still holds the row back. + assert!( + list_redrivable_deliveries(&pool, 8, 3600, 0, 0) + .await + .unwrap() + .is_empty(), + "grace_secs must floor eligibility even when backoff computes to 0" + ); + } } diff --git a/src/expiry.rs b/src/expiry.rs index 488573c..6e891e0 100644 --- a/src/expiry.rs +++ b/src/expiry.rs @@ -25,6 +25,10 @@ pub async fn sweep_once(state: &Arc) -> anyhow::Result { let expired = db::expire_overdue(&state.pool).await?; for payment in &expired { info!(payment_id = %payment.id, "payment intent expired"); + state.settlement_metrics.record_expired(); + if let Some(secs) = crate::metrics::seconds_since_rfc3339(&payment.created_at) { + state.settlement_metrics.record_latency_secs(secs); + } webhook::dispatch(state, payment, EXPIRED_EVENT, None).await; } Ok(expired.len()) @@ -82,6 +86,8 @@ mod tests { webhook_redrive_concurrency: 4, webhook_redrive_max_attempts: 8, webhook_redrive_grace_secs: 60, + webhook_redrive_backoff_initial_secs: 0, + webhook_redrive_backoff_max_secs: 0, poll_interval_secs: 10, payment_ttl_secs: 3600, rate_limit_requests_per_sec: 1000, @@ -112,6 +118,10 @@ mod tests { http: reqwest::Client::new(), webhook_http: reqwest::Client::new(), webhook_metrics: crate::metrics::WebhookMetrics::new(), + auth_metrics: crate::metrics::AuthMetrics::new(), + request_metrics: crate::metrics::RequestMetrics::new(), + settlement_metrics: crate::metrics::SettlementMetrics::new(), + task_health: crate::TaskHealth::new(), } } diff --git a/src/horizon.rs b/src/horizon.rs index f406abb..a9ad477 100644 --- a/src/horizon.rs +++ b/src/horizon.rs @@ -565,6 +565,27 @@ async fn settle( } info!(payment_id = %payment.id, status, %tx_hash, "payment settled"); + /* Settlement metrics (issue #133). `underpaid` is counted but excluded + from the latency histogram since the intent is still open — its eventual + top-up to `completed`/`overpaid` will record the real end-to-end latency + at that point. */ + match event { + "payment.completed" => { + state.settlement_metrics.record_completed(); + if let Some(secs) = crate::metrics::seconds_since_rfc3339(&payment.created_at) { + state.settlement_metrics.record_latency_secs(secs); + } + } + "payment.overpaid" => { + state.settlement_metrics.record_overpaid(); + if let Some(secs) = crate::metrics::seconds_since_rfc3339(&payment.created_at) { + state.settlement_metrics.record_latency_secs(secs); + } + } + "payment.underpaid" => state.settlement_metrics.record_underpaid(), + _ => {} + } + // Reflect the new state in the copy we hand to the webhook. let mut settled = payment.clone(); settled.status = status.to_string(); diff --git a/src/lib.rs b/src/lib.rs index 1cbc132..f41e0b2 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -76,6 +76,17 @@ pub struct AppState { /// histogram. Exposed via `GET /metrics` so operators can see delivery /// success rate, retry volume, and failure spikes at a glance. pub webhook_metrics: metrics::WebhookMetrics, + /// Auth middleware outcome counters: success/failure (by reason) counts. + /// Exposed via `GET /metrics` so credential-stuffing or misconfigured + /// clients are visible without grepping logs. + pub auth_metrics: metrics::AuthMetrics, + /// HTTP request counters (by method/route/status) and a latency + /// histogram. Exposed via `GET /metrics` (issue #133). + pub request_metrics: metrics::RequestMetrics, + /// Payment settlement outcome counters (completed/overpaid/underpaid/ + /// expired) and a created-to-settled latency histogram. Exposed via + /// `GET /metrics` (issue #133). + pub settlement_metrics: metrics::SettlementMetrics, /// Background task health: tracks started, stopped, and failed task counts /// for monitoring and alerting. pub task_health: TaskHealth, diff --git a/src/main.rs b/src/main.rs index 573968c..f37ce90 100644 --- a/src/main.rs +++ b/src/main.rs @@ -9,7 +9,7 @@ use stellargate::{ config::{Config, ListenerMode}, db, expiry, horizon, metrics::WebhookMetrics, - webhook, AppState, + webhook, AppState, TaskHealth, }; use tokio::sync::watch; use tracing::{info, warn}; @@ -52,7 +52,10 @@ async fn main() -> Result<()> { http, webhook_http, webhook_metrics: WebhookMetrics::new(), - task_health: crate::TaskHealth::new(), + auth_metrics: stellargate::metrics::AuthMetrics::new(), + request_metrics: stellargate::metrics::RequestMetrics::new(), + settlement_metrics: stellargate::metrics::SettlementMetrics::new(), + task_health: TaskHealth::new(), }); if cfg.gateway_configured() { @@ -152,10 +155,6 @@ async fn main() -> Result<()> { if let Some(h) = stream_handle { join_task!(h); } - join_task!(poller_handle); - join_task!(sweeper_handle); - join_task!(redrive_handle); - if let Some(h) = stream_handle { join_task!(h); } }; if tokio::time::timeout(timeout, bg).await.is_err() { info!("background tasks did not finish within 30s; forcing exit"); diff --git a/src/metrics.rs b/src/metrics.rs index d5932fe..c6b3983 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -1,5 +1,5 @@ -//! In-process metrics: atomic counters and a latency histogram for webhook -//! delivery outcomes. +//! In-process metrics: atomic counters and latency histograms for HTTP +//! requests, payment settlement, webhook delivery, and auth decisions. //! //! All types are cheaply clonable (backed by `Arc`-wrapped atomics) so they //! can be stored on `AppState` and shared across handlers and background tasks @@ -9,13 +9,35 @@ //! `GET /metrics` returns a plain-text Prometheus-compatible snapshot so any //! standard scraper can ingest the data with zero configuration. +use std::collections::HashMap; use std::sync::atomic::{AtomicU64, Ordering}; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; /// Histogram buckets for webhook delivery latency (milliseconds). /// Covers the range from sub-10 ms fast paths up to the 10 s default timeout. const LATENCY_BUCKETS_MS: &[u64] = &[10, 50, 100, 250, 500, 1_000, 2_500, 5_000, 10_000]; +/// Histogram buckets for HTTP request latency (milliseconds). API handlers +/// are in-process DB work, not outbound calls, so this is scaled well below +/// the webhook bucket set above. +const REQUEST_LATENCY_BUCKETS_MS: &[u64] = &[5, 10, 25, 50, 100, 250, 500, 1_000, 2_500]; + +/// Histogram buckets for settlement latency (seconds) — the time from a +/// payment intent's `created_at` to it reaching a terminal state. Spans from +/// near-instant stream settlement up to the default 1h payment TTL. +const SETTLEMENT_LATENCY_BUCKETS_SECS: &[u64] = &[1, 5, 10, 30, 60, 300, 900, 1_800, 3_600]; + +/// Seconds elapsed between `created_at` (an RFC 3339 timestamp, as stored in +/// `payments.created_at`) and now. Returns `None` if the timestamp can't be +/// parsed — never expected in practice, but settlement metrics are +/// best-effort observability and must never be the reason a settlement fails. +pub fn seconds_since_rfc3339(created_at: &str) -> Option { + let then = time::OffsetDateTime::parse(created_at, &time::format_description::well_known::Rfc3339) + .ok()?; + let elapsed = time::OffsetDateTime::now_utc() - then; + Some(elapsed.whole_seconds().max(0) as u64) +} + #[derive(Clone)] pub struct WebhookMetrics { inner: Arc, @@ -129,11 +151,267 @@ impl Default for WebhookMetrics { } } +/// Outcome counters for `auth_middleware`, so credential-stuffing or +/// misconfigured-client traffic is visible in the `/metrics` scrape rather +/// than only in logs. +#[derive(Clone)] +pub struct AuthMetrics { + inner: Arc, +} + +struct AuthMetricsInner { + /// Requests that presented a valid API key. + success: AtomicU64, + /// Requests with no (or a malformed) `Authorization: Bearer` header. + failure_missing_key: AtomicU64, + /// Requests with a well-formed key that didn't match any merchant. + failure_invalid_key: AtomicU64, + /// Requests that failed the key lookup itself (database error). + failure_internal_error: AtomicU64, +} + +impl Default for AuthMetricsInner { + fn default() -> Self { + Self { + success: AtomicU64::new(0), + failure_missing_key: AtomicU64::new(0), + failure_invalid_key: AtomicU64::new(0), + failure_internal_error: AtomicU64::new(0), + } + } +} + +impl AuthMetrics { + pub fn new() -> Self { + Self { + inner: Arc::new(AuthMetricsInner::default()), + } + } + + pub fn record_success(&self) { + self.inner.success.fetch_add(1, Ordering::Relaxed); + } + pub fn record_failure_missing_key(&self) { + self.inner.failure_missing_key.fetch_add(1, Ordering::Relaxed); + } + pub fn record_failure_invalid_key(&self) { + self.inner.failure_invalid_key.fetch_add(1, Ordering::Relaxed); + } + pub fn record_failure_internal_error(&self) { + self.inner + .failure_internal_error + .fetch_add(1, Ordering::Relaxed); + } + + // ── Snapshot accessors ──────────────────────────────────────────────── + + pub fn success(&self) -> u64 { + self.inner.success.load(Ordering::Relaxed) + } + pub fn failure_missing_key(&self) -> u64 { + self.inner.failure_missing_key.load(Ordering::Relaxed) + } + pub fn failure_invalid_key(&self) -> u64 { + self.inner.failure_invalid_key.load(Ordering::Relaxed) + } + pub fn failure_internal_error(&self) -> u64 { + self.inner.failure_internal_error.load(Ordering::Relaxed) + } +} + +impl Default for AuthMetrics { + fn default() -> Self { + Self::new() + } +} + +/// HTTP request counters and a latency histogram, so throughput and error +/// rates are visible without parsing logs (issue #133). +/// +/// Counted by `(method, route, status)`, where `route` is the matched route +/// *template* (e.g. `/payments/:id`), not the raw request path — this keeps +/// the label set bounded regardless of how many distinct ids clients request. +/// Requests that matched no route (404s on arbitrary paths) are counted under +/// the fixed `"unmatched"` route for the same reason. +#[derive(Clone)] +pub struct RequestMetrics { + inner: Arc, +} + +struct RequestMetricsInner { + counts: Mutex>, + latency_sum_ms: AtomicU64, + latency_count: AtomicU64, + latency_buckets: [AtomicU64; REQUEST_LATENCY_BUCKETS_MS.len() + 1], +} + +impl RequestMetrics { + pub fn new() -> Self { + Self { + inner: Arc::new(RequestMetricsInner { + counts: Mutex::new(HashMap::new()), + latency_sum_ms: AtomicU64::new(0), + latency_count: AtomicU64::new(0), + latency_buckets: Default::default(), + }), + } + } + + /// Record one completed request. `route` should be a matched-path + /// template, not a raw path, to keep cardinality bounded. + pub fn record(&self, method: &str, route: &str, status: u16) { + let mut counts = self.inner.counts.lock().unwrap(); + *counts + .entry((method.to_string(), route.to_string(), status)) + .or_insert(0) += 1; + } + + pub fn record_latency_ms(&self, ms: u64) { + self.inner.latency_sum_ms.fetch_add(ms, Ordering::Relaxed); + self.inner.latency_count.fetch_add(1, Ordering::Relaxed); + for (i, &bound) in REQUEST_LATENCY_BUCKETS_MS.iter().enumerate() { + if ms <= bound { + self.inner.latency_buckets[i].fetch_add(1, Ordering::Relaxed); + } + } + self.inner.latency_buckets[REQUEST_LATENCY_BUCKETS_MS.len()] + .fetch_add(1, Ordering::Relaxed); + } + + // ── Snapshot accessors ──────────────────────────────────────────────── + + /// Snapshot of `(method, route, status) -> count`. Cloned out from under + /// the lock so rendering never holds it. + pub fn counts(&self) -> HashMap<(String, String, u16), u64> { + self.inner.counts.lock().unwrap().clone() + } + pub fn latency_sum_ms(&self) -> u64 { + self.inner.latency_sum_ms.load(Ordering::Relaxed) + } + pub fn latency_count(&self) -> u64 { + self.inner.latency_count.load(Ordering::Relaxed) + } + pub fn latency_bucket(&self, i: usize) -> u64 { + self.inner.latency_buckets[i].load(Ordering::Relaxed) + } +} + +impl Default for RequestMetrics { + fn default() -> Self { + Self::new() + } +} + +/// Payment settlement outcome counters and a latency histogram (time from a +/// payment intent's `created_at` to it reaching a terminal state), so +/// throughput and detection latency are observable without parsing logs +/// (issue #133). +#[derive(Clone)] +pub struct SettlementMetrics { + inner: Arc, +} + +struct SettlementMetricsInner { + completed: AtomicU64, + overpaid: AtomicU64, + underpaid: AtomicU64, + expired: AtomicU64, + latency_sum_secs: AtomicU64, + latency_count: AtomicU64, + latency_buckets: [AtomicU64; SETTLEMENT_LATENCY_BUCKETS_SECS.len() + 1], +} + +impl Default for SettlementMetricsInner { + fn default() -> Self { + Self { + completed: AtomicU64::new(0), + overpaid: AtomicU64::new(0), + underpaid: AtomicU64::new(0), + expired: AtomicU64::new(0), + latency_sum_secs: AtomicU64::new(0), + latency_count: AtomicU64::new(0), + latency_buckets: Default::default(), + } + } +} + +impl SettlementMetrics { + pub fn new() -> Self { + Self { + inner: Arc::new(SettlementMetricsInner::default()), + } + } + + pub fn record_completed(&self) { + self.inner.completed.fetch_add(1, Ordering::Relaxed); + } + pub fn record_overpaid(&self) { + self.inner.overpaid.fetch_add(1, Ordering::Relaxed); + } + pub fn record_underpaid(&self) { + self.inner.underpaid.fetch_add(1, Ordering::Relaxed); + } + pub fn record_expired(&self) { + self.inner.expired.fetch_add(1, Ordering::Relaxed); + } + + /// Record settlement latency in seconds (`settled_at - created_at`). + pub fn record_latency_secs(&self, secs: u64) { + self.inner + .latency_sum_secs + .fetch_add(secs, Ordering::Relaxed); + self.inner.latency_count.fetch_add(1, Ordering::Relaxed); + for (i, &bound) in SETTLEMENT_LATENCY_BUCKETS_SECS.iter().enumerate() { + if secs <= bound { + self.inner.latency_buckets[i].fetch_add(1, Ordering::Relaxed); + } + } + self.inner.latency_buckets[SETTLEMENT_LATENCY_BUCKETS_SECS.len()] + .fetch_add(1, Ordering::Relaxed); + } + + // ── Snapshot accessors ──────────────────────────────────────────────── + + pub fn completed(&self) -> u64 { + self.inner.completed.load(Ordering::Relaxed) + } + pub fn overpaid(&self) -> u64 { + self.inner.overpaid.load(Ordering::Relaxed) + } + pub fn underpaid(&self) -> u64 { + self.inner.underpaid.load(Ordering::Relaxed) + } + pub fn expired(&self) -> u64 { + self.inner.expired.load(Ordering::Relaxed) + } + pub fn latency_sum_secs(&self) -> u64 { + self.inner.latency_sum_secs.load(Ordering::Relaxed) + } + pub fn latency_count(&self) -> u64 { + self.inner.latency_count.load(Ordering::Relaxed) + } + pub fn latency_bucket(&self, i: usize) -> u64 { + self.inner.latency_buckets[i].load(Ordering::Relaxed) + } +} + +impl Default for SettlementMetrics { + fn default() -> Self { + Self::new() + } +} + // ── Prometheus text exposition ──────────────────────────────────────────────── -/// Render webhook delivery metrics as a Prometheus-compatible plain-text snapshot. -/// Called by `GET /metrics`. -pub fn render(webhook: &WebhookMetrics) -> String { +/// Render webhook delivery, auth outcome, HTTP request, and settlement +/// metrics as a Prometheus-compatible plain-text snapshot. Called by +/// `GET /metrics`. +pub fn render( + webhook: &WebhookMetrics, + auth: &AuthMetrics, + request: &RequestMetrics, + settlement: &SettlementMetrics, +) -> String { let mut out = String::with_capacity(1024); // stellargate_webhook_deliveries_total — counter vec by outcome @@ -181,5 +459,108 @@ pub fn render(webhook: &WebhookMetrics) -> String { webhook.latency_count() )); + // stellargate_auth_attempts_total — counter vec by outcome/reason + out.push_str( + "# HELP stellargate_auth_attempts_total Total auth middleware decisions by outcome and reason.\n", + ); + out.push_str("# TYPE stellargate_auth_attempts_total counter\n"); + out.push_str(&format!( + "stellargate_auth_attempts_total{{outcome=\"success\"}} {}\n", + auth.success() + )); + out.push_str(&format!( + "stellargate_auth_attempts_total{{outcome=\"failure\",reason=\"missing_key\"}} {}\n", + auth.failure_missing_key() + )); + out.push_str(&format!( + "stellargate_auth_attempts_total{{outcome=\"failure\",reason=\"invalid_key\"}} {}\n", + auth.failure_invalid_key() + )); + out.push_str(&format!( + "stellargate_auth_attempts_total{{outcome=\"failure\",reason=\"internal_error\"}} {}\n", + auth.failure_internal_error() + )); + + // stellargate_http_requests_total — counter vec by method/route/status + out.push_str( + "# HELP stellargate_http_requests_total Total HTTP requests by method, matched route, and status code.\n", + ); + out.push_str("# TYPE stellargate_http_requests_total counter\n"); + let mut counts: Vec<((String, String, u16), u64)> = request.counts().into_iter().collect(); + counts.sort(); + for ((method, route, status), count) in counts { + out.push_str(&format!( + "stellargate_http_requests_total{{method=\"{method}\",route=\"{route}\",status=\"{status}\"}} {count}\n" + )); + } + + // stellargate_http_request_duration_ms — histogram + out.push_str("# HELP stellargate_http_request_duration_ms HTTP request latency in milliseconds.\n"); + out.push_str("# TYPE stellargate_http_request_duration_ms histogram\n"); + for (i, &bound) in REQUEST_LATENCY_BUCKETS_MS.iter().enumerate() { + out.push_str(&format!( + "stellargate_http_request_duration_ms_bucket{{le=\"{}\"}} {}\n", + bound, + request.latency_bucket(i) + )); + } + out.push_str(&format!( + "stellargate_http_request_duration_ms_bucket{{le=\"+Inf\"}} {}\n", + request.latency_bucket(REQUEST_LATENCY_BUCKETS_MS.len()) + )); + out.push_str(&format!( + "stellargate_http_request_duration_ms_sum {}\n", + request.latency_sum_ms() + )); + out.push_str(&format!( + "stellargate_http_request_duration_ms_count {}\n", + request.latency_count() + )); + + // stellargate_payments_settled_total — counter vec by outcome + out.push_str( + "# HELP stellargate_payments_settled_total Total payment intents reaching a terminal state, by outcome.\n", + ); + out.push_str("# TYPE stellargate_payments_settled_total counter\n"); + out.push_str(&format!( + "stellargate_payments_settled_total{{outcome=\"completed\"}} {}\n", + settlement.completed() + )); + out.push_str(&format!( + "stellargate_payments_settled_total{{outcome=\"overpaid\"}} {}\n", + settlement.overpaid() + )); + out.push_str(&format!( + "stellargate_payments_settled_total{{outcome=\"underpaid\"}} {}\n", + settlement.underpaid() + )); + out.push_str(&format!( + "stellargate_payments_settled_total{{outcome=\"expired\"}} {}\n", + settlement.expired() + )); + + // stellargate_payment_settlement_latency_seconds — histogram + out.push_str("# HELP stellargate_payment_settlement_latency_seconds Time from a payment intent's creation to it reaching a terminal state, in seconds.\n"); + out.push_str("# TYPE stellargate_payment_settlement_latency_seconds histogram\n"); + for (i, &bound) in SETTLEMENT_LATENCY_BUCKETS_SECS.iter().enumerate() { + out.push_str(&format!( + "stellargate_payment_settlement_latency_seconds_bucket{{le=\"{}\"}} {}\n", + bound, + settlement.latency_bucket(i) + )); + } + out.push_str(&format!( + "stellargate_payment_settlement_latency_seconds_bucket{{le=\"+Inf\"}} {}\n", + settlement.latency_bucket(SETTLEMENT_LATENCY_BUCKETS_SECS.len()) + )); + out.push_str(&format!( + "stellargate_payment_settlement_latency_seconds_sum {}\n", + settlement.latency_sum_secs() + )); + out.push_str(&format!( + "stellargate_payment_settlement_latency_seconds_count {}\n", + settlement.latency_count() + )); + out } diff --git a/src/webhook.rs b/src/webhook.rs index c8872bd..895b97e 100644 --- a/src/webhook.rs +++ b/src/webhook.rs @@ -239,6 +239,8 @@ pub async fn redrive_once(state: &Arc) -> usize { &state.pool, state.config.webhook_redrive_max_attempts as i64, state.config.webhook_redrive_grace_secs, + state.config.webhook_redrive_backoff_initial_secs, + state.config.webhook_redrive_backoff_max_secs, ) .await { diff --git a/tests/api_tests.rs b/tests/api_tests.rs index be2e53c..85e6469 100644 --- a/tests/api_tests.rs +++ b/tests/api_tests.rs @@ -28,6 +28,8 @@ fn make_config() -> Config { webhook_redrive_concurrency: 4, webhook_redrive_max_attempts: 8, webhook_redrive_grace_secs: 60, + webhook_redrive_backoff_initial_secs: 0, + webhook_redrive_backoff_max_secs: 0, poll_interval_secs: 10, payment_ttl_secs: 3600, /* High enough that these tests never trip the limiter; dedicated @@ -67,6 +69,9 @@ async fn server_with_config(cfg: Config) -> (TestServer, db::Db) { http, webhook_http: reqwest::Client::new(), webhook_metrics: stellargate::metrics::WebhookMetrics::new(), + auth_metrics: stellargate::metrics::AuthMetrics::new(), + request_metrics: stellargate::metrics::RequestMetrics::new(), + settlement_metrics: stellargate::metrics::SettlementMetrics::new(), task_health: stellargate::TaskHealth::new(), })) .into_make_service_with_connect_info::(); @@ -160,6 +165,94 @@ async fn test_invalid_api_key_returns_401() { res.assert_status(StatusCode::UNAUTHORIZED); } +/// Auth outcomes must be observable via `/metrics` (issue #139), not just as +/// a bare 401 with nothing left behind for an operator to alert on. +#[tokio::test] +async fn test_auth_outcomes_are_counted_in_metrics() { + let server = test_server().await; + + server.get("/payments").await; // missing key + server + .post("/payments") + .add_header("Authorization", "Bearer not-a-real-key") + .json(&json!({ "amount": "10", "asset": "XLM" })) + .await; // invalid key + let key = provision_merchant(&server).await; + server + .get("/payments") + .add_header("Authorization", format!("Bearer {key}")) + .await; // valid key + + let res = server.get("/metrics").await; + res.assert_status_ok(); + let body = res.text(); + assert!( + body.contains("stellargate_auth_attempts_total{outcome=\"success\"} 1"), + "got: {body}" + ); + assert!( + body.contains( + "stellargate_auth_attempts_total{outcome=\"failure\",reason=\"missing_key\"} 1" + ), + "got: {body}" + ); + assert!( + body.contains( + "stellargate_auth_attempts_total{outcome=\"failure\",reason=\"invalid_key\"} 1" + ), + "got: {body}" + ); +} + +/// HTTP requests must be counted by matched route template (not raw path), +/// so `/payments/:id`-style requests don't blow up label cardinality, and +/// the response status must be reflected too (issue #133). +#[tokio::test] +async fn test_http_requests_are_counted_by_matched_route_in_metrics() { + let server = test_server().await; + + server.get("/health").await; + let key = provision_merchant(&server).await; + let res = server + .post("/payments") + .add_header("Authorization", format!("Bearer {key}")) + .json(&json!({ "amount": "10", "asset": "XLM" })) + .await; + let payment_id = res.json::()["id"].as_str().unwrap().to_string(); + + // A real id (200) and a made-up one (404) both hit GET /payments/:id — the + // route label must be the template, not either literal id. + server + .get(&format!("/payments/{payment_id}")) + .add_header("Authorization", format!("Bearer {key}")) + .await; + server.get("/payments/some-other-id").await; + + let res = server.get("/metrics").await; + res.assert_status_ok(); + let body = res.text(); + assert!( + body.contains("stellargate_http_requests_total{method=\"GET\",route=\"/health\",status=\"200\"} 1"), + "got: {body}" + ); + assert!( + body.contains( + "stellargate_http_requests_total{method=\"GET\",route=\"/payments/:id\",status=\"200\"} 1" + ), + "route label must be the matched-path template, not the literal id; got: {body}" + ); + assert!( + body.contains( + "stellargate_http_requests_total{method=\"GET\",route=\"/payments/:id\",status=\"404\"} 1" + ), + "got: {body}" + ); + assert!( + body.contains("stellargate_http_request_duration_ms_count"), + "got: {body}" + ); +} + #[tokio::test] async fn test_create_payment() { let server = test_server().await; diff --git a/tests/concurrency_tests.rs b/tests/concurrency_tests.rs index c4f71a6..1b35bdd 100644 --- a/tests/concurrency_tests.rs +++ b/tests/concurrency_tests.rs @@ -82,6 +82,8 @@ fn make_state(pool: db::Db, _webhook_url: Option) -> Arc { webhook_redrive_concurrency: 4, webhook_redrive_max_attempts: 8, webhook_redrive_grace_secs: 60, + webhook_redrive_backoff_initial_secs: 0, + webhook_redrive_backoff_max_secs: 0, poll_interval_secs: 10, payment_ttl_secs: 3600, rate_limit_requests_per_sec: 10000, @@ -97,6 +99,9 @@ fn make_state(pool: db::Db, _webhook_url: Option) -> Arc { http: reqwest::Client::new(), webhook_http: reqwest::Client::new(), webhook_metrics: stellargate::metrics::WebhookMetrics::new(), + auth_metrics: stellargate::metrics::AuthMetrics::new(), + request_metrics: stellargate::metrics::RequestMetrics::new(), + settlement_metrics: stellargate::metrics::SettlementMetrics::new(), task_health: stellargate::TaskHealth::new(), }) } @@ -364,3 +369,38 @@ async fn reprocessing_past_transactions_never_double_credits() { assert_state(&pool, &payment_id, "completed", "10").await; } } + +/// A completed settlement must be counted in `SettlementMetrics`, and its +/// latency histogram must gain an observation — otherwise `/metrics` would +/// silently under-report throughput (issue #133). +#[tokio::test] +async fn settlement_is_recorded_in_metrics() { + let pool = memory_pool().await; + let payment_id = seed_pending_payment(&pool, None).await; + let state = make_state(pool, None); + let hp = make_horizon_payment(); + + assert_eq!(state.settlement_metrics.completed(), 0); + assert_eq!(state.settlement_metrics.latency_count(), 0); + + let settled = reconcile_payment(&state, &hp).await.unwrap(); + assert!(settled, "an exact payment must settle"); + + assert_eq!( + state.settlement_metrics.completed(), + 1, + "a completed settlement must be counted" + ); + assert_eq!( + state.settlement_metrics.latency_count(), + 1, + "settlement must add one observation to the latency histogram" + ); + + // Sanity: the id we settled is the one we seeded. + let payment = db::get_payment(&state.pool, &payment_id) + .await + .unwrap() + .unwrap(); + assert_eq!(payment.status, "completed"); +} diff --git a/tests/rate_limit_tests.rs b/tests/rate_limit_tests.rs index 86348cc..9e59598 100644 --- a/tests/rate_limit_tests.rs +++ b/tests/rate_limit_tests.rs @@ -32,6 +32,8 @@ fn make_config(rate_limit_requests_per_sec: u32) -> Config { webhook_redrive_concurrency: 4, webhook_redrive_max_attempts: 8, webhook_redrive_grace_secs: 60, + webhook_redrive_backoff_initial_secs: 0, + webhook_redrive_backoff_max_secs: 0, poll_interval_secs: 10, payment_ttl_secs: 3600, rate_limit_requests_per_sec, @@ -64,6 +66,9 @@ async fn server_with_config(cfg: Config) -> (TestServer, db::Db) { http, webhook_http: reqwest::Client::new(), webhook_metrics: stellargate::metrics::WebhookMetrics::new(), + auth_metrics: stellargate::metrics::AuthMetrics::new(), + request_metrics: stellargate::metrics::RequestMetrics::new(), + settlement_metrics: stellargate::metrics::SettlementMetrics::new(), task_health: stellargate::TaskHealth::new(), })) .into_make_service_with_connect_info::(); diff --git a/tests/trustline_tests.rs b/tests/trustline_tests.rs index 72c1acf..7ea7eb0 100644 --- a/tests/trustline_tests.rs +++ b/tests/trustline_tests.rs @@ -59,6 +59,8 @@ async fn make_state(horizon_url: String) -> Arc { webhook_redrive_concurrency: 4, webhook_redrive_max_attempts: 8, webhook_redrive_grace_secs: 60, + webhook_redrive_backoff_initial_secs: 0, + webhook_redrive_backoff_max_secs: 0, poll_interval_secs: 10, payment_ttl_secs: 3600, rate_limit_requests_per_sec: 10000, @@ -73,6 +75,9 @@ async fn make_state(horizon_url: String) -> Arc { http: reqwest::Client::new(), webhook_http: reqwest::Client::new(), webhook_metrics: stellargate::metrics::WebhookMetrics::new(), + auth_metrics: stellargate::metrics::AuthMetrics::new(), + request_metrics: stellargate::metrics::RequestMetrics::new(), + settlement_metrics: stellargate::metrics::SettlementMetrics::new(), task_health: stellargate::TaskHealth::new(), }) } diff --git a/tests/webhook_dispatch_tests.rs b/tests/webhook_dispatch_tests.rs index a894331..76f6ac3 100644 --- a/tests/webhook_dispatch_tests.rs +++ b/tests/webhook_dispatch_tests.rs @@ -35,6 +35,8 @@ fn make_config(webhook_secret: &str, retry_attempts: u32) -> Config { webhook_redrive_concurrency: 4, webhook_redrive_max_attempts: 8, webhook_redrive_grace_secs: 60, + webhook_redrive_backoff_initial_secs: 0, + webhook_redrive_backoff_max_secs: 0, poll_interval_secs: 10, payment_ttl_secs: 3600, cors_allowed_origins: vec![], @@ -66,6 +68,9 @@ async fn setup_state(cfg: Config) -> AppState { http: reqwest::Client::new(), webhook_http: reqwest::Client::new(), webhook_metrics: stellargate::metrics::WebhookMetrics::new(), + auth_metrics: stellargate::metrics::AuthMetrics::new(), + request_metrics: stellargate::metrics::RequestMetrics::new(), + settlement_metrics: stellargate::metrics::SettlementMetrics::new(), task_health: stellargate::TaskHealth::new(), } }