From 45d0cb77b0d1df25470ee5d07485ddbb13e606c7 Mon Sep 17 00:00:00 2001 From: tongfengyuan <71140753@chinatelecom.cn> Date: Wed, 2 Sep 2026 00:50:25 +0800 Subject: [PATCH 1/9] feat(rwi): run RWI webhook handler on a dedicated tokio runtime Add a dedicated RWI webhook tokio runtime (configurable worker count via [proxy] rwi_webhook_worker_threads, default 2) so the webhook's sequential blocking-ish HTTP POSTs to the router (and any backpressure from a slow router) run off the SIP runtime shared by signalling, the HTTP route path and the CDR saver. - utils: set_rwi_webhook_runtime / rwi_webhook_spawn / rwi_webhook_runtime_handle helpers. - bin/rustpbx: build the rwi-webhook runtime alongside sip/media. - rwi/webhook: dispatch the handler via rwi_webhook_spawn; raise the webhook broadcast channel 512 -> 100000. --- src/bin/rustpbx.rs | 14 ++++++++++++-- src/config.rs | 11 +++++++++++ src/rwi/webhook.rs | 4 ++-- src/utils.rs | 48 +++++++++++++++++++++++++++++++++++++++++++++- 4 files changed, 72 insertions(+), 5 deletions(-) diff --git a/src/bin/rustpbx.rs b/src/bin/rustpbx.rs index b3b164bf5..c3f93fbbe 100644 --- a/src/bin/rustpbx.rs +++ b/src/bin/rustpbx.rs @@ -328,10 +328,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() @@ -346,6 +347,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 bf7e89546..15853010a 100644 --- a/src/config.rs +++ b/src/config.rs @@ -1097,6 +1097,12 @@ 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, pub ws_handler: Option, pub ami_path: Option, pub rwi_path: Option, @@ -1289,6 +1295,10 @@ fn default_media_worker_threads() -> usize { if n > sip { n - sip } else { 1 } } +fn default_rwi_webhook_worker_threads() -> usize { + 2 +} + fn default_auth_cache_size() -> usize { 10000 } @@ -1754,6 +1764,7 @@ 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(), } } } diff --git a/src/rwi/webhook.rs b/src/rwi/webhook.rs index 205009b67..4accfecc7 100644 --- a/src/rwi/webhook.rs +++ b/src/rwi/webhook.rs @@ -8,7 +8,7 @@ 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; +const WEBHOOK_CHANNEL_SIZE: usize = 100000; /// Max number of recent (call_id, timestamp) pairs kept for dedup. const DEDUP_CACHE_SIZE: usize = 4096; @@ -108,7 +108,7 @@ pub fn start_rwi_webhook_handler( config: LocatorWebhookConfig, ) -> broadcast::Sender { let (tx, rx) = broadcast::channel(WEBHOOK_CHANNEL_SIZE); - crate::utils::spawn(run_rwi_webhook_handler(config, rx)); + crate::utils::rwi_webhook_spawn(run_rwi_webhook_handler(config, rx)); tx } 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. /// From 2f4efa9ae004f7231fcb45a238483982f6159470 Mon Sep 17 00:00:00 2001 From: tongfengyuan <71140753@chinatelecom.cn> Date: Tue, 1 Sep 2026 23:50:48 +0800 Subject: [PATCH 2/9] feat(rwi): export event pipeline metrics Add Prometheus metrics covering the RWI event pipeline: - rwi_events_sent_total: events pushed into the webhook broadcast channel by the gateway dispatch (fanout_webhook_tap). - rwi_events_pushed_total: events successfully delivered (2xx) by the webhook handler's HTTP push. - rwi_events_push_failed_total: webhook HTTP pushes that failed or returned a non-2xx status. - rwi_events_dropped_total: events lost to broadcast lag (the handler consumer fell behind the 100k channel and skipped events). - rwi_event_queue_size: webhook broadcast channel capacity. - rwi_event_queue_current: events currently queued in the channel (sampled every 5 s on the RWI webhook runtime; slow-router backpressure shows up as current climbing toward size). Queue gauges are sampled on the dedicated RWI webhook runtime so queue observation never contends with the SIP runtime. --- src/rwi/gateway.rs | 1 + src/rwi/webhook.rs | 22 ++++++++++++++++++++++ 2 files changed, 23 insertions(+) diff --git a/src/rwi/gateway.rs b/src/rwi/gateway.rs index 03e9df405..51104aeb1 100644 --- a/src/rwi/gateway.rs +++ b/src/rwi/gateway.rs @@ -331,6 +331,7 @@ impl RwiGateway { fn fanout_webhook_tap(&self, entry: &EventCacheEntry) { if let Some(tx) = &self.webhook_tx { let _ = tx.send(entry.clone()); + metrics::counter!("rwi_events_sent_total").increment(1); } let _ = self.event_tap.send(entry.clone()); } diff --git a/src/rwi/webhook.rs b/src/rwi/webhook.rs index 4accfecc7..8a6b6a7fe 100644 --- a/src/rwi/webhook.rs +++ b/src/rwi/webhook.rs @@ -108,10 +108,28 @@ pub fn start_rwi_webhook_handler( config: LocatorWebhookConfig, ) -> broadcast::Sender { let (tx, rx) = broadcast::channel(WEBHOOK_CHANNEL_SIZE); + spawn_queue_metrics(tx.clone()); crate::utils::rwi_webhook_spawn(run_rwi_webhook_handler(config, rx)); tx } +/// 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) { + crate::utils::rwi_webhook_spawn(async move { + metrics::gauge!("rwi_event_queue_size").set(WEBHOOK_CHANNEL_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, @@ -133,6 +151,7 @@ async fn run_rwi_webhook_handler( Ok(entry) => 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) => { @@ -198,6 +217,7 @@ async fn run_rwi_webhook_handler( entry.call_id.as_str() }; if success { + metrics::counter!("rwi_events_pushed_total").increment(1); info!( url = %record.url, event_type, @@ -207,6 +227,7 @@ async fn run_rwi_webhook_handler( "RWI webhook delivered" ); } else { + metrics::counter!("rwi_events_push_failed_total").increment(1); warn!( url = %record.url, event_type, @@ -220,6 +241,7 @@ async fn run_rwi_webhook_handler( } Err(e) => { consecutive_send_failures += 1; + metrics::counter!("rwi_events_push_failed_total").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, so the body must be visible at From 0e19a967f97af2dd81d07da54a248ae3f647a9d4 Mon Sep 17 00:00:00 2001 From: tongfengyuan <71140753@chinatelecom.cn> Date: Wed, 2 Sep 2026 00:31:22 +0800 Subject: [PATCH 3/9] feat(rwi): add retry limit to the webhook HTTP push Adds [proxy.locator_webhook] retries (default 0 = single attempt, hard cap 5). After a failed attempt the push retries with exponential backoff (200 ms base, doubling) for retryable outcomes: transport errors, 5xx and 429. Other 4xx are permanent and return immediately. Each attempt is bounded by the client's request timeout ([proxy.locator_webhook] timeout_ms, default 5 s), which already applies per request through build_keepalive_client. Also counts retries as rwi_events_push_retries_total, and documents the per-request timeout in send_payload. --- src/config.rs | 5 ++++ src/rwi/webhook.rs | 61 ++++++++++++++++++++++++++++++++++++++++++++-- 2 files changed, 64 insertions(+), 2 deletions(-) diff --git a/src/config.rs b/src/config.rs index 15853010a..c8c0e9ab6 100644 --- a/src/config.rs +++ b/src/config.rs @@ -888,6 +888,11 @@ 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, } /// Global recovery for Step IVR when the external provider cannot continue. diff --git a/src/rwi/webhook.rs b/src/rwi/webhook.rs index 8a6b6a7fe..71ace8579 100644 --- a/src/rwi/webhook.rs +++ b/src/rwi/webhook.rs @@ -29,6 +29,7 @@ struct RwiWebhookSender { headers: std::collections::HashMap, allowed_events: Vec, client: reqwest::Client, + retries: u32, } impl RwiWebhookSender { @@ -40,6 +41,7 @@ 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), } } @@ -56,14 +58,58 @@ impl RwiWebhookSender { async fn send_payload( &self, payload: &serde_json::Value, + ) -> 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).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").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, ) -> 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 @@ -80,6 +126,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. #[derive(Debug, Clone)] @@ -269,6 +321,7 @@ pub async fn send_test_event( events: Vec::new(), headers: headers.cloned(), timeout_ms: Some(5000), + retries: Some(2), }); let test_payload = json!({ "rwi": "1.0", @@ -340,6 +393,7 @@ mod tests { events: vec![], headers: None, timeout_ms: Some(5000), + retries: Some(2), }; let tx = start_rwi_webhook_handler(config); tokio::time::sleep(Duration::from_millis(50)).await; @@ -372,6 +426,7 @@ mod tests { events: vec![], headers: None, timeout_ms: Some(5000), + retries: Some(2), }; let tx = start_rwi_webhook_handler(config); tokio::time::sleep(Duration::from_millis(50)).await; @@ -460,6 +515,7 @@ mod tests { events: vec![], headers: None, timeout_ms: Some(5000), + retries: Some(2), }); let payload = json!({"event_type": "test", "call_id": "c1"}); @@ -495,6 +551,7 @@ mod tests { events: vec![], headers: None, timeout_ms: Some(5000), + retries: Some(2), }); let payload = json!({"event_type": "test"}); From 9f3dfc9770b28ad12b23ff2b4faf9b13a9ce3a98 Mon Sep 17 00:00:00 2001 From: tongfengyuan <71140753@chinatelecom.cn> Date: Wed, 2 Sep 2026 00:59:25 +0800 Subject: [PATCH 4/9] feat(rwi): make the webhook event queue length configurable [proxy] rwi_webhook_channel_size (default 100000) sets the capacity of the broadcast channel between the gateway and the webhook handler. The queue-size gauge reports the configured value. Also exports WEBHOOK_CHANNEL_SIZE so tests can pass an explicit size. --- src/app.rs | 5 ++++- src/config.rs | 11 +++++++++++ src/rwi/webhook.rs | 18 ++++++++++-------- 3 files changed, 25 insertions(+), 9 deletions(-) diff --git a/src/app.rs b/src/app.rs index 8a1966011..1c44d590e 100644 --- a/src/app.rs +++ b/src/app.rs @@ -626,7 +626,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/config.rs b/src/config.rs index c8c0e9ab6..d984c8070 100644 --- a/src/config.rs +++ b/src/config.rs @@ -1108,6 +1108,12 @@ pub struct ProxyConfig { /// 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, @@ -1304,6 +1310,10 @@ fn default_rwi_webhook_worker_threads() -> usize { 2 } +fn default_rwi_webhook_channel_size() -> usize { + 100000 +} + fn default_auth_cache_size() -> usize { 10000 } @@ -1770,6 +1780,7 @@ impl Default for ProxyConfig { 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/webhook.rs b/src/rwi/webhook.rs index 71ace8579..982ceae5a 100644 --- a/src/rwi/webhook.rs +++ b/src/rwi/webhook.rs @@ -7,8 +7,9 @@ use std::collections::{HashSet, VecDeque}; 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 = 100000; +/// 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 = 100000; /// Max number of recent (call_id, timestamp) pairs kept for dedup. const DEDUP_CACHE_SIZE: usize = 4096; @@ -158,9 +159,10 @@ fn truncate_payload(payload: &serde_json::Value) -> String { /// 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); - spawn_queue_metrics(tx.clone()); + 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 } @@ -169,9 +171,9 @@ pub fn start_rwi_webhook_handler( /// 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) { +fn spawn_queue_metrics(tx: broadcast::Sender, size: usize) { crate::utils::rwi_webhook_spawn(async move { - metrics::gauge!("rwi_event_queue_size").set(WEBHOOK_CHANNEL_SIZE as f64); + 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 @@ -395,7 +397,7 @@ mod tests { timeout_ms: Some(5000), retries: Some(2), }; - 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(), @@ -428,7 +430,7 @@ mod tests { timeout_ms: Some(5000), retries: Some(2), }; - 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(); From 341131ca72ad43426f47eec9ae997fadf54d3a9f Mon Sep 17 00:00:00 2001 From: tongfengyuan <71140753@chinatelecom.cn> Date: Wed, 2 Sep 2026 06:53:24 +0800 Subject: [PATCH 5/9] fix(rwi): queue latency histogram tracks queueing wait, renamed accordingly MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The opt-in latency histogram now measures the QUEUEING wait — from the gateway enqueuing the event to the webhook handler dequeuing it — instead of the end-to-end push duration. The HTTP push time is excluded on purpose: a slow router inflates push time, not queue wait. - rename rwi_event_push_latency_seconds -> rwi_event_queue_latency_seconds - sample at dequeue (rx.recv Ok), before dedup/push - exclude the HTTP push itself; slow-router backpressure does not show up here - update [rwi_webhook] track_latency doc --- src/config.rs | 8 ++++++- src/rwi/webhook.rs | 54 +++++++++++++++++++++++++++++++++++++--------- 2 files changed, 51 insertions(+), 11 deletions(-) diff --git a/src/config.rs b/src/config.rs index d984c8070..a1927921a 100644 --- a/src/config.rs +++ b/src/config.rs @@ -893,6 +893,12 @@ pub struct LocatorWebhookConfig { /// 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_latency: Option, } /// Global recovery for Step IVR when the external provider cannot continue. @@ -1311,7 +1317,7 @@ fn default_rwi_webhook_worker_threads() -> usize { } fn default_rwi_webhook_channel_size() -> usize { - 100000 + crate::rwi::webhook::WEBHOOK_CHANNEL_SIZE } fn default_auth_cache_size() -> usize { diff --git a/src/rwi/webhook.rs b/src/rwi/webhook.rs index 982ceae5a..3e286d969 100644 --- a/src/rwi/webhook.rs +++ b/src/rwi/webhook.rs @@ -9,7 +9,7 @@ use tracing::{debug, info, warn}; /// 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 = 100000; +pub const WEBHOOK_CHANNEL_SIZE: usize = 512; /// Max number of recent (call_id, timestamp) pairs kept for dedup. const DEDUP_CACHE_SIZE: usize = 4096; @@ -31,6 +31,7 @@ struct RwiWebhookSender { allowed_events: Vec, client: reqwest::Client, retries: u32, + track_latency: bool, } impl RwiWebhookSender { @@ -43,6 +44,7 @@ impl RwiWebhookSender { client: crate::http_util::build_keepalive_client(Some(timeout), None) .unwrap_or_else(|_| reqwest::Client::new()), retries: config.retries.unwrap_or(0), + track_latency: config.track_latency.unwrap_or(false), } } @@ -59,6 +61,7 @@ impl RwiWebhookSender { async fn send_payload( &self, payload: &serde_json::Value, + 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. @@ -96,7 +99,11 @@ impl RwiWebhookSender { ); } } - metrics::counter!("rwi_events_push_retries_total").increment(1); + 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; } @@ -202,7 +209,22 @@ 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_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); @@ -251,7 +273,7 @@ async fn run_rwi_webhook_handler( "event": event_value, }); - match sender.send_payload(&payload).await { + match sender.send_payload(&payload, event_type).await { Ok(record) => { if consecutive_send_failures > 0 { info!( @@ -271,7 +293,8 @@ async fn run_rwi_webhook_handler( entry.call_id.as_str() }; if success { - metrics::counter!("rwi_events_pushed_total").increment(1); + metrics::counter!("rwi_events_pushed_total", "event_type" => event_type) + .increment(1); info!( url = %record.url, event_type, @@ -281,7 +304,11 @@ async fn run_rwi_webhook_handler( "RWI webhook delivered" ); } else { - metrics::counter!("rwi_events_push_failed_total").increment(1); + metrics::counter!( + "rwi_events_push_failed_total", + "event_type" => event_type + ) + .increment(1); warn!( url = %record.url, event_type, @@ -295,7 +322,11 @@ async fn run_rwi_webhook_handler( } Err(e) => { consecutive_send_failures += 1; - metrics::counter!("rwi_events_push_failed_total").increment(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, so the body must be visible at @@ -324,6 +355,7 @@ pub async fn send_test_event( headers: headers.cloned(), timeout_ms: Some(5000), retries: Some(2), + track_latency: None, }); let test_payload = json!({ "rwi": "1.0", @@ -337,7 +369,7 @@ pub async fn send_test_event( } }); - sender.send_payload(&test_payload).await.map(|_| ()) + sender.send_payload(&test_payload, "test").await.map(|_| ()) } #[cfg(test)] @@ -396,6 +428,7 @@ mod tests { headers: None, timeout_ms: Some(5000), retries: Some(2), + track_latency: None, }; let tx = start_rwi_webhook_handler(config, WEBHOOK_CHANNEL_SIZE); tokio::time::sleep(Duration::from_millis(50)).await; @@ -429,6 +462,7 @@ mod tests { headers: None, timeout_ms: Some(5000), retries: Some(2), + track_latency: None, }; let tx = start_rwi_webhook_handler(config, WEBHOOK_CHANNEL_SIZE); tokio::time::sleep(Duration::from_millis(50)).await; @@ -521,7 +555,7 @@ mod tests { }); let payload = json!({"event_type": "test", "call_id": "c1"}); - let record = sender.send_payload(&payload).await.expect("send ok"); + let record = sender.send_payload(&payload, "test").await.expect("send ok"); assert_eq!(record.url, server.url()); assert_eq!(record.status_code, Some(200)); @@ -559,7 +593,7 @@ mod tests { 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 record = sender.send_payload(&payload).await.expect("http ok"); + let record = sender.send_payload(&payload, "test").await.expect("http ok"); assert_eq!(record.status_code, Some(500)); } From 7766eeacf7c9835a6c95f685d6f5acd6449587ce Mon Sep 17 00:00:00 2001 From: tongfengyuan <71140753@chinatelecom.cn> Date: Wed, 2 Sep 2026 07:03:36 +0800 Subject: [PATCH 6/9] refactor(rwi): rename track_latency config to track_queue_latency Matches the metric semantics after the queue-latency histogram rename: the flag gates rwi_event_queue_latency_seconds (queueing wait), not the HTTP push. --- src/config.rs | 2 +- src/rwi/webhook.rs | 12 ++++++------ 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/src/config.rs b/src/config.rs index a1927921a..380cf53a1 100644 --- a/src/config.rs +++ b/src/config.rs @@ -898,7 +898,7 @@ pub struct LocatorWebhookConfig { /// Excludes the HTTP push itself. Disabled by default — opt in /// explicitly. #[serde(default)] - pub track_latency: Option, + pub track_queue_latency: Option, } /// Global recovery for Step IVR when the external provider cannot continue. diff --git a/src/rwi/webhook.rs b/src/rwi/webhook.rs index 3e286d969..2518659ad 100644 --- a/src/rwi/webhook.rs +++ b/src/rwi/webhook.rs @@ -31,7 +31,7 @@ struct RwiWebhookSender { allowed_events: Vec, client: reqwest::Client, retries: u32, - track_latency: bool, + track_queue_latency: bool, } impl RwiWebhookSender { @@ -44,7 +44,7 @@ impl RwiWebhookSender { client: crate::http_util::build_keepalive_client(Some(timeout), None) .unwrap_or_else(|_| reqwest::Client::new()), retries: config.retries.unwrap_or(0), - track_latency: config.track_latency.unwrap_or(false), + track_queue_latency: config.track_queue_latency.unwrap_or(false), } } @@ -213,7 +213,7 @@ async fn run_rwi_webhook_handler( // 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_latency { + if sender.track_queue_latency { let queued = (chrono::Utc::now() - entry.cached_at) .num_milliseconds() as f64 / 1000.0; @@ -355,7 +355,7 @@ pub async fn send_test_event( headers: headers.cloned(), timeout_ms: Some(5000), retries: Some(2), - track_latency: None, + track_queue_latency: None, }); let test_payload = json!({ "rwi": "1.0", @@ -428,7 +428,7 @@ mod tests { headers: None, timeout_ms: Some(5000), retries: Some(2), - track_latency: None, + track_queue_latency: None, }; let tx = start_rwi_webhook_handler(config, WEBHOOK_CHANNEL_SIZE); tokio::time::sleep(Duration::from_millis(50)).await; @@ -462,7 +462,7 @@ mod tests { headers: None, timeout_ms: Some(5000), retries: Some(2), - track_latency: None, + track_queue_latency: None, }; let tx = start_rwi_webhook_handler(config, WEBHOOK_CHANNEL_SIZE); tokio::time::sleep(Duration::from_millis(50)).await; From fe05c13ee86ef09635ddfe502af796d89e2c6311 Mon Sep 17 00:00:00 2001 From: tongfengyuan <71140753@chinatelecom.cn> Date: Wed, 2 Sep 2026 07:08:15 +0800 Subject: [PATCH 7/9] docs(rwi): document webhook runtime/queue config and event metrics - rwi_events_reference (en/zh): [rwi_webhook] retries and track_queue_latency fields, [proxy] rwi_webhook_worker_threads / rwi_webhook_channel_size keys, and the webhook metrics table (enqueued/pushed/failed/retries/dropped/queue size/queue current/ queue latency). - observability.md: add the RWI Events section to the metrics reference. --- docs/observability.md | 13 ++++++++++++ docs/rwi_events_reference.md | 36 +++++++++++++++++++++++++++++++++ docs/rwi_events_reference_en.md | 32 ++++++++++++++++++++++++++++- 3 files changed, 80 insertions(+), 1 deletion(-) diff --git a/docs/observability.md b/docs/observability.md index 4173dd267..7332fae7d 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -158,6 +158,19 @@ All metrics emitted by RustPBX, organized by category: | `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 acb929739..58eb5d996 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 a7a2fc883..4e8c3555c 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` | --- From bc7b9e60fed541a3eaf8ae4444b77a76ee8037a5 Mon Sep 17 00:00:00 2001 From: tongfengyuan <71140753@chinatelecom.cn> Date: Wed, 2 Sep 2026 07:14:52 +0800 Subject: [PATCH 8/9] fix(rwi): rename enqueued counter and add event_type label The pipeline metrics commit shipped the gateway dispatch counter under its old name (rwi_events_sent_total) and without the event_type label. Rename to rwi_event_enqueued_total and tag with event_type, matching the other event pipeline metrics. --- src/rwi/gateway.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/rwi/gateway.rs b/src/rwi/gateway.rs index 51104aeb1..5edd5ddfe 100644 --- a/src/rwi/gateway.rs +++ b/src/rwi/gateway.rs @@ -331,7 +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_events_sent_total").increment(1); + metrics::counter!( + "rwi_event_enqueued_total", + "event_type" => entry.event.event_type + ) + .increment(1); } let _ = self.event_tap.send(entry.clone()); } From 94b94bdeda6e26ef8545fbb0d268ca76137a8114 Mon Sep 17 00:00:00 2001 From: tongfengyuan <71140753@chinatelecom.cn> Date: Wed, 2 Sep 2026 07:14:52 +0800 Subject: [PATCH 9/9] fix(rwi): emit call_ringing on every provisional, once per call MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CallRinging was only emitted in the no-SDP (180) branch of the callee provisional handling; the 183-with-SDP (early media) branch — the majority of trunk calls — emitted nothing, so most calls never reported ringing. Emit CallRinging at the top of the provisional handling so BOTH a 183 with SDP and a 180 without fire it, guarded by a media.ringing_event_sent flag so it fires exactly once per call. --- src/proxy/proxy_call/media_state.rs | 4 ++++ src/proxy/proxy_call/sip_session/session.rs | 17 +++++++++++++++++ 2 files changed, 21 insertions(+) diff --git a/src/proxy/proxy_call/media_state.rs b/src/proxy/proxy_call/media_state.rs index 64fdeb337..55f99023f 100644 --- a/src/proxy/proxy_call/media_state.rs +++ b/src/proxy/proxy_call/media_state.rs @@ -13,6 +13,9 @@ pub struct MediaState { pub callee_offer_cached_webrtc: Option, pub answer: Option, pub early_media_sent: bool, + /// RWI `call_ringing` fires once per call — on the FIRST provisional + /// (183 with SDP, or 180 without). Later provisionals do not re-fire. + pub ringing_event_sent: bool, pub callee_answer_sdp: Option, pub bridge: Option, } @@ -26,6 +29,7 @@ impl MediaState { callee_offer_cached_webrtc: None, answer: None, early_media_sent: false, + ringing_event_sent: false, callee_answer_sdp: None, bridge: None, } diff --git a/src/proxy/proxy_call/sip_session/session.rs b/src/proxy/proxy_call/sip_session/session.rs index 8d2c0611e..b61316275 100644 --- a/src/proxy/proxy_call/sip_session/session.rs +++ b/src/proxy/proxy_call/sip_session/session.rs @@ -5347,6 +5347,23 @@ impl SipSession { } let callee_sdp = String::from_utf8_lossy(response.body()).to_string(); + debug!( + session_id = %self.id, + session_id = %self.context.session_id, + status = %response.status_code, + sdp_len = callee_sdp.len(), + "callee provisional response received" + ); + // Ringing is signalled by EITHER provisional: a 183 + // with SDP (early media started) or a 180 without. + // Fires once per call — later provisionals do not + // re-fire the event. + if !self.media.ringing_event_sent { + self.media.ringing_event_sent = true; + self.emit_typed_rwi_event(&crate::rwi::CallRinging { + call_id: self.context.session_id.clone(), + }); + } if !callee_sdp.is_empty() && callee_sdp.contains("v=0") { if !self.media.early_media_sent { self.media.early_media_sent = true;