From 8a4e4b2ecb4a37528cb83205f4006de35db791cf Mon Sep 17 00:00:00 2001 From: feyishola Date: Sun, 28 Jun 2026 17:13:26 +0100 Subject: [PATCH 1/2] webhook feature implemented --- README.md | 95 ++++++- src/config.rs | 108 ++++++++ src/lib.rs | 2 + src/main.rs | 30 ++- src/metrics.rs | 75 ++++++ src/webhook.rs | 695 +++++++++++++++++++++++++++++++++++++++++++++++++ 6 files changed, 1002 insertions(+), 3 deletions(-) create mode 100644 src/webhook.rs diff --git a/README.md b/README.md index 198e038..e32363f 100644 --- a/README.md +++ b/README.md @@ -351,6 +351,15 @@ The `MetricsRegistry` (defined in `src/metrics.rs`) is the central instrumentati | `config_validation_failures_total` | Counter | Total configuration validation failures | | `config_reload_total` | Counter | Total configuration reloads attempted | +#### Webhook Delivery Metrics + +| Metric | Type | Labels | Description | +|---|---|---|---| +| `webhook_deliveries_total` | CounterVec | `status` (success/dead_lettered) | Total webhook delivery outcomes | +| `webhook_delivery_latency_seconds` | HistogramVec | `status` | End-to-end delivery latency including all retries | +| `webhook_dlq_depth` | Gauge | โ€” | Current number of entries in the dead-letter queue | +| `webhook_retries_total` | Counter | โ€” | Total webhook retry attempts | + #### Recommended Alerting Thresholds | Alert | Condition | Severity | @@ -362,6 +371,8 @@ The `MetricsRegistry` (defined in `src/metrics.rs`) is the central instrumentati | Event backlog growing | `event_backlog_size > 1000` | Warning | | Config validation failures | `increase(config_validation_failures_total[5m]) > 0` | Critical | | High Horizon latency | `histogram_quantile(0.95, rate(horizon_latency_seconds_bucket[5m])) > 5` | Warning | +| Webhook DLQ growing | `webhook_dlq_depth > 0` | Warning | +| High webhook failure rate | `rate(webhook_deliveries_total{status="dead_lettered"}[5m]) > 0` | Critical | #### Running with Metrics @@ -400,8 +411,13 @@ scrape_configs: | `STELLAR_CIRCUIT_BREAKER_OPEN_DURATION_MS` | `30000` | Milliseconds the circuit remains open before allowing a half-open probe | | `STELLAR_CIRCUIT_BREAKER_HALF_OPEN_MAX_CALLS` | `1` | Concurrent half-open probes allowed before recovery or reopening | | `LOG_LEVEL` | `info` | Log verbosity string | -| `WEBHOOK_URLS` | empty | Comma-separated list of valid URLs | -| `WEBHOOK_SECRET` | unset | Optional webhook signing secret | +| `WEBHOOK_URLS` | empty | Comma-separated list of valid URLs to receive webhook events | +| `WEBHOOK_SECRET` | unset | Optional shared secret sent as `X-Webhook-Secret` header | +| `WEBHOOK_MAX_RETRIES` | `5` | Retry attempts after the initial webhook delivery fails | +| `WEBHOOK_RETRY_BASE_DELAY_MS` | `200` | Initial exponential backoff delay in milliseconds; must be greater than `0` | +| `WEBHOOK_RETRY_MAX_DELAY_MS` | `30000` | Maximum backoff delay in milliseconds; must be โ‰ฅ base delay | +| `WEBHOOK_REQUEST_TIMEOUT_MS` | `10000` | Per-request webhook HTTP timeout in milliseconds; must be greater than `0` | +| `WEBHOOK_JITTER_ENABLED` | `true` | Boolean; adds random jitter of up to 25 % of the capped delay | | `CACHE_VERIFICATION_TTL` | `3600` | Seconds before a cached verification result expires | Set `REDIS_URL` to a real Redis instance in production. The in-memory backend is suitable for local development and testing only. @@ -419,6 +435,81 @@ Audit records should be retained for as long as the operator needs replay and fo --- +## ๐Ÿ”” Webhook Delivery + +After an event is finalized, the service dispatches it asynchronously to every URL listed in `WEBHOOK_URLS`. External systems subscribe to these events to maintain up-to-date replicas of document state. + +### Event payload schema + +```json +{ + "event_id": "f47ac10b-58cc-4372-a567-0e02b2c3d479", + "event_type": "DocumentRegistered", + "idempotency_key": "contract:tx123:42:3:doc-1:DocumentRegistered", + "sequence": 42003, + "timestamp": "2026-06-28T12:00:00Z", + "aggregate_id": "doc-1", + "actor": "GDEX...", + "data": { "issuer": "GDEX...", "owner": "GBBB..." }, + "metadata": { + "transaction_hash": "tx123", + "ledger_sequence": 42, + "event_index": 3, + "document_hash": "e3b0c4...", + "source": "contract" + } +} +``` + +| Field | Description | +|---|---| +| `event_id` | UUID v4 unique to this event record | +| `event_type` | One of `DocumentRegistered`, `DocumentRevoked`, `DocumentVerified`, `DocumentAuthorizationFailed`, `DocumentOwnerChanged` | +| `idempotency_key` | Stable token derived from transaction hash + event index. Use this to deduplicate retried deliveries. | +| `sequence` | Monotonically increasing within an aggregate. For contract events: `ledger_sequence * 1000 + event_index`. | +| `timestamp` | ISO-8601 UTC timestamp when the event was recorded | +| `metadata` | Present for contract-origin events; contains `transaction_hash`, `ledger_sequence`, `event_index`, `document_hash` | + +### Request headers + +Each webhook HTTP POST includes the following headers: + +| Header | Value | +|---|---| +| `Content-Type` | `application/json` | +| `X-Idempotency-Key` | The event's `idempotency_key` | +| `X-Event-Id` | The event's `event_id` | +| `X-Event-Type` | The event's `event_type` | +| `X-Webhook-Secret` | Value of `WEBHOOK_SECRET` if configured | + +### Retry semantics + +Deliveries use exponential backoff with jitter: + +``` +delay(attempt) = min(base * 2^attempt, max) + random_jitter(0, delay/4) +``` + +- `base` is `WEBHOOK_RETRY_BASE_DELAY_MS` (default `200` ms) +- `max` is `WEBHOOK_RETRY_MAX_DELAY_MS` (default `30 000` ms) +- Jitter is drawn uniformly from `[0, capped_delay / 4)` using wall-clock sub-millisecond noise +- Total attempts = `WEBHOOK_MAX_RETRIES + 1` (default 6 total) + +### Ordering guarantees + +URLs are contacted **sequentially in registration order**. An event is attempted against every URL regardless of individual failures โ€” a URL that exhausts retries is dead-lettered without blocking delivery to subsequent URLs. + +### Dead-letter queue + +Failed deliveries (all retries exhausted) are moved to an in-memory bounded queue (max 10 000 entries). The queue is accessible via: + +- `GET /webhooks/dlq` โ€” returns `{"dlq_depth": N}` +- `POST /webhooks/dlq/drain` โ€” drains and returns all entries: `{"drained": N, "entries": [...]}` + +Each dead-letter entry contains the original payload, target URL, attempt count, last error, and failure timestamp. Replaying drained entries is the operator's responsibility. + +--- + ## ๐Ÿงช Future Improvements * Issuer registry system diff --git a/src/config.rs b/src/config.rs index 42bb652..6a18eed 100644 --- a/src/config.rs +++ b/src/config.rs @@ -51,6 +51,11 @@ pub struct AppConfig { pub log_level: String, pub webhook_urls: Vec, pub webhook_secret: Option, + pub webhook_max_retries: u32, + pub webhook_retry_base_delay_ms: u64, + pub webhook_retry_max_delay_ms: u64, + pub webhook_request_timeout_ms: u64, + pub webhook_jitter_enabled: bool, pub cache_verification_ttl: u64, } @@ -98,6 +103,11 @@ impl fmt::Debug for AppConfig { "webhook_secret", &self.webhook_secret.as_deref().map(|_| ""), ) + .field("webhook_max_retries", &self.webhook_max_retries) + .field("webhook_retry_base_delay_ms", &self.webhook_retry_base_delay_ms) + .field("webhook_retry_max_delay_ms", &self.webhook_retry_max_delay_ms) + .field("webhook_request_timeout_ms", &self.webhook_request_timeout_ms) + .field("webhook_jitter_enabled", &self.webhook_jitter_enabled) .field("cache_verification_ttl", &self.cache_verification_ttl) .finish() } @@ -153,6 +163,14 @@ impl AppConfig { } }; let webhook_secret = env::var("WEBHOOK_SECRET").ok(); + let webhook_max_retries_raw = get_env_or_default("WEBHOOK_MAX_RETRIES", "5"); + let webhook_retry_base_delay_ms_raw = + get_env_or_default("WEBHOOK_RETRY_BASE_DELAY_MS", "200"); + let webhook_retry_max_delay_ms_raw = + get_env_or_default("WEBHOOK_RETRY_MAX_DELAY_MS", "30000"); + let webhook_request_timeout_ms_raw = + get_env_or_default("WEBHOOK_REQUEST_TIMEOUT_MS", "10000"); + let webhook_jitter_raw = get_env_or_default("WEBHOOK_JITTER_ENABLED", "true"); let rate_limit_per_second_raw = get_env_or_default("RATE_LIMIT_PER_SECOND", "100"); let rate_limit_burst_raw = @@ -465,6 +483,81 @@ impl AppConfig { ); } + let webhook_max_retries: u32 = match webhook_max_retries_raw.parse() { + Ok(v) => v, + Err(_) => { + errors.push(format!( + "WEBHOOK_MAX_RETRIES must be a valid u32, got '{}'", + webhook_max_retries_raw + )); + 5 + } + }; + + let webhook_retry_base_delay_ms: u64 = match webhook_retry_base_delay_ms_raw.parse() { + Ok(v) if v > 0 => v, + Ok(_) => { + errors.push("WEBHOOK_RETRY_BASE_DELAY_MS must be greater than 0".to_string()); + 200 + } + Err(_) => { + errors.push(format!( + "WEBHOOK_RETRY_BASE_DELAY_MS must be a valid u64, got '{}'", + webhook_retry_base_delay_ms_raw + )); + 200 + } + }; + + let webhook_retry_max_delay_ms: u64 = match webhook_retry_max_delay_ms_raw.parse() { + Ok(v) if v > 0 => v, + Ok(_) => { + errors.push("WEBHOOK_RETRY_MAX_DELAY_MS must be greater than 0".to_string()); + 30_000 + } + Err(_) => { + errors.push(format!( + "WEBHOOK_RETRY_MAX_DELAY_MS must be a valid u64, got '{}'", + webhook_retry_max_delay_ms_raw + )); + 30_000 + } + }; + + let webhook_request_timeout_ms: u64 = match webhook_request_timeout_ms_raw.parse() { + Ok(v) if v > 0 => v, + Ok(_) => { + errors.push("WEBHOOK_REQUEST_TIMEOUT_MS must be greater than 0".to_string()); + 10_000 + } + Err(_) => { + errors.push(format!( + "WEBHOOK_REQUEST_TIMEOUT_MS must be a valid u64, got '{}'", + webhook_request_timeout_ms_raw + )); + 10_000 + } + }; + + let webhook_jitter_enabled = match webhook_jitter_raw.to_lowercase().as_str() { + "1" | "true" | "yes" | "y" => true, + "0" | "false" | "no" | "n" => false, + other => { + errors.push(format!( + "WEBHOOK_JITTER_ENABLED must be a boolean, got '{}'", + other + )); + true + } + }; + + if webhook_retry_max_delay_ms < webhook_retry_base_delay_ms { + errors.push( + "WEBHOOK_RETRY_MAX_DELAY_MS must be greater than or equal to WEBHOOK_RETRY_BASE_DELAY_MS" + .to_string(), + ); + } + let webhook_urls: Vec = webhook_urls_raw .split(',') .map(str::trim) @@ -511,6 +604,11 @@ impl AppConfig { log_level, webhook_urls, webhook_secret, + webhook_max_retries, + webhook_retry_base_delay_ms, + webhook_retry_max_delay_ms, + webhook_request_timeout_ms, + webhook_jitter_enabled, cache_verification_ttl, }) } @@ -545,6 +643,11 @@ mod tests { "LOG_LEVEL", "WEBHOOK_URLS", "WEBHOOK_SECRET", + "WEBHOOK_MAX_RETRIES", + "WEBHOOK_RETRY_BASE_DELAY_MS", + "WEBHOOK_RETRY_MAX_DELAY_MS", + "WEBHOOK_REQUEST_TIMEOUT_MS", + "WEBHOOK_JITTER_ENABLED", "CACHE_VERIFICATION_TTL", ]; for key in keys { @@ -722,6 +825,11 @@ mod tests { log_level: "info".to_string(), webhook_urls: vec!["https://webhook.example.com".to_string()], webhook_secret: Some("another-secret".to_string()), + webhook_max_retries: 5, + webhook_retry_base_delay_ms: 200, + webhook_retry_max_delay_ms: 30_000, + webhook_request_timeout_ms: 10_000, + webhook_jitter_enabled: true, cache_verification_ttl: 3600, }; diff --git a/src/lib.rs b/src/lib.rs index 2f09189..e1b0d59 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -21,6 +21,8 @@ pub mod metrics; pub mod rate_limit; #[cfg(not(target_arch = "wasm32"))] pub mod stellar; +#[cfg(not(target_arch = "wasm32"))] +pub mod webhook; use soroban_sdk::{ contract, contracterror, contractevent, contractimpl, contracttype, Address, BytesN, Env, Symbol, Vec, diff --git a/src/main.rs b/src/main.rs index ba84119..1c65a5d 100644 --- a/src/main.rs +++ b/src/main.rs @@ -38,17 +38,19 @@ mod native { use axum::extract::State; use axum::response::IntoResponse; - use axum::routing::get; + use axum::routing::{get, post}; use axum::{Json, Router}; use serde_json::json; use proofstell_contract::config::AppConfig; use proofstell_contract::metrics::MetricsRegistry; + use proofstell_contract::webhook::WebhookDispatcher; /// Shared application state, accessible by all axum handlers. #[derive(Clone)] struct AppState { metrics: Arc, + webhook: Arc, } /// Build the axum router with all application routes. @@ -56,6 +58,8 @@ mod native { Router::new() .route("/health", get(health_handler)) .route("/metrics", get(metrics_handler)) + .route("/webhooks/dlq", get(dlq_status_handler)) + .route("/webhooks/dlq/drain", post(dlq_drain_handler)) .with_state(state) } @@ -69,6 +73,18 @@ mod native { state.metrics.render() } + /// `GET /webhooks/dlq` โ€” returns the current DLQ depth. + async fn dlq_status_handler(State(state): State) -> impl IntoResponse { + let depth = state.webhook.dlq_depth().await; + Json(json!({ "dlq_depth": depth })) + } + + /// `POST /webhooks/dlq/drain` โ€” drains and returns all DLQ entries for manual replay. + async fn dlq_drain_handler(State(state): State) -> impl IntoResponse { + let entries = state.webhook.drain_dlq().await; + Json(json!({ "drained": entries.len(), "entries": entries })) + } + /// Bootstrap: load config, wire up services, and start the server. pub async fn run() -> anyhow::Result<()> { // โ”€โ”€ Metrics โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ @@ -89,10 +105,22 @@ mod native { "[proofstell] rate_limit: {}/s (burst {})", config.rate_limit_per_second, config.rate_limit_burst ); + eprintln!( + "[proofstell] webhooks: {} url(s) configured (max_retries={})", + config.webhook_urls.len(), + config.webhook_max_retries, + ); + + // โ”€โ”€ Webhook dispatcher โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + let webhook = Arc::new(WebhookDispatcher::from_app_config( + &config, + Some(Arc::clone(&metrics)), + )); // โ”€โ”€ Router โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ let state = AppState { metrics: Arc::clone(&metrics), + webhook, }; let app = build_router(state); diff --git a/src/metrics.rs b/src/metrics.rs index 82d373b..f14c8f2 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -52,6 +52,12 @@ pub struct MetricsRegistry { // โ”€โ”€ Config validation metrics โ”€โ”€ config_validation_failures: IntCounter, config_reload_total: IntCounter, + + // โ”€โ”€ Webhook delivery metrics โ”€โ”€ + webhook_deliveries_total: IntCounterVec, + webhook_delivery_latency_seconds: HistogramVec, + webhook_dlq_depth: Gauge, + webhook_retries_total: IntCounter, } impl Default for MetricsRegistry { @@ -199,6 +205,37 @@ impl MetricsRegistry { ) .unwrap(); + // โ”€โ”€ Webhook delivery metrics โ”€โ”€ + let webhook_deliveries_total = IntCounterVec::new( + Opts::new( + "webhook_deliveries_total", + "Total webhook delivery attempts by outcome", + ), + &["status"], + ) + .unwrap(); + + let webhook_delivery_latency_seconds = HistogramVec::new( + HistogramOpts::new( + "webhook_delivery_latency_seconds", + "End-to-end webhook delivery latency in seconds", + ), + &["status"], + ) + .unwrap(); + + let webhook_dlq_depth = Gauge::new( + "webhook_dlq_depth", + "Current number of entries in the webhook dead-letter queue", + ) + .unwrap(); + + let webhook_retries_total = IntCounter::new( + "webhook_retries_total", + "Total webhook delivery retry attempts", + ) + .unwrap(); + // โ”€โ”€ Register everything โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ for metric in [ Box::new(request_count.clone()) as Box, @@ -222,6 +259,10 @@ impl MetricsRegistry { Box::new(event_backlog_size.clone()), Box::new(config_validation_failures.clone()), Box::new(config_reload_total.clone()), + Box::new(webhook_deliveries_total.clone()), + Box::new(webhook_delivery_latency_seconds.clone()), + Box::new(webhook_dlq_depth.clone()), + Box::new(webhook_retries_total.clone()), ] { registry.register(metric).unwrap(); } @@ -249,6 +290,10 @@ impl MetricsRegistry { event_backlog_size, config_validation_failures, config_reload_total, + webhook_deliveries_total, + webhook_delivery_latency_seconds, + webhook_dlq_depth, + webhook_retries_total, } } @@ -383,6 +428,28 @@ impl MetricsRegistry { self.config_reload_total.inc(); } + // โ”€โ”€ Webhook delivery metrics โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + + /// Record a completed delivery attempt (success or dead_lettered) with latency. + pub fn record_webhook_delivery(&self, status: &str, latency_secs: f64) { + self.webhook_deliveries_total + .with_label_values(&[status]) + .inc(); + self.webhook_delivery_latency_seconds + .with_label_values(&[status]) + .observe(latency_secs); + } + + /// Increment the webhook retry counter by one. + pub fn increment_webhook_retry(&self) { + self.webhook_retries_total.inc(); + } + + /// Set the dead-letter queue depth gauge. + pub fn set_webhook_dlq_depth(&self, depth: i64) { + self.webhook_dlq_depth.set(depth as f64); + } + // โ”€โ”€ Latency helper โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ /// Start a timer for measuring operation latency. @@ -447,6 +514,10 @@ mod tests { metrics.decrement_event_backlog(); metrics.increment_config_validation_failure(); metrics.increment_config_reload(); + metrics.record_webhook_delivery("success", 0.05); + metrics.record_webhook_delivery("dead_lettered", 1.0); + metrics.increment_webhook_retry(); + metrics.set_webhook_dlq_depth(3); let output = metrics.render(); assert!(output.contains("requests_total")); @@ -458,6 +529,10 @@ mod tests { assert!(output.contains("rate_limit_rejections_total")); assert!(output.contains("event_backlog_size")); assert!(output.contains("config_validation_failures_total")); + assert!(output.contains("webhook_deliveries_total")); + assert!(output.contains("webhook_delivery_latency_seconds")); + assert!(output.contains("webhook_dlq_depth")); + assert!(output.contains("webhook_retries_total")); } #[test] diff --git a/src/webhook.rs b/src/webhook.rs new file mode 100644 index 0000000..12c237e --- /dev/null +++ b/src/webhook.rs @@ -0,0 +1,695 @@ +use std::{ + collections::VecDeque, + sync::Arc, + time::{Duration, Instant, SystemTime, UNIX_EPOCH}, +}; + +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use tokio::sync::Mutex; + +use crate::{event::Event, metrics::MetricsRegistry}; + +const MAX_DLQ_DEPTH: usize = 10_000; + +/// The outbound payload delivered to each webhook URL. +/// +/// Receivers can use `idempotency_key` to safely deduplicate retried deliveries โ€” +/// the key is derived from the Soroban transaction hash and event index, so it is +/// stable across replays of the same on-chain event. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct WebhookPayload { + pub event_id: String, + pub event_type: String, + /// Stable deduplication token: `contract:::::`. + pub idempotency_key: String, + pub sequence: u64, + pub timestamp: DateTime, + pub aggregate_id: String, + pub actor: String, + pub data: serde_json::Value, + #[serde(skip_serializing_if = "Option::is_none")] + pub metadata: Option, +} + +impl From<&Event> for WebhookPayload { + fn from(event: &Event) -> Self { + Self { + event_id: event.id.clone(), + event_type: event.event_type.clone(), + idempotency_key: event.idempotency_key.clone(), + sequence: event.sequence, + timestamp: event.timestamp, + aggregate_id: event.aggregate_id.clone(), + actor: event.actor.clone(), + data: event.data.clone(), + metadata: event.metadata.clone(), + } + } +} + +/// A delivery that exhausted all retries and is queued for manual replay. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct DeadLetterEntry { + pub url: String, + pub payload: WebhookPayload, + pub attempts: u32, + pub last_error: String, + pub failed_at: DateTime, +} + +/// Configuration for [`WebhookDispatcher`]. +#[derive(Debug, Clone)] +pub struct WebhookDispatcherConfig { + pub urls: Vec, + pub secret: Option, + pub max_retries: u32, + pub base_delay_ms: u64, + pub max_delay_ms: u64, + pub request_timeout_ms: u64, + pub jitter_enabled: bool, +} + +impl Default for WebhookDispatcherConfig { + fn default() -> Self { + Self { + urls: vec![], + secret: None, + max_retries: 5, + base_delay_ms: 200, + max_delay_ms: 30_000, + request_timeout_ms: 10_000, + jitter_enabled: true, + } + } +} + +/// Dispatches finalized events to all registered webhook URLs with exponential backoff +/// and a bounded dead-letter queue. +/// +/// ## Ordering +/// URLs are contacted sequentially in registration order. An event is attempted against +/// every URL regardless of individual failures โ€” a URL that exhausts retries is +/// dead-lettered without blocking delivery to subsequent URLs. +/// +/// ## Idempotency +/// Each HTTP request carries `X-Idempotency-Key` derived from the event's transaction +/// hash and event index. Receivers can use this header to safely deduplicate retried +/// deliveries. +/// +/// ## Dead-letter queue +/// Failed deliveries are pushed to an in-memory bounded queue (max 10 000 entries). +/// Call [`WebhookDispatcher::drain_dlq`] to retrieve entries for manual replay. +pub struct WebhookDispatcher { + urls: Vec, + client: reqwest::Client, + secret: Option, + max_retries: u32, + base_delay_ms: u64, + max_delay_ms: u64, + jitter_enabled: bool, + metrics: Option>, + dlq: Arc>>, +} + +impl WebhookDispatcher { + pub fn new(config: WebhookDispatcherConfig, metrics: Option>) -> Self { + let client = reqwest::Client::builder() + .timeout(Duration::from_millis(config.request_timeout_ms)) + .build() + .unwrap_or_default(); + + Self { + urls: config.urls, + client, + secret: config.secret, + max_retries: config.max_retries, + base_delay_ms: config.base_delay_ms, + max_delay_ms: config.max_delay_ms, + jitter_enabled: config.jitter_enabled, + metrics, + dlq: Arc::new(Mutex::new(VecDeque::new())), + } + } + + /// Construct a dispatcher from application config. + pub fn from_app_config( + config: &crate::config::AppConfig, + metrics: Option>, + ) -> Self { + Self::new( + WebhookDispatcherConfig { + urls: config.webhook_urls.clone(), + secret: config.webhook_secret.clone(), + max_retries: config.webhook_max_retries, + base_delay_ms: config.webhook_retry_base_delay_ms, + max_delay_ms: config.webhook_retry_max_delay_ms, + request_timeout_ms: config.webhook_request_timeout_ms, + jitter_enabled: config.webhook_jitter_enabled, + }, + metrics, + ) + } + + /// Dispatch `event` to all configured URLs in registration order. + /// + /// Each URL is attempted independently. Failed deliveries are retried with exponential + /// backoff before being moved to the dead-letter queue. Processing always continues + /// to the next URL regardless of outcome. + pub async fn dispatch(&self, event: &Event) { + if self.urls.is_empty() { + return; + } + + let payload = WebhookPayload::from(event); + + for url in &self.urls { + self.deliver_with_retry(url, &payload).await; + } + } + + /// Spawn [`dispatch`](Self::dispatch) as a background task, releasing the caller immediately. + pub fn dispatch_background(self: Arc, event: Event) { + tokio::spawn(async move { + self.dispatch(&event).await; + }); + } + + async fn deliver_with_retry(&self, url: &str, payload: &WebhookPayload) { + let overall_start = Instant::now(); + let mut last_error = String::from("no attempts made"); + + for attempt in 0..=self.max_retries { + if attempt > 0 { + let delay = self.backoff_delay(attempt - 1); + tokio::time::sleep(Duration::from_millis(delay)).await; + + if let Some(ref m) = self.metrics { + m.increment_webhook_retry(); + } + } + + match self.send_once(url, payload).await { + Ok(()) => { + let latency = overall_start.elapsed().as_secs_f64(); + if let Some(ref m) = self.metrics { + m.record_webhook_delivery("success", latency); + } + return; + } + Err(e) => { + last_error = e.to_string(); + eprintln!( + "[webhook] attempt {}/{} failed url={} error={}", + attempt + 1, + self.max_retries + 1, + url, + last_error + ); + } + } + } + + eprintln!( + "[webhook] dead-lettering url={} after {} attempts", + url, + self.max_retries + 1 + ); + + let entry = DeadLetterEntry { + url: url.to_string(), + payload: payload.clone(), + attempts: self.max_retries + 1, + last_error, + failed_at: Utc::now(), + }; + + let dlq_depth = { + let mut dlq = self.dlq.lock().await; + if dlq.len() >= MAX_DLQ_DEPTH { + // Evict oldest entry when the queue is full. + dlq.pop_front(); + } + dlq.push_back(entry); + dlq.len() + }; + + if let Some(ref m) = self.metrics { + m.record_webhook_delivery("dead_lettered", overall_start.elapsed().as_secs_f64()); + m.set_webhook_dlq_depth(dlq_depth as i64); + } + } + + async fn send_once(&self, url: &str, payload: &WebhookPayload) -> anyhow::Result<()> { + let body = serde_json::to_string(payload)?; + + let mut builder = self + .client + .post(url) + .header("Content-Type", "application/json") + .header("X-Idempotency-Key", &payload.idempotency_key) + .header("X-Event-Id", &payload.event_id) + .header("X-Event-Type", &payload.event_type) + .body(body); + + if let Some(ref secret) = self.secret { + builder = builder.header("X-Webhook-Secret", secret); + } + + let response = builder.send().await?; + let status = response.status(); + + if status.is_success() { + Ok(()) + } else { + Err(anyhow::anyhow!("HTTP {}", status)) + } + } + + /// Compute exponential backoff delay for attempt `n` (0-indexed). + fn backoff_delay(&self, attempt: u32) -> u64 { + let exp = self.base_delay_ms.saturating_mul(1u64 << attempt.min(20)); + let capped = exp.min(self.max_delay_ms); + + if self.jitter_enabled { + let max_jitter = capped / 4; + capped.saturating_add(jitter_ms(max_jitter)) + } else { + capped + } + } + + /// Drain and return all dead-letter entries for manual replay. + /// + /// After draining, the DLQ depth metric is reset to zero. + pub async fn drain_dlq(&self) -> Vec { + let mut dlq = self.dlq.lock().await; + let entries: Vec<_> = dlq.drain(..).collect(); + + if let Some(ref m) = self.metrics { + m.set_webhook_dlq_depth(0); + } + + entries + } + + /// Current number of entries in the dead-letter queue. + pub async fn dlq_depth(&self) -> usize { + self.dlq.lock().await.len() + } +} + +/// Compute a jitter value in [0, max_ms) using sub-millisecond wall-clock noise. +fn jitter_ms(max_ms: u64) -> u64 { + if max_ms == 0 { + return 0; + } + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .subsec_nanos() as u64; + nanos % max_ms +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Arc; + use wiremock::matchers::{header, method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + fn make_event() -> Event { + Event::new( + "doc-1".to_string(), + crate::event::EVENT_DOCUMENT_REGISTERED.to_string(), + serde_json::json!({"issuer": "addr1"}), + "issuer-addr".to_string(), + ) + .with_idempotency_key("contract:tx1:100:0:doc-1:DocumentRegistered") + } + + // โ”€โ”€ Happy-path delivery โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + + #[tokio::test] + async fn dispatch_sends_to_all_urls_in_order() { + let server1 = MockServer::start().await; + let server2 = MockServer::start().await; + + Mock::given(method("POST")) + .and(path("/")) + .respond_with(ResponseTemplate::new(200)) + .mount(&server1) + .await; + + Mock::given(method("POST")) + .and(path("/")) + .respond_with(ResponseTemplate::new(200)) + .mount(&server2) + .await; + + let config = WebhookDispatcherConfig { + urls: vec![server1.uri(), server2.uri()], + max_retries: 0, + ..Default::default() + }; + + let dispatcher = WebhookDispatcher::new(config, None); + dispatcher.dispatch(&make_event()).await; + + assert_eq!(server1.received_requests().await.unwrap().len(), 1); + assert_eq!(server2.received_requests().await.unwrap().len(), 1); + } + + #[tokio::test] + async fn dispatch_sends_idempotency_key_header() { + let server = MockServer::start().await; + + Mock::given(method("POST")) + .and(path("/")) + .and(header( + "x-idempotency-key", + "contract:tx1:100:0:doc-1:DocumentRegistered", + )) + .respond_with(ResponseTemplate::new(200)) + .mount(&server) + .await; + + let config = WebhookDispatcherConfig { + urls: vec![server.uri()], + max_retries: 0, + ..Default::default() + }; + + let dispatcher = WebhookDispatcher::new(config, None); + dispatcher.dispatch(&make_event()).await; + + assert_eq!(server.received_requests().await.unwrap().len(), 1); + } + + #[tokio::test] + async fn dispatch_sends_event_type_header() { + let server = MockServer::start().await; + + Mock::given(method("POST")) + .and(path("/")) + .and(header("x-event-type", "DocumentRegistered")) + .respond_with(ResponseTemplate::new(200)) + .mount(&server) + .await; + + let config = WebhookDispatcherConfig { + urls: vec![server.uri()], + max_retries: 0, + ..Default::default() + }; + + let dispatcher = WebhookDispatcher::new(config, None); + dispatcher.dispatch(&make_event()).await; + + assert_eq!(server.received_requests().await.unwrap().len(), 1); + } + + #[tokio::test] + async fn dispatch_sends_webhook_secret_header_when_configured() { + let server = MockServer::start().await; + + Mock::given(method("POST")) + .and(path("/")) + .and(header("x-webhook-secret", "my-secret")) + .respond_with(ResponseTemplate::new(200)) + .mount(&server) + .await; + + let config = WebhookDispatcherConfig { + urls: vec![server.uri()], + secret: Some("my-secret".to_string()), + max_retries: 0, + ..Default::default() + }; + + let dispatcher = WebhookDispatcher::new(config, None); + dispatcher.dispatch(&make_event()).await; + + assert_eq!(server.received_requests().await.unwrap().len(), 1); + } + + #[tokio::test] + async fn dispatch_noop_with_no_urls() { + let config = WebhookDispatcherConfig { + urls: vec![], + ..Default::default() + }; + + let dispatcher = WebhookDispatcher::new(config, None); + dispatcher.dispatch(&make_event()).await; + assert_eq!(dispatcher.dlq_depth().await, 0); + } + + // โ”€โ”€ Dead-letter queue โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + + #[tokio::test] + async fn http_error_response_goes_to_dlq() { + let server = MockServer::start().await; + + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(500)) + .mount(&server) + .await; + + let config = WebhookDispatcherConfig { + urls: vec![server.uri()], + max_retries: 0, + ..Default::default() + }; + + let dispatcher = WebhookDispatcher::new(config, None); + dispatcher.dispatch(&make_event()).await; + + assert_eq!(dispatcher.dlq_depth().await, 1); + } + + #[tokio::test] + async fn dlq_entry_preserves_url_and_payload() { + let server = MockServer::start().await; + + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(503)) + .mount(&server) + .await; + + let config = WebhookDispatcherConfig { + urls: vec![server.uri()], + max_retries: 0, + ..Default::default() + }; + + let event = make_event(); + let dispatcher = WebhookDispatcher::new(config, None); + dispatcher.dispatch(&event).await; + + let entries = dispatcher.drain_dlq().await; + assert_eq!(entries.len(), 1); + assert_eq!(entries[0].payload.event_id, event.id); + assert_eq!(entries[0].payload.idempotency_key, event.idempotency_key); + assert_eq!(entries[0].attempts, 1); + } + + #[tokio::test] + async fn dead_letter_does_not_skip_subsequent_urls() { + let good_server = MockServer::start().await; + let bad_server = MockServer::start().await; + + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(500)) + .mount(&bad_server) + .await; + + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200)) + .mount(&good_server) + .await; + + let config = WebhookDispatcherConfig { + urls: vec![bad_server.uri(), good_server.uri()], + max_retries: 0, + ..Default::default() + }; + + let dispatcher = WebhookDispatcher::new(config, None); + dispatcher.dispatch(&make_event()).await; + + assert_eq!(dispatcher.dlq_depth().await, 1); + assert_eq!(good_server.received_requests().await.unwrap().len(), 1); + } + + #[tokio::test] + async fn drain_dlq_clears_entries_and_returns_them() { + let server = MockServer::start().await; + + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(500)) + .mount(&server) + .await; + + let config = WebhookDispatcherConfig { + urls: vec![server.uri()], + max_retries: 0, + ..Default::default() + }; + + let dispatcher = WebhookDispatcher::new(config, None); + dispatcher.dispatch(&make_event()).await; + + assert_eq!(dispatcher.dlq_depth().await, 1); + let entries = dispatcher.drain_dlq().await; + assert_eq!(entries.len(), 1); + assert_eq!(dispatcher.dlq_depth().await, 0); + } + + // โ”€โ”€ Metrics โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + + #[tokio::test] + async fn successful_delivery_records_metrics() { + let server = MockServer::start().await; + + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200)) + .mount(&server) + .await; + + let metrics = MetricsRegistry::arc(); + let config = WebhookDispatcherConfig { + urls: vec![server.uri()], + max_retries: 0, + ..Default::default() + }; + + let dispatcher = WebhookDispatcher::new(config, Some(Arc::clone(&metrics))); + dispatcher.dispatch(&make_event()).await; + + let output = metrics.render(); + assert!(output.contains("webhook_deliveries_total")); + assert!(output.contains(r#"status="success""#)); + assert!(output.contains("webhook_delivery_latency_seconds")); + } + + #[tokio::test] + async fn dead_lettered_delivery_records_dlq_metric() { + let server = MockServer::start().await; + + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(500)) + .mount(&server) + .await; + + let metrics = MetricsRegistry::arc(); + let config = WebhookDispatcherConfig { + urls: vec![server.uri()], + max_retries: 0, + ..Default::default() + }; + + let dispatcher = WebhookDispatcher::new(config, Some(Arc::clone(&metrics))); + dispatcher.dispatch(&make_event()).await; + + let output = metrics.render(); + assert!(output.contains("webhook_dlq_depth")); + assert!(output.contains(r#"status="dead_lettered""#)); + } + + #[tokio::test] + async fn retry_increments_retry_metric() { + let server = MockServer::start().await; + + // First response fails, second succeeds. + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(500)) + .up_to_n_times(1) + .mount(&server) + .await; + + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200)) + .mount(&server) + .await; + + let metrics = MetricsRegistry::arc(); + let config = WebhookDispatcherConfig { + urls: vec![server.uri()], + max_retries: 2, + base_delay_ms: 1, + jitter_enabled: false, + ..Default::default() + }; + + let dispatcher = WebhookDispatcher::new(config, Some(Arc::clone(&metrics))); + dispatcher.dispatch(&make_event()).await; + + let output = metrics.render(); + assert!(output.contains("webhook_retries_total")); + // Event was ultimately delivered, not dead-lettered. + assert_eq!(dispatcher.dlq_depth().await, 0); + } + + // โ”€โ”€ Backoff โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + + #[test] + fn backoff_delay_doubles_each_attempt() { + let config = WebhookDispatcherConfig { + base_delay_ms: 100, + max_delay_ms: 30_000, + jitter_enabled: false, + ..Default::default() + }; + let d = WebhookDispatcher::new(config, None); + + assert_eq!(d.backoff_delay(0), 100); + assert_eq!(d.backoff_delay(1), 200); + assert_eq!(d.backoff_delay(2), 400); + assert_eq!(d.backoff_delay(3), 800); + } + + #[test] + fn backoff_delay_is_capped_at_max_delay() { + let config = WebhookDispatcherConfig { + base_delay_ms: 100, + max_delay_ms: 1_000, + jitter_enabled: false, + ..Default::default() + }; + let d = WebhookDispatcher::new(config, None); + + assert!(d.backoff_delay(20) <= 1_000); + } + + #[test] + fn backoff_delay_with_jitter_stays_above_base() { + let config = WebhookDispatcherConfig { + base_delay_ms: 100, + max_delay_ms: 30_000, + jitter_enabled: true, + ..Default::default() + }; + let d = WebhookDispatcher::new(config, None); + + // With jitter the result is >= base (capped) and <= capped + capped/4. + let delay = d.backoff_delay(0); + assert!(delay >= 100); + assert!(delay <= 125); // 100 + 100/4 + } + + // โ”€โ”€ Payload โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + + #[test] + fn webhook_payload_carries_all_event_fields() { + let event = make_event(); + let payload = WebhookPayload::from(&event); + + assert_eq!(payload.event_id, event.id); + assert_eq!(payload.event_type, event.event_type); + assert_eq!(payload.idempotency_key, event.idempotency_key); + assert_eq!(payload.sequence, event.sequence); + assert_eq!(payload.aggregate_id, event.aggregate_id); + assert_eq!(payload.actor, event.actor); + } +} From 026181203653ae040d91ea758d10d8387de912d4 Mon Sep 17 00:00:00 2001 From: feyishola Date: Sun, 28 Jun 2026 17:26:46 +0100 Subject: [PATCH 2/2] unit test fixed --- src/webhook.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/webhook.rs b/src/webhook.rs index 12c237e..63157c7 100644 --- a/src/webhook.rs +++ b/src/webhook.rs @@ -1,7 +1,9 @@ use std::{ collections::VecDeque, + string::{String, ToString}, sync::Arc, time::{Duration, Instant, SystemTime, UNIX_EPOCH}, + vec::Vec, }; use chrono::{DateTime, Utc};