diff --git a/crates/ingest/Cargo.toml b/crates/ingest/Cargo.toml index 8ebffc3..ae6064d 100644 --- a/crates/ingest/Cargo.toml +++ b/crates/ingest/Cargo.toml @@ -27,6 +27,4 @@ dotenvy = "0.15" axum.workspace = true octo-webhooks.workspace = true proptest.workspace = true -# Mock Horizon server for the payments / resume-replay tests. Pinned for the same MSRV reason -# as crates/api: newer 0.6.x needs edition2024 (Rust >= 1.85); toolchain is pinned to 1.84.1. -wiremock = "=0.6.2" +sqlx.workspace = true diff --git a/crates/ingest/tests/webhook_e2e_tests.rs b/crates/ingest/tests/webhook_e2e_tests.rs index f91432b..0c88511 100644 --- a/crates/ingest/tests/webhook_e2e_tests.rs +++ b/crates/ingest/tests/webhook_e2e_tests.rs @@ -2,13 +2,14 @@ //! locally-hosted sink. Requires Postgres via `DATABASE_URL`. use axum::extract::State; -use axum::http::HeaderMap; +use axum::http::{HeaderMap, StatusCode}; use axum::routing::post; use axum::Router; use octo_ingest::horizon::PaymentRecord; use octo_ingest::Ingestor; use octo_store::{NewWallet, Store}; -use octo_webhooks::{sign, WebhookSender}; +use octo_webhooks::{sign, Event, WebhookSender}; +use std::sync::atomic::{AtomicU32, Ordering}; use std::sync::{Arc, Mutex, Once}; use uuid::Uuid; @@ -146,3 +147,138 @@ async fn deposit_fires_signed_webhook() { assert_eq!(json["data"]["amount_stroops"], 35_000_000); assert_eq!(json["data"]["attributed"], false); } + +/// A sink that always responds with a server error, forcing the sender through its full retry +/// budget before giving up. +async fn always_fails_sink(State(hits): State>) -> StatusCode { + hits.fetch_add(1, Ordering::SeqCst); + StatusCode::INTERNAL_SERVER_ERROR +} + +/// `WebhookSender::dispatch` loops over every active endpoint for a wallet sequentially. A +/// endpoint that always fails must not starve or block delivery to a healthy sibling endpoint +/// registered on the same wallet. +#[tokio::test] +async fn dispatch_delivers_to_healthy_endpoint_despite_sibling_failure() { + let Some(url) = database_url() else { + eprintln!("SKIPPED: set DATABASE_URL"); + return; + }; + // Allow the test to deliver to localhost sinks. + std::env::set_var("OCTO_ALLOW_LOCAL_WEBHOOKS", "1"); + + let store = Store::connect(&url).await.expect("connect"); + store.migrate().await.expect("migrate"); + + // A sink that always succeeds immediately. + let healthy_captured: Shared = Arc::new(Mutex::new(None)); + let healthy_app = Router::new() + .route("/hook", post(sink)) + .with_state(healthy_captured.clone()); + let healthy_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let healthy_addr = healthy_listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(healthy_listener, healthy_app).await.unwrap(); + }); + + // A sink that always fails (500s every attempt). + let failing_hits: Arc = Arc::new(AtomicU32::new(0)); + let failing_app = Router::new() + .route("/hook", post(always_fails_sink)) + .with_state(failing_hits.clone()); + let failing_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let failing_addr = failing_listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(failing_listener, failing_app).await.unwrap(); + }); + + // One wallet, two sibling endpoints: one doomed, one healthy. + let wallet = store + .create_wallet(NewWallet { + network: "testnet", + stellar_account_g: &format!("{BASE}-{}", Uuid::new_v4().simple()), + sealed_ciphertext: b"ct", + sealed_nonce: b"n", + sealed_salt: b"s", + sealed_scheme: 1, + label: None, + user_id: None, + description: None, + }) + .await + .unwrap(); + + let healthy_secret = "healthy-secret-123"; + let failing_ep = store + .create_webhook_endpoint( + wallet.id, + &format!("http://{failing_addr}/hook"), + "failing-secret-456", + ) + .await + .unwrap(); + let healthy_ep = store + .create_webhook_endpoint( + wallet.id, + &format!("http://{healthy_addr}/hook"), + healthy_secret, + ) + .await + .unwrap(); + + let sender = WebhookSender::new(store.clone()); + let event = Event { + event_type: "deposit.created".into(), + data: serde_json::json!({ "amount_stroops": 1 }), + }; + + // `active_webhook_endpoints` has no defined ordering, so this must hold regardless of which + // endpoint dispatch happens to process first. + let delivered = sender.dispatch(wallet.id, &event).await; + assert_eq!( + delivered, 1, + "the healthy sibling must still be delivered to despite the other endpoint always failing" + ); + + // The healthy sink actually received the signed payload. + for _ in 0..50 { + if healthy_captured.lock().unwrap().is_some() { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + } + let cap = healthy_captured + .lock() + .unwrap() + .clone() + .expect("healthy endpoint was delivered to"); + let sig = cap.signature.expect("signature header present"); + assert!( + sign::verify(healthy_secret.as_bytes(), &cap.body, &sig), + "webhook signature must verify with the healthy endpoint's own secret" + ); + + // The failing endpoint was actually attempted, i.e. it was not skipped, only that it never + // succeeded. + assert!( + failing_hits.load(Ordering::SeqCst) >= 1, + "the failing endpoint must still have been attempted despite always erroring" + ); + + // Both endpoints get their own `webhook_deliveries` row with the correct respective outcome. + let failing_status: String = + sqlx::query_scalar("SELECT status FROM webhook_deliveries WHERE endpoint_id = $1") + .bind(failing_ep.id) + .fetch_one(store.pool()) + .await + .unwrap(); + assert_eq!(failing_status, "failed"); + + let healthy_status: String = + sqlx::query_scalar("SELECT status FROM webhook_deliveries WHERE endpoint_id = $1") + .bind(healthy_ep.id) + .fetch_one(store.pool()) + .await + .unwrap(); + assert_eq!(healthy_status, "delivered"); +}