feat(metrics): add HTTP request and payment settlement metrics

.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

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"

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` |

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”

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<AppState>) -> 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 <key>` 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<Arc<AppState>>,
     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<Arc<AppState>>,
+    req: Request,
+    next: Next,
+) -> axum::response::Response {
+    let method = req.method().to_string();
+    let route = req
+        .extensions()
+        .get::<MatchedPath>()
+        .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<RateLimitState>,
     req: Request,
@@ -365,7 +434,12 @@ async fn check_horizon_ready(state: &Arc<AppState>) -> Result<(), String> {
 
 /// `GET /metrics` — Prometheus-compatible plain-text metrics snapshot.
 async fn metrics_handler(State(state): State<Arc<AppState>>) -> 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,
         [(

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(