webhooksmith

Webhook delivery for Rust backed by Postgres or SQLite. HMAC-SHA256 signing, automatic retry with exponential backoff, dead letter queue, and event type filtering. Postgres backend: full transactional outbox (atomic writes). SQLite backend: persistent delivery without Postgres.

---

Crates

Crate	Description
webhooksmith	Core engine — sending, delivery worker, DLQ, monitoring. Supports Postgres and SQLite.
webhooksmith-axum	Axum extractor for verifying incoming webhooks + admin HTTP router
webhooksmith-actix	Actix-web extractor for verifying incoming webhooks

---

How it works

1. Register partner webhook endpoints in Postgres or SQLite.
2. Call engine.send() — the event is persisted atomically.
3. The background worker picks it up and POSTs it with an HMAC-SHA256 signature.
4. On failure: exponential backoff → dead letter queue → manual retry.

No Redis. No queuing service. Just your existing database.

flowchart LR
    A[Webhook event created<br/>engine.send / send_in_tx] --> B[(Outbox table<br/>transactional write)]
    B --> C[Background dispatcher<br/>engine.run]
    C --> D{HTTP delivery attempt<br/>HMAC-SHA256 signed}
    D -->|2xx response| E[Delivered]
    D -->|failure or timeout| F{Retries exhausted?}
    F -->|no| G[Exponential backoff<br/>with jitter]
    G --> C
    F -->|yes| H[[Dead letter queue]]
    H -->|manual retry_dead| C

Loading

---

Quick start

Postgres (default — best for production, transactional outbox):

[dependencies]
webhooksmith = "0.2"
tokio = { version = "1", features = ["full"] }
serde_json = "1"

SQLite (desktop apps, CLI tools, embedded, no Postgres needed):

[dependencies]
webhooksmith = { version = "0.2", features = ["sqlite"] }
tokio = { version = "1", features = ["full"] }
serde_json = "1"

With SQLite, replace WebhookEngine with SqliteEngine — same API:

use webhooksmith::SqliteEngine;
let engine = SqliteEngine::new("sqlite:webhooks.db").await?;
engine.migrate().await?;

SQLite transactional outbox: SqliteEngine::send_in_tx() provides true atomicity. The event is written on the transaction's connection — rollback removes it, commit persists it. Both backends support the full transactional outbox pattern.

use webhooksmith::WebhookEngine;
use serde_json::json;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let engine = WebhookEngine::builder()
        .database_url("postgres://user:pass@localhost/mydb")
        .build()
        .await?;

    engine.migrate().await?;  // creates 3 tables, safe on every startup

    let endpoint = engine
        .register("https://partner.example.com/webhooks", "your-signing-secret-32chars")
        .await?;

    // Send a webhook
    engine.send("order.created", json!({"id": 1001}), endpoint.id).await?;

    // Start background delivery worker
    engine.run().await;
}

---

Transactional outbox

Write business data and webhook event in the same transaction — neither exists without the other:

let mut tx = engine.pool().begin().await?;

sqlx::query!("INSERT INTO orders (id) VALUES ($1)", order_id)
    .execute(&mut *tx).await?;

engine.send_in_tx("order.created", json!({"id": order_id}), endpoint.id, &mut tx).await?;

tx.commit().await?;  // webhook only queued if this succeeds

Fan-out to all endpoints at once, atomically:

let mut tx = engine.pool().begin().await?;
engine.broadcast_in_tx("order.created", json!({"id": 1001}), &mut tx).await?;
tx.commit().await?;

---

Key properties

Property	Detail
Transactional outbox	Business data and webhook event written atomically — no phantom events, no silent drops (Postgres + SQLite)
Circuit breaker	After 5 consecutive failures, endpoint paused with exponential backoff (5→10→20 min, capped at 320)
Idempotency keys	send_idempotent("order.created", payload, endpoint_id, "order-1001") — safe to retry, never duplicates
Event type filtering	Endpoints subscribe to "order.*" or "payment.captured" — broadcast() routes automatically
Admin HTTP API	webhooksmith_axum::admin(engine) — stats, endpoint list, DLQ inspection, bulk retry
Graceful shutdown	run_graceful(ctrl_c_signal) — current batch drains before exit (Kubernetes-safe)
SSRF protection	Private IPs, loopback, link-local blocked at registration AND at delivery (DNS rebinding protection)
HMAC-SHA256 signing	Constant-time verification, replay protection, Svix-compatible format
Dead letter queue	Events that exhaust retries are inspectable and requeueable, not silently dropped
Multi-backend	Postgres (full transactional outbox) or SQLite (no Postgres required)
Multi-framework receiver	Verify incoming webhooks in axum (webhooksmith-axum) or actix-web (webhooksmith-actix)

---

All APIs at a glance

Sending: send · send_in_tx · send_idempotent · send_idempotent_in_tx broadcast · broadcast_in_tx · broadcast_idempotent

Endpoints: register · register_with · endpoint · list_endpoints · list_endpoints_paged update_endpoint · enable_endpoint · disable_endpoint · delete_endpoint set_event_filter · clear_event_filter

Worker: run · run_graceful · run_once

Events & monitoring: event · events_by_status · events_global · queue_stats · delivery_log

DLQ: dead_events · dead_events_paged · retry_dead · retry_all_dead

Cleanup & ops: cleanup_delivered · cleanup_dead · recover_stuck_deliveries

---

All sending methods

Method	What it does
send(event_type, payload, endpoint_id)	Send to one endpoint
send_in_tx(event_type, payload, endpoint_id, &mut tx)	Send to one endpoint, inside your transaction
send_idempotent(event_type, payload, endpoint_id, key)	Send, deduplicated by key — safe to retry
send_idempotent_in_tx(event_type, payload, endpoint_id, key, &mut tx)	Idempotent + transactional
broadcast(event_type, payload)	Fan-out to ALL enabled endpoints
broadcast_in_tx(event_type, payload, &mut tx)	Fan-out inside your transaction
broadcast_idempotent(event_type, payload, key)	Fan-out, deduplicated per endpoint by key

Validation rules:

• event_type must be non-empty, non-whitespace, ≤ 256 bytes, no control characters
• payload must be valid JSON, ≤ 1 MB

Return value (WebhookEvent):

WebhookEvent {
    id: Uuid,                           // unique event ID
    endpoint_id: Uuid,                  // which endpoint this is for
    event_type: "order.created",
    payload: {"id": 1001, "total": 49.99},
    status: Pending,                    // Pending | Delivering | Delivered | Failed | Dead
    attempts: 0,
    scheduled_at: "2026-01-01T00:00:00Z",
    delivering_since: None,
    idempotency_key: None,
    created_at: "2026-01-01T00:00:00Z",
}

---

Event type filtering

Each endpoint can subscribe to specific event types. broadcast() routes automatically — only matching endpoints receive the event.

// Subscribe to order events only, using the NewEndpoint builder
engine.register_with(
    webhooksmith::NewEndpoint::new("https://orders.example.com/hook", "your-signing-secret-32chars")
        .events(["order.*"]),
).await?;

// broadcast() routes — payment.captured does NOT go to order-only endpoints
engine.broadcast("order.created", payload).await?;

// Update filter at any time
engine.set_event_filter(ep_id, vec!["order.*".into(), "payment.captured".into()]).await?;
engine.clear_event_filter(ep_id).await?;  // back to receiving all

Pattern rules:

Pattern	Matches
"order.created"	Exactly "order.created"
"order.*"	Any event starting with "order." (e.g. "order.created", "order.cancelled")
"*"	Everything
None (no filter)	Everything — default, backward compatible
Some(vec![]) (empty)	Nothing

send() to a specific endpoint always delivers — the filter only applies to broadcast().

---

Idempotency keys

Protect against double-sends when your code retries on network errors. The same key for the same endpoint always returns the same event, no matter how many times you call it.

// First call: creates the event
let ev1 = engine.send_idempotent("order.created", payload.clone(), ep.id, "order-1001").await?;

// Second call (e.g. after a retry): returns the SAME event
let ev2 = engine.send_idempotent("order.created", payload.clone(), ep.id, "order-1001").await?;

assert_eq!(ev1.id, ev2.id); // same event, not a duplicate

Key is scoped per (endpoint_id, key). The same key is independent across different endpoints, which makes broadcast_idempotent safe to call multiple times.

---

Running the worker

// Blocks forever — put this at the end of main()
engine.run().await;

// Graceful shutdown — current batch drains before exit
engine.run_graceful(async {
    tokio::signal::ctrl_c().await.ok();
}).await;

// One cycle — useful for testing or cron-style invocation
let delivered_count: usize = engine.run_once().await?;

What happens in each cycle:

1. Reset events stuck in delivering state for longer than stuck_timeout (crash recovery)
2. Claim a batch of due events with SELECT FOR UPDATE SKIP LOCKED
3. Deliver each event concurrently via HTTP POST with HMAC signature
4. Record success or failure; schedule retry or move to DLQ

---

Endpoint management

// Register
let ep = engine.register("https://partner.com/hooks", "secret-32chars-min").await?;

// Register with full config, using the NewEndpoint builder
let ep = engine.register_with(
    webhooksmith::NewEndpoint::new("https://partner.com/hooks", "secret-32chars-min")
        .description("Partner A")
        .max_attempts(10)       // retries before DLQ (default: 10, min: 1)
        .initial_delay_ms(1000), // first retry delay ms (default: 1000, min: 1)
).await?;

// Fetch one
let ep: Option<Endpoint> = engine.endpoint(ep.id).await?;

// List all (ordered by created_at)
let all: Vec<Endpoint> = engine.list_endpoints().await?;

// Paginated list
let page: Vec<Endpoint> = engine.list_endpoints_paged(20, 0).await?; // limit=20, offset=0

// Update (only fields you set are changed)
let updated = engine.update_endpoint(ep.id, webhooksmith::UpdateEndpoint {
    url: Some("https://new-partner.com/hooks".into()),
    max_attempts: Some(5),
    ..Default::default()
}).await?;

// Enable / disable (stops the worker from delivering to this endpoint)
engine.disable_endpoint(ep.id).await?;
engine.enable_endpoint(ep.id).await?;

// Delete (cascade-deletes all events for this endpoint)
engine.delete_endpoint(ep.id).await?;

Validation rules:

• url must be http:// or https:// and not point to a private/loopback/link-local address
• signing_secret must be ≥ 16 characters
• max_attempts must be ≥ 1
• initial_delay_ms must be ≥ 1

Endpoint fields:

Endpoint {
    id: Uuid,
    url: "https://partner.com/hooks",
    signing_secret: "your-secret",
    description: Some("Partner A"),
    enabled: true,
    max_attempts: 10,
    initial_delay_ms: 1000,
    event_filter: Some(["order.*"]),    // None = receive all
    consecutive_failures: 0,            // circuit breaker counter
    circuit_open_until: None,           // Some(DateTime) when circuit is open
    created_at: "2026-01-01T00:00:00Z",
    updated_at: "2026-01-01T00:00:00Z",
}

---

Monitoring & queue operations

// Count events by status — one DB query
let stats: QueueStats = engine.queue_stats().await?;
// QueueStats { pending: 42, delivering: 3, failed: 1, dead: 0, delivered: 1500 }

// Events for one endpoint with a specific status (paginated, newest first)
let failed: Vec<WebhookEvent> = engine
    .events_by_status(ep.id, EventStatus::Failed, 20, 0)
    .await?;

// Events across ALL endpoints (for global monitoring dashboards)
let all_failed: Vec<WebhookEvent> = engine
    .events_global(EventStatus::Failed, 50, 0)
    .await?;

// Full delivery attempt history for one event
let log: Vec<DeliveryAttempt> = engine.delivery_log(event.id).await?;
// DeliveryAttempt { attempted_at, response_status: Some(500), duration_ms: Some(243), error: Some("HTTP 500"), success: false }

// Get one event by ID
let ev: Option<WebhookEvent> = engine.event(event_id).await?;

---

Circuit breaker

After 5 consecutive delivery failures the endpoint's circuit opens. The worker skips it until the open window expires (5 min → 10 min → 20 min … capped at 320 min, doubling each time). A single successful delivery resets the counter. Manual DLQ retry (retry_all_dead) also resets it.

let ep = engine.endpoint(ep_id).await?.unwrap();
println!("failures: {}", ep.consecutive_failures);
println!("open until: {:?}", ep.circuit_open_until); // None = closed

---

Dead letter queue

When an event exceeds max_attempts failures it moves to status = Dead.

// List dead events for one endpoint (all, unordered)
let dead: Vec<WebhookEvent> = engine.dead_events(ep.id).await?;

// Paginated (newest first)
let page: Vec<WebhookEvent> = engine.dead_events_paged(ep.id, 20, 0).await?;

// Requeue one event (resets attempts to 0)
engine.retry_dead(event.id).await?;
// Returns HooksmithError::InvalidState if event is not dead
// Returns HooksmithError::EventNotFound if event doesn't exist

// Requeue all dead events for one endpoint — returns count requeued
let requeued: u64 = engine.retry_all_dead(ep.id).await?;

---

Cleanup

Delivered and dead events accumulate indefinitely unless cleaned up.

use std::time::Duration;

// Delete delivered events older than 7 days — returns count deleted
let removed = engine.cleanup_delivered(Duration::from_secs(7 * 86_400)).await?;

// Delete dead events older than 30 days — returns count deleted
let removed = engine.cleanup_dead(Duration::from_secs(30 * 86_400)).await?;

Only the specified status is touched — pending, delivering, and failed events are never deleted by cleanup.

---

Crash recovery

The worker calls this automatically every cycle. You can also call it manually for ops use:

// Reset events stuck in 'delivering' for > 120 seconds (returns count reset)
let reset = engine.recover_stuck_deliveries(Duration::from_secs(120)).await?;

---

Receiving webhooks

axum

[dependencies]
webhooksmith-axum = "0.2"

use axum::{Router, routing::post, http::StatusCode};
use webhooksmith_axum::{WebhookSecretLayer, VerifiedWebhook, TypedWebhook};
use serde::Deserialize;

#[derive(Deserialize)]
struct OrderCreated { order_id: u64 }

// VerifiedWebhook gives you the raw verified payload
async fn handle_raw(VerifiedWebhook(body): VerifiedWebhook) -> StatusCode {
    println!("{}: {:?}", body.event_type, body.body);
    StatusCode::OK
}

// TypedWebhook<T> deserializes the body for you
async fn handle_typed(TypedWebhook(order): TypedWebhook<OrderCreated>) -> StatusCode {
    println!("order: {}", order.order_id);
    StatusCode::OK
}

let app: Router = Router::new()
    .route("/webhooks", post(handle_raw))
    .route("/orders", post(handle_typed))
    .layer(WebhookSecretLayer::new("your-signing-secret"));

→ webhooksmith-axum README

actix-web

[dependencies]
webhooksmith-actix = "0.2"

use actix_web::{web, App, HttpResponse, Responder};
use webhooksmith_actix::{WebhookSecret, TypedWebhook};
use serde::Deserialize;

#[derive(Deserialize)]
struct OrderCreated { order_id: u64 }

async fn handle(webhook: TypedWebhook<OrderCreated>) -> impl Responder {
    println!("order: {}", webhook.payload.order_id);
    HttpResponse::Ok().finish()
}

App::new()
    .app_data(WebhookSecret::new("your-signing-secret"))
    .route("/webhooks", web::post().to(handle))

→ webhooksmith-actix README

Admin panel (axum)

use webhooksmith_axum::admin;

let app = Router::new()
    .nest("/admin", admin(Arc::clone(&engine)));
// GET /admin/stats · GET /admin/endpoints · GET /admin/dlq/:endpoint_id
// POST /admin/dlq/:endpoint_id/retry-all · GET /admin/metrics (Prometheus)

→ Admin API docs

---

What is sent to the endpoint

The worker sends an HTTP POST with:

POST https://partner.example.com/webhooks
Content-Type: application/json
x-hooksmith-signature: v1,<hex-encoded-hmac-sha256>
x-hooksmith-timestamp: 1735689600
x-hooksmith-event-id: 550e8400-e29b-41d4-a716-446655440000
x-hooksmith-event-type: order.created

{"id": 1001, "total": 49.99}

Signature format (Svix-compatible): The HMAC-SHA256 is computed over {timestamp}.{body} using your signing secret.

Verifying manually without the axum or actix-web crates:

use webhooksmith::signing;

fn verify_request(secret: &str, timestamp_header: &str, signature_header: &str, body: &[u8]) -> bool {
    let timestamp: i64 = timestamp_header.parse().unwrap_or(0);
    signing::verify(secret, timestamp, body, signature_header)
    // Returns false if: wrong secret, timestamp older than 5 minutes, or tampered body
}

---

Retry behaviour

Failed deliveries are retried with exponential backoff and full jitter:

Attempt	Max delay
1st retry	initial_delay_ms × 2 (default: 2s)
2nd retry	initial_delay_ms × 4 (default: 4s)
…	…
Any	Capped at 1 hour

After max_attempts failures (default: 10), the event moves to status = Dead and is not retried again automatically. Use retry_dead or retry_all_dead to requeue.

HTTP responses treated as failure: anything outside 2xx. 3xx redirects are explicitly not followed.

---

Configuration

let engine = WebhookEngine::builder()
    .database_url("postgres://user:pass@localhost/mydb")
    // OR: .pool(existing_sqlx_pool)

    // Worker
    .batch_size(50)                              // events per cycle (default: 50, min: 1)
    .poll_interval(Duration::from_millis(500))   // idle sleep (default: 500ms)
    .http_timeout(Duration::from_secs(30))       // per-request timeout (default: 30s)
    .stuck_timeout(Duration::from_secs(120))     // reaper threshold (default: 120s)
    // http_timeout MUST be < stuck_timeout (panics otherwise)

    // Postgres pool
    .max_connections(20)                         // pool size (default: 20)
    .acquire_timeout(Duration::from_secs(10))    // connection wait limit (default: 10s)

    // Development only — skips SSRF URL validation
    // .allow_insecure_urls()

    .build()
    .await?;

---

SSRF protection

The following URL targets are blocked at endpoint registration and at delivery time (DNS rebinding protection):

• Loopback: 127.x.x.x, ::1, localhost
• Private IPv4: 10.x, 172.16–31.x, 192.168.x
• Link-local: 169.254.x.x (AWS metadata endpoint), fe80::/10
• CGNAT: 100.64.0.0/10
• IPv6 unique local: fc00::/7
• IPv4-mapped IPv6: ::ffff:10.x.x.x, etc.

HTTP redirects are blocked at delivery time (the redirect target is not visited).

---

Error types

use webhooksmith::HooksmithError;

match result {
    Err(HooksmithError::Database(e))           => // sqlx database error
    Err(HooksmithError::Http(e))               => // reqwest HTTP error
    Err(HooksmithError::EndpointNotFound(id))  => // endpoint doesn't exist
    Err(HooksmithError::EventNotFound(id))     => // event doesn't exist
    Err(HooksmithError::InvalidState(id))      => // operation invalid for current state
    Err(HooksmithError::PayloadTooLarge(n, m)) => // payload n bytes exceeds m byte limit
    Err(HooksmithError::Config(msg))           => // validation error (URL, secret, etc.)
    Err(HooksmithError::Signing(msg))          => // HMAC error
}

---

Database schema

engine.migrate() creates three tables:

webhook_endpoints  -- registered delivery targets
webhook_events     -- outbound events, one row per (endpoint, event)
webhook_delivery_attempts  -- log of every HTTP call made

All tables use UUIDs as primary keys. Postgres uses TIMESTAMPTZ; SQLite stores timestamps as ISO 8601 TEXT. The webhook_events table has a partial index on (status, scheduled_at) for efficient worker queries.

---

How-to examples

docker compose up -d   # start Postgres

# Minimal setup — connect, register, send, deliver
cargo run --example basic -p webhooksmith

# Transactional outbox — atomic writes, rollback safety
cargo run --example outbox -p webhooksmith

# Full axum integration — engine in State, graceful shutdown
cargo run --example axum_integration -p webhooksmith

# Multi-tenant — per-customer endpoints, broadcast, idempotent sends
cargo run --example multi_tenant -p webhooksmith

# Monitoring — queue stats, DLQ alerts, retry failed events, cleanup
cargo run --example monitoring -p webhooksmith

# Full end-to-end demo with real axum receiver
cargo run --example demo -p webhooksmith

SQLite examples need no setup — just pass a file path or sqlite::memory:.

---

How-to guides

Guide	Topic
01 — Transactional outbox	Atomic business write + webhook event
02 — Circuit breaker	Per-endpoint failure protection
03 — Admin API	Stats, DLQ, bulk retry via HTTP
04 — Receiving webhooks	axum + actix-web integrations
05 — Event filtering	Per-endpoint event subscriptions
06 — DLQ and monitoring	Dead letter queue, cleanup, delivery log
07 — Graceful shutdown	Kubernetes-safe worker lifecycle

---

Running the tests

docker compose up -d
DATABASE_URL=postgres://webhooksmith:webhooksmith@localhost:5432/webhooksmith cargo test --features sqlite

276 tests covering unit, integration (real Postgres + real HTTP + in-memory SQLite), true end-to-end (real actix-web receiver over TCP), circuit breaker, admin API, adversarial, stress, bombardment, and event filter routing.

---

Requirements

• Rust 1.75+
• Postgres backend (default): Postgres 14+ (gen_random_uuid(), FOR UPDATE SKIP LOCKED, partial unique indexes, triggers)
• SQLite backend: SQLite 3.35+ (supports RETURNING; WAL mode enabled automatically)

---

See CHANGELOG.md for release history.

License

MIT OR Apache-2.0