From 5dba598450c02c8e782adda67e221e2400fa033f Mon Sep 17 00:00:00 2001 From: tongfengyuan <71140753@chinatelecom.cn> Date: Mon, 7 Sep 2026 18:42:52 +0800 Subject: [PATCH] feat(rwi): dedicated webhook runtime, event metrics, configurable retry and queue MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rebased onto current main, which already carries a fixed-count retry loop, the event-type-aware dedup key and full-body (never-truncated) delivery logging — this keeps those and layers the feature set on top: - dedicated tokio runtime for the RWI webhook handler (queue drain no longer competes with call/media workers), spawned via utils::rwi_webhook_spawn - configurable push retry: [proxy.locator_webhook] retries (default 0 = single attempt, hard cap 5) with exponential backoff (200 ms base); retryable = transport error / 5xx / 429 — supersedes main's fixed WEBHOOK_RETRY_COUNT(3)/500 ms outer loop with the richer policy inside send_payload - event pipeline metrics: rwi_event_queue_{size,current} gauges, rwi_events_{pushed,push_failed,retries}_total counters with an event_type label, and opt-in rwi_event_queue_latency_seconds histogram (gateway enqueue → handler dequeue, excludes the HTTP push; [proxy.locator_webhook] track_queue_latency) - configurable broadcast queue length: [proxy] rwi_webhook_channel_size (default 512) - docs: observability.md metric list + rwi_events_reference*.md config Existing retry tests updated for the config-driven policy (retries: Some(3) preserves the 3/4-request expectations); all 26 webhook tests pass. --- docs/observability.md | 13 ++ docs/rwi_events_reference.md | 36 ++++ docs/rwi_events_reference_en.md | 32 +++- src/app.rs | 5 +- src/bin/rustpbx.rs | 14 +- src/config.rs | 33 ++++ src/rwi/gateway.rs | 5 + src/rwi/webhook.rs | 313 +++++++++++++++++++++----------- src/utils.rs | 48 ++++- 9 files changed, 385 insertions(+), 114 deletions(-) diff --git a/docs/observability.md b/docs/observability.md index 8b7c05f1e..ce5a44c5b 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -183,6 +183,19 @@ inflate it, backlog does. | `rustpbx_transcription_latency_seconds` | Histogram | `language` | Transcription processing time | | `rustpbx_transcription_audio_seconds` | Histogram | `language` | Audio duration transcribed | +#### RWI Events + +| Metric | Type | Labels | Description | +|---|---|---|---| +| `rwi_event_enqueued_total` | Counter | `event_type` | Events pushed into the webhook queue by gateway dispatch | +| `rwi_events_pushed_total` | Counter | `event_type` | Events delivered with a 2xx response | +| `rwi_events_push_failed_total` | Counter | `event_type` | Pushes that errored or returned non-2xx | +| `rwi_events_push_retries_total` | Counter | `event_type` | Retry attempts after a failed push | +| `rwi_events_dropped_total` | Counter | - | Events lost to broadcast lag (consumer fell behind) | +| `rwi_event_queue_size` | Gauge | - | Webhook queue capacity (`[proxy] rwi_webhook_channel_size`) | +| `rwi_event_queue_current` | Gauge | - | Events currently queued (sampled every 5 s) | +| `rwi_event_queue_latency_seconds` | Histogram | `event_type` | Queueing wait (enqueued -> handler dequeued); opt-in via `[rwi_webhook] track_queue_latency` | + #### Routing | Metric | Type | Labels | Description | diff --git a/docs/rwi_events_reference.md b/docs/rwi_events_reference.md index a16dcad9a..1844503a9 100644 --- a/docs/rwi_events_reference.md +++ b/docs/rwi_events_reference.md @@ -43,6 +43,12 @@ Authorization: Bearer url = "https://myapp.example.com/rwi-events" timeout_ms = 5000 headers = { Authorization = "Bearer your-token" } +# 推送失败(传输错误、5xx、429)后的重试次数。其他 4xx 为永久失败,立即返回。 +# 退避时间从 200 ms 起指数递增。上限 5 次。 +retries = 2 +# 可选:在 rwi_event_queue_latency_seconds 直方图中统计事件排队延迟 +# (入队 -> 处理器出队)。默认关闭,需显式开启。 +track_queue_latency = true # 空 = 全部事件(推荐)。如需白名单过滤,请使用有效的事件类型。 # 注意:坐席状态是 "agent_state_changed"(旧的 "dn_state_changed" 已废弃移除); # 录音数据(下载 URL、文件大小)通过 "recording_metadata_available" 和 @@ -52,6 +58,36 @@ headers = { Authorization = "Bearer your-token" } events = [] ``` +| 字段 | 类型 | 默认值 | 说明 | +|------|------|--------|------| +| `url` | String | (必填) | 接收 POST 请求的 HTTP 端点 | +| `timeout_ms` | u64 | 5000 | HTTP 请求超时(毫秒,每次尝试) | +| `headers` | HashMap | (可选) | 每个请求携带的自定义 HTTP 头 | +| `events` | Vec\ | [](全部) | 事件类型白名单;为空转发全部事件 | +| `retries` | u32 | 0 | 推送失败后的重试次数(传输错误、5xx、429);硬上限 5;退避从 200 ms 起指数递增 | +| `track_queue_latency` | bool | false | 记录排队等待直方图 `rwi_event_queue_latency_seconds` | + +Webhook 处理器运行在专用的 tokio 运行时上,其 HTTP 推送不会与 SIP 运行时 +争抢资源。worker 数量与事件队列长度在 `[proxy]` 下配置: + +| 键 | 默认值 | 说明 | +|----|--------|------| +| `[proxy] rwi_webhook_worker_threads` | 2 | webhook 推送消费者的专用 tokio worker 数 | +| `[proxy] rwi_webhook_channel_size` | 512 | 事件队列长度(广播通道容量) | + +### Webhook 指标 + +| 指标 | 类型 | 标签 | 说明 | +|------|------|------|------| +| `rwi_event_enqueued_total` | Counter | `event_type` | 网关分发推入队列的事件数 | +| `rwi_events_pushed_total` | Counter | `event_type` | 收到 2xx 响应成功投递的事件数 | +| `rwi_events_push_failed_total` | Counter | `event_type` | 推送出错或返回非 2xx 的事件数 | +| `rwi_events_push_retries_total` | Counter | `event_type` | 推送失败后的重试次数 | +| `rwi_events_dropped_total` | Counter | - | 因队列积压被跳过的事件数 | +| `rwi_event_queue_size` | Gauge | - | 配置的队列容量 | +| `rwi_event_queue_current` | Gauge | - | 当前排队中的事件数(每 5 秒采样) | +| `rwi_event_queue_latency_seconds` | Histogram | `event_type` | 排队等待时长(入队 -> 处理器出队);通过 `track_queue_latency` 开启 | + --- ## 3. 信封格式 diff --git a/docs/rwi_events_reference_en.md b/docs/rwi_events_reference_en.md index 6f1c0f91b..7d1123d01 100644 --- a/docs/rwi_events_reference_en.md +++ b/docs/rwi_events_reference_en.md @@ -43,6 +43,12 @@ Or via query parameter: `GET /rwi/v1?token=` url = "https://myapp.example.com/rwi-events" timeout_ms = 5000 headers = { Authorization = "Bearer your-token" } +# Retries after a failed push (transport error, 5xx or 429). Other 4xx are +# permanent and return immediately. Backoff doubles from 200 ms. Hard cap 5. +retries = 2 +# Opt-in: track event queueing latency (enqueued -> handler dequeued) in the +# rwi_event_queue_latency_seconds histogram. Disabled by default. +track_queue_latency = true # empty = all events (recommended). To allow-list, use valid event types. # Note: agent status is "agent_state_changed" (the old "dn_state_changed" was # removed); recording data (download URL, file size) is delivered via @@ -56,9 +62,33 @@ events = [] | Field | Type | Default | Description | |-------|------|---------|-------------| | `url` | String | (required) | HTTP endpoint receiving POST requests | -| `timeout_ms` | u64 | 5000 | HTTP request timeout in milliseconds | +| `timeout_ms` | u64 | 5000 | HTTP request timeout in milliseconds (per attempt) | | `headers` | HashMap | (optional) | Custom HTTP headers sent with every request | | `events` | Vec\ | [] (all) | Event type whitelist; empty forwards all events | +| `retries` | u32 | 0 | Retries after a failed push (transport error, 5xx, 429); hard cap 5; exponential backoff from 200 ms | +| `track_queue_latency` | bool | false | Record the queueing-wait histogram `rwi_event_queue_latency_seconds` | + +The webhook handler runs on a dedicated tokio runtime so its HTTP push never +contends with the SIP runtime. The worker count and the event queue length +are configured under `[proxy]`: + +| Key | Default | Description | +|-----|---------|-------------| +| `[proxy] rwi_webhook_worker_threads` | 2 | Dedicated tokio workers for the webhook push consumer | +| `[proxy] rwi_webhook_channel_size` | 512 | Event queue length (broadcast channel capacity) | + +### Webhook Metrics + +| Metric | Type | Labels | Description | +|-------|------|--------|-------------| +| `rwi_event_enqueued_total` | Counter | `event_type` | Events pushed into the queue by gateway dispatch | +| `rwi_events_pushed_total` | Counter | `event_type` | Events delivered with a 2xx response | +| `rwi_events_push_failed_total` | Counter | `event_type` | Pushes that errored or returned non-2xx | +| `rwi_events_push_retries_total` | Counter | `event_type` | Retry attempts after a failed push | +| `rwi_events_dropped_total` | Counter | - | Events lost to queue lag (consumer fell behind) | +| `rwi_event_queue_size` | Gauge | - | Configured queue capacity | +| `rwi_event_queue_current` | Gauge | - | Events currently queued (sampled every 5 s) | +| `rwi_event_queue_latency_seconds` | Histogram | `event_type` | Queueing wait (enqueued -> handler dequeued); opt-in via `track_queue_latency` | --- diff --git a/src/app.rs b/src/app.rs index 9c4cee202..6abb71476 100644 --- a/src/app.rs +++ b/src/app.rs @@ -638,7 +638,10 @@ impl AppStateBuilder { if let Some(webhook_config) = config.rwi_webhook.clone() && let Some(gateway_ref) = core.rwi_gateway.clone() { - let webhook_tx = crate::rwi::webhook::start_rwi_webhook_handler(webhook_config); + let webhook_tx = crate::rwi::webhook::start_rwi_webhook_handler( + webhook_config, + config.proxy.rwi_webhook_channel_size, + ); let mut gw = gateway_ref.write(); gw.set_webhook_tx(webhook_tx); } diff --git a/src/bin/rustpbx.rs b/src/bin/rustpbx.rs index e189c73bd..cb2ed1cc4 100644 --- a/src/bin/rustpbx.rs +++ b/src/bin/rustpbx.rs @@ -335,10 +335,11 @@ fn main() -> Result<()> { // heavy RTP forwarding does not starve SIP timer/transaction tasks. let sip_workers = config.proxy.sip_worker_threads.max(1); let media_workers = config.proxy.media_worker_threads.max(1); + let rwi_webhook_workers = config.proxy.rwi_webhook_worker_threads.max(1); println!( - "SIP workers={} Media workers={}", - sip_workers, media_workers + "SIP workers={} Media workers={} RWI webhook workers={}", + sip_workers, media_workers, rwi_webhook_workers ); let media_runtime = tokio::runtime::Builder::new_multi_thread() @@ -353,6 +354,15 @@ fn main() -> Result<()> { // SIP runtime so high-concurrency recording cannot starve SIP timers. rustpbx::media::media_recorder::set_recorder_runtime(media_runtime.handle().clone()); + let rwi_webhook_runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(rwi_webhook_workers) + .thread_name("rwi-webhook") + .thread_stack_size(8 * 1024 * 1024) + .enable_all() + .build() + .map_err(|e| anyhow::anyhow!("Failed to build RWI webhook runtime: {}", e))?; + rustpbx::utils::set_rwi_webhook_runtime(rwi_webhook_runtime.handle().clone()); + let sip_runtime = tokio::runtime::Builder::new_multi_thread() .worker_threads(sip_workers) .thread_name("sip-worker") diff --git a/src/config.rs b/src/config.rs index ffdaf2a45..41207b519 100644 --- a/src/config.rs +++ b/src/config.rs @@ -930,6 +930,17 @@ pub struct LocatorWebhookConfig { pub events: Vec, pub headers: Option>, pub timeout_ms: Option, + /// Retries for the webhook HTTP push after a failed attempt (transport + /// error, 5xx or 429). 0 = single attempt (default). Exponential backoff + /// between attempts (200 ms base, doubling). + #[serde(default)] + pub retries: Option, + /// Track event queueing latency (gateway enqueued -> webhook handler + /// dequeued) in the `rwi_event_queue_latency_seconds` histogram. + /// Excludes the HTTP push itself. Disabled by default — opt in + /// explicitly. + #[serde(default)] + pub track_queue_latency: Option, } /// Global recovery for Step IVR when the external provider cannot continue. @@ -1139,6 +1150,18 @@ pub struct ProxyConfig { pub sip_worker_threads: usize, #[serde(default = "default_media_worker_threads")] pub media_worker_threads: usize, + /// Dedicated tokio worker threads for the RWI HTTP webhook push consumer. + /// Isolates the webhook's outbound HTTP (and any backpressure from a slow + /// router) from the SIP runtime shared by signalling, the HTTP route path + /// and the CDR saver. + #[serde(default = "default_rwi_webhook_worker_threads")] + pub rwi_webhook_worker_threads: usize, + /// RWI webhook event queue length: capacity of the broadcast channel + /// between the gateway and the webhook handler. When more than this many + /// events are queued, slow consumers skip ahead (Lagged) and the missed + /// events are counted as dropped. + #[serde(default = "default_rwi_webhook_channel_size")] + pub rwi_webhook_channel_size: usize, pub ws_handler: Option, pub ami_path: Option, pub rwi_path: Option, @@ -1331,6 +1354,14 @@ fn default_media_worker_threads() -> usize { if n > sip { n - sip } else { 1 } } +fn default_rwi_webhook_worker_threads() -> usize { + 2 +} + +fn default_rwi_webhook_channel_size() -> usize { + crate::rwi::webhook::WEBHOOK_CHANNEL_SIZE +} + fn default_auth_cache_size() -> usize { 10000 } @@ -1796,6 +1827,8 @@ impl Default for ProxyConfig { hold_music: None, sip_worker_threads: default_sip_worker_threads(), media_worker_threads: default_media_worker_threads(), + rwi_webhook_worker_threads: default_rwi_webhook_worker_threads(), + rwi_webhook_channel_size: default_rwi_webhook_channel_size(), } } } diff --git a/src/rwi/gateway.rs b/src/rwi/gateway.rs index 93bad0652..245bb1765 100644 --- a/src/rwi/gateway.rs +++ b/src/rwi/gateway.rs @@ -331,6 +331,11 @@ impl RwiGateway { fn fanout_webhook_tap(&self, entry: &EventCacheEntry) { if let Some(tx) = &self.webhook_tx { let _ = tx.send(entry.clone()); + metrics::counter!( + "rwi_event_enqueued_total", + "event_type" => entry.event.event_type + ) + .increment(1); } let _ = self.event_tap.send(entry.clone()); } diff --git a/src/rwi/webhook.rs b/src/rwi/webhook.rs index d5cd854c5..2e7aebda8 100644 --- a/src/rwi/webhook.rs +++ b/src/rwi/webhook.rs @@ -4,26 +4,22 @@ use anyhow::anyhow; use chrono::{DateTime, Utc}; use serde_json::json; use std::collections::{HashSet, VecDeque}; -use std::time::Duration; use tokio::sync::broadcast; use tracing::{debug, info, warn}; -/// Buffer size for the broadcast channel between gateway and webhook handler. -const WEBHOOK_CHANNEL_SIZE: usize = 512; +/// Default buffer size for the broadcast channel between gateway and webhook +/// handler. Overridable via [proxy] rwi_webhook_channel_size. +pub const WEBHOOK_CHANNEL_SIZE: usize = 512; /// Max number of recent (call_id, timestamp) pairs kept for dedup. const DEDUP_CACHE_SIZE: usize = 4096; -/// Idempotent retry policy: a failed delivery (transport error or non-2xx -/// status) is retried up to [`WEBHOOK_RETRY_COUNT`] times with -/// [`WEBHOOK_RETRY_INTERVAL_MS`] between attempts. Every attempt re-sends the -/// byte-identical payload (same `event_id`), so receivers can safely dedupe. -const WEBHOOK_RETRY_COUNT: u32 = 3; -const WEBHOOK_RETRY_INTERVAL_MS: u64 = 500; struct RwiWebhookSender { url: String, headers: std::collections::HashMap, allowed_events: Vec, client: reqwest::Client, + retries: u32, + track_queue_latency: bool, } impl RwiWebhookSender { @@ -35,6 +31,8 @@ impl RwiWebhookSender { allowed_events: config.events, client: crate::http_util::build_keepalive_client(Some(timeout), None) .unwrap_or_else(|_| reqwest::Client::new()), + retries: config.retries.unwrap_or(0), + track_queue_latency: config.track_queue_latency.unwrap_or(false), } } @@ -55,14 +53,64 @@ impl RwiWebhookSender { &self, payload: &serde_json::Value, body: &str, + event_type: &'static str, + ) -> Result { + // Attempts = 1 + retries. A retryable outcome is a transport error, + // a 5xx, or a 429; other 4xx are permanent and return immediately. + // Backoff doubles from 200 ms. Each attempt is bounded by the + // client's request timeout (`timeout_ms`, default 5 s). + let attempts = self.retries.min(MAX_PUSH_RETRIES) as usize + 1; + let mut attempt: usize = 0; + loop { + attempt += 1; + match self.send_once(payload, body).await { + Ok(record) => { + let status = record.status_code.unwrap_or(0); + let retryable = status == 429 || status >= 500; + if attempt >= attempts || !retryable { + return Ok(record); + } + warn!( + url = %self.url, + attempt, + attempts, + status_code = status, + "RWI webhook push failed, retrying" + ); + } + Err(e) => { + if attempt >= attempts { + return Err(e); + } + warn!( + url = %self.url, + attempt, + attempts, + error = %e, + "RWI webhook push errored, retrying" + ); + } + } + metrics::counter!( + "rwi_events_push_retries_total", + "event_type" => event_type + ) + .increment(1); + let backoff = PUSH_RETRY_BACKOFF_MS.saturating_mul(1 << (attempt - 1).min(5)); + tokio::time::sleep(std::time::Duration::from_millis(backoff)).await; + } + } + + async fn send_once( + &self, + payload: &serde_json::Value, + body: &str, ) -> Result { let start = std::time::Instant::now(); let mut req = self.client.post(&self.url).json(payload); for (key, value) in &self.headers { req = req.header(key, value); } - // The client is built with a connect/read timeout, so we don't wrap - // an additional timeout here. let resp = req .send() .await @@ -79,6 +127,12 @@ impl RwiWebhookSender { } } +/// Hard cap on configured webhook push retries (protects the dedicated +/// runtime's queue from unbounded redelivery backlogs). +const MAX_PUSH_RETRIES: u32 = 5; +/// Base backoff between webhook push retries (doubles per attempt). +const PUSH_RETRY_BACKOFF_MS: u64 = 200; + /// Captured metadata for a single webhook delivery attempt, used for /// structured observability logging. `body` carries the *complete* request /// payload — never truncated — so the sender-side log can serve as a @@ -96,9 +150,11 @@ pub struct WebhookCallRecord { /// Returns a `broadcast::Sender` that the gateway can use to send events. pub fn start_rwi_webhook_handler( config: LocatorWebhookConfig, + channel_size: usize, ) -> broadcast::Sender { - let (tx, rx) = broadcast::channel(WEBHOOK_CHANNEL_SIZE); - crate::utils::spawn(run_rwi_webhook_handler(config, rx)); + let (tx, rx) = broadcast::channel(channel_size.max(1)); + spawn_queue_metrics(tx.clone(), channel_size.max(1)); + crate::utils::rwi_webhook_spawn(run_rwi_webhook_handler(config, rx)); tx } @@ -108,9 +164,7 @@ pub fn start_rwi_webhook_handler( /// DISTINCT events for the same call can legitimately share one timestamp /// (observed in e2e: `call_created` vs `queue_joined` at session start) — /// a key without the type silently dropped one of them. -fn webhook_dedup_key( - entry: &EventCacheEntry, -) -> (String, DateTime, String) { +fn webhook_dedup_key(entry: &EventCacheEntry) -> (String, DateTime, String) { ( entry.call_id.clone(), entry.cached_at, @@ -118,6 +172,23 @@ fn webhook_dedup_key( ) } +/// Periodically export RWI webhook queue depth gauges: the channel's +/// capacity and the number of events currently queued (produced but not +/// yet seen by the handler). Slow-router backpressure shows up here as +/// `current` climbing toward `size`. +fn spawn_queue_metrics(tx: broadcast::Sender, size: usize) { + crate::utils::rwi_webhook_spawn(async move { + metrics::gauge!("rwi_event_queue_size").set(size as f64); + let mut interval = tokio::time::interval(std::time::Duration::from_secs(5)); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + interval.tick().await; // skip the immediate first tick + loop { + metrics::gauge!("rwi_event_queue_current").set(tx.len() as f64); + interval.tick().await; + } + }); +} + async fn run_rwi_webhook_handler( config: LocatorWebhookConfig, mut rx: broadcast::Receiver, @@ -136,9 +207,24 @@ async fn run_rwi_webhook_handler( loop { let entry = match rx.recv().await { - Ok(entry) => entry, + Ok(entry) => { + // Opt-in queueing latency: enqueued (gateway dispatch) -> + // dequeued here. Excludes the HTTP push itself; a slow router + // does NOT inflate this — queue wait does. + if sender.track_queue_latency { + let queued = + (chrono::Utc::now() - entry.cached_at).num_milliseconds() as f64 / 1000.0; + metrics::histogram!( + "rwi_event_queue_latency_seconds", + "event_type" => entry.event.event_type + ) + .record(queued); + } + entry + } Err(broadcast::error::RecvError::Lagged(n)) => { warn!("RWI webhook lagged, missed {} events", n); + metrics::counter!("rwi_events_dropped_total").increment(n as u64); continue; } Err(broadcast::error::RecvError::Closed) => { @@ -195,97 +281,75 @@ async fn run_rwi_webhook_handler( // serves as a compensation record when the receiver misses an event. let body = payload.to_string(); - let total_attempts = 1 + WEBHOOK_RETRY_COUNT; - for attempt in 1..=total_attempts { - if attempt > 1 { - tokio::time::sleep(Duration::from_millis(WEBHOOK_RETRY_INTERVAL_MS)).await; - } - match sender.send_payload(&payload, &body).await { - Ok(record) => { - let success = record - .status_code - .map(|c| (200..300).contains(&c)) - .unwrap_or(false); - let call_id = if entry.call_id.is_empty() { - "-" - } else { - entry.call_id.as_str() - }; - if success { - if consecutive_send_failures > 0 { - info!( - url = %record.url, - consecutive_failures = consecutive_send_failures, - "RWI webhook delivery recovered" - ); - } - consecutive_send_failures = 0; + match sender.send_payload(&payload, &body, event_type).await { + Ok(record) => { + let success = record + .status_code + .map(|c| (200..300).contains(&c)) + .unwrap_or(false); + let call_id = if entry.call_id.is_empty() { + "-" + } else { + entry.call_id.as_str() + }; + if success { + if consecutive_send_failures > 0 { info!( url = %record.url, - event_type, - call_id, - attempt, - status_code = record.status_code.unwrap_or(0), - latency_ms = record.latency_ms, - body = %record.body, - "RWI webhook delivered" + consecutive_failures = consecutive_send_failures, + "RWI webhook delivery recovered" ); - break; } - if attempt < total_attempts { - warn!( - url = %record.url, - event_type, - call_id, - attempt, - max_attempts = total_attempts, - status_code = record.status_code.unwrap_or(0), - latency_ms = record.latency_ms, - "RWI webhook returned non-success status, retrying" - ); - continue; - } - consecutive_send_failures += 1; - warn!( + consecutive_send_failures = 0; + metrics::counter!("rwi_events_pushed_total", "event_type" => event_type) + .increment(1); + info!( url = %record.url, event_type, call_id, - attempts = total_attempts, status_code = record.status_code.unwrap_or(0), latency_ms = record.latency_ms, body = %record.body, - "RWI webhook returned non-success status, giving up" + "RWI webhook delivered" ); - } - Err(e) => { - if attempt < total_attempts { - warn!( - url = %sender.url, - event_type, - call_id = %entry.call_id, - attempt, - max_attempts = total_attempts, - error = %e, - "RWI webhook send failed, retrying" - ); - continue; - } + } else { consecutive_send_failures += 1; - // INFO with the full request body: when the receiver is - // down this log is the only place to see which events - // (and payloads) were generated. The body is never - // truncated, so the log doubles as a compensation record. - info!( - url = %sender.url, + metrics::counter!( + "rwi_events_push_failed_total", + "event_type" => event_type + ) + .increment(1); + warn!( + url = %record.url, event_type, - call_id = %entry.call_id, - attempts = total_attempts, - body = %body, - error = %e, - "RWI webhook send failed" + call_id, + status_code = record.status_code.unwrap_or(0), + latency_ms = record.latency_ms, + body = %record.body, + "RWI webhook returned non-success status, giving up" ); } } + Err(e) => { + consecutive_send_failures += 1; + metrics::counter!( + "rwi_events_push_failed_total", + "event_type" => event_type + ) + .increment(1); + // INFO with the full request body: when the receiver is + // down this log is the only place to see which events + // (and payloads) were generated. The body is never + // truncated, so the log doubles as a compensation record. + info!( + url = %sender.url, + event_type, + call_id = %entry.call_id, + body = %body, + error = %e, + "RWI webhook send failed" + ); + } } } } @@ -300,6 +364,8 @@ pub async fn send_test_event( events: Vec::new(), headers: headers.cloned(), timeout_ms: Some(5000), + retries: Some(2), + track_queue_latency: None, }); let test_payload = json!({ "rwi": "1.0", @@ -313,8 +379,9 @@ pub async fn send_test_event( } }); + let body = test_payload.to_string(); sender - .send_payload(&test_payload, &test_payload.to_string()) + .send_payload(&test_payload, &body, "test") .await .map(|_| ()) } @@ -374,8 +441,10 @@ mod tests { events: vec![], headers: None, timeout_ms: Some(5000), + retries: Some(2), + track_queue_latency: None, }; - let tx = start_rwi_webhook_handler(config); + let tx = start_rwi_webhook_handler(config, WEBHOOK_CHANNEL_SIZE); tokio::time::sleep(Duration::from_millis(50)).await; let entry = EventCacheEntry { cached_at: chrono::Utc::now(), @@ -411,8 +480,10 @@ mod tests { events: vec![], headers: None, timeout_ms: Some(5000), + retries: None, + track_queue_latency: None, }; - let tx = start_rwi_webhook_handler(config); + let tx = start_rwi_webhook_handler(config, WEBHOOK_CHANNEL_SIZE); tokio::time::sleep(Duration::from_millis(50)).await; let ts = chrono::Utc::now(); @@ -471,8 +542,10 @@ mod tests { events: vec![], headers: None, timeout_ms: Some(5000), + retries: None, + track_queue_latency: None, }; - let tx = start_rwi_webhook_handler(config); + let tx = start_rwi_webhook_handler(config, WEBHOOK_CHANNEL_SIZE); tokio::time::sleep(Duration::from_millis(50)).await; let ts = chrono::Utc::now(); @@ -509,8 +582,10 @@ mod tests { events: vec![], headers: None, timeout_ms: Some(5000), + retries: Some(2), + track_queue_latency: None, }; - let tx = start_rwi_webhook_handler(config); + let tx = start_rwi_webhook_handler(config, WEBHOOK_CHANNEL_SIZE); tokio::time::sleep(Duration::from_millis(50)).await; let now = chrono::Utc::now(); @@ -597,11 +672,16 @@ mod tests { events: vec![], headers: None, timeout_ms: Some(5000), + retries: Some(2), + track_queue_latency: None, }); let payload = json!({"event_type": "test", "call_id": "c1"}); let body = payload.to_string(); - let record = sender.send_payload(&payload, &body).await.expect("send ok"); + let record = sender + .send_payload(&payload, &body, "test") + .await + .expect("send ok"); assert_eq!(record.url, server.url()); assert_eq!(record.status_code, Some(200)); @@ -622,6 +702,8 @@ mod tests { events: vec![], headers: None, timeout_ms: Some(5000), + retries: None, + track_queue_latency: None, }); // 1800 bytes of '请' (3 bytes each): byte 1024 falls inside a char, @@ -630,7 +712,10 @@ mod tests { let body = payload.to_string(); assert!(body.len() > 1024); - let record = sender.send_payload(&payload, &body).await.expect("send ok"); + let record = sender + .send_payload(&payload, &body, "test") + .await + .expect("send ok"); assert_eq!(record.status_code, Some(200)); assert_eq!(record.body, body, "body must not be truncated"); } @@ -656,13 +741,18 @@ mod tests { events: vec![], headers: None, timeout_ms: Some(5000), + retries: Some(2), + track_queue_latency: None, }); let payload = json!({"event_type": "test"}); // send_payload treats any HTTP response as Ok (it only errors on // transport failure); the status code is captured in the record. let body = payload.to_string(); - let record = sender.send_payload(&payload, &body).await.expect("http ok"); + let record = sender + .send_payload(&payload, &body, "test") + .await + .expect("http ok"); assert_eq!(record.status_code, Some(500)); } @@ -727,10 +817,9 @@ mod tests { } } - /// Failed deliveries (non-2xx) are retried up to `WEBHOOK_RETRY_COUNT` - /// times with `WEBHOOK_RETRY_INTERVAL_MS` between attempts, and every - /// attempt re-sends a byte-identical payload (stable `event_id`) so - /// receivers can dedupe. + /// Failed deliveries (retryable statuses: 5xx/429) are retried up to the + /// configured `retries` count, and every attempt re-sends a + /// byte-identical payload (stable `event_id`) so receivers can dedupe. #[tokio::test] async fn test_webhook_retries_until_success_with_identical_payload() { let server = RetryTestServer::start(2, axum::http::StatusCode::SERVICE_UNAVAILABLE).await; @@ -739,8 +828,10 @@ mod tests { events: vec![], headers: None, timeout_ms: Some(5000), + retries: Some(3), + track_queue_latency: None, }; - let tx = start_rwi_webhook_handler(config); + let tx = start_rwi_webhook_handler(config, WEBHOOK_CHANNEL_SIZE); tokio::time::sleep(Duration::from_millis(50)).await; tx.send(retry_test_entry("retry-1")).ok(); @@ -763,8 +854,8 @@ mod tests { ); } - /// After the initial attempt plus all retries fail, the handler gives up - /// (exactly `1 + WEBHOOK_RETRY_COUNT` requests) and stays alive so + /// After the initial attempt plus all configured retries fail, the + /// handler gives up (exactly `1 + retries` requests) and stays alive so /// subsequent events are still delivered. #[tokio::test] async fn test_webhook_gives_up_after_max_retries_and_stays_alive() { @@ -774,8 +865,10 @@ mod tests { events: vec![], headers: None, timeout_ms: Some(5000), + retries: Some(3), + track_queue_latency: None, }; - let tx = start_rwi_webhook_handler(config); + let tx = start_rwi_webhook_handler(config, WEBHOOK_CHANNEL_SIZE); tokio::time::sleep(Duration::from_millis(50)).await; tx.send(retry_test_entry("retry-2")).ok(); @@ -807,8 +900,10 @@ mod tests { events: vec![], headers: None, timeout_ms: Some(5000), + retries: None, + track_queue_latency: None, }; - let tx = start_rwi_webhook_handler(config); + let tx = start_rwi_webhook_handler(config, WEBHOOK_CHANNEL_SIZE); tokio::time::sleep(Duration::from_millis(50)).await; let transcript = "请检查录音质量与通话摘要。".repeat(100); diff --git a/src/utils.rs b/src/utils.rs index 4e6c234e8..133560e3c 100644 --- a/src/utils.rs +++ b/src/utils.rs @@ -196,7 +196,6 @@ where // instead of the SIP runtime, preventing RTP load from starving SIP timers. // --------------------------------------------------------------------------- static MEDIA_RUNTIME: OnceLock = OnceLock::new(); - /// Atomically set the global media runtime handle. Must be called exactly /// once at startup, before any media task is spawned. pub fn set_media_runtime(handle: Handle) { @@ -253,6 +252,53 @@ pub fn media_enter() -> Option> { MEDIA_RUNTIME.get().map(|h| h.enter()) } +// --------------------------------------------------------------------------- +// RWI webhook runtime isolation: a dedicated tokio runtime for the RWI HTTP +// push consumer. The webhook handler performs sequential blocking-ish HTTP +// POSTs to the router; running it on its own pool keeps its egress (and any +// backpressure from a slow router) off the SIP runtime shared by signalling, +// the HTTP route path, and the CDR saver. +// --------------------------------------------------------------------------- +static RWI_WEBHOOK_RUNTIME: OnceLock = OnceLock::new(); + +/// Atomically set the global RWI webhook runtime handle. Must be called +/// exactly once at startup, before the webhook handler is spawned. +pub fn set_rwi_webhook_runtime(handle: Handle) { + RWI_WEBHOOK_RUNTIME + .set(handle) + .expect("set_rwi_webhook_runtime called more than once"); +} + +/// Spawn a future onto the dedicated RWI webhook runtime. Falls back to the +/// ambient tokio runtime if the runtime has not been initialised (e.g. during +/// tests). +#[track_caller] +pub fn rwi_webhook_spawn(future: T) -> tokio::task::JoinHandle +where + T: std::future::Future + Send + 'static, + T::Output: Send + 'static, +{ + let location = std::panic::Location::caller(); + let loc = format!("{}:{}", location.file(), location.line()); + let _guard = TaskGuard::new(loc); + if let Some(handle) = RWI_WEBHOOK_RUNTIME.get() { + handle.spawn(async move { + let _guard = _guard; + future.await + }) + } else { + tokio::spawn(async move { + let _guard = _guard; + future.await + }) + } +} + +/// Return the configured RWI webhook runtime handle, if any. +pub fn rwi_webhook_runtime_handle() -> Option { + RWI_WEBHOOK_RUNTIME.get().cloned() +} + /// Collect tokio runtime metrics from the current and media runtimes. /// Returns a serde_json map with key metrics useful for leak detection. ///