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` | --- 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/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..380cf53a1 100644 --- a/src/config.rs +++ b/src/config.rs @@ -888,6 +888,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. @@ -1097,6 +1108,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, @@ -1289,6 +1312,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 } @@ -1754,6 +1785,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/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; diff --git a/src/rwi/gateway.rs b/src/rwi/gateway.rs index 03e9df405..5edd5ddfe 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 205009b67..2518659ad 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 = 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; @@ -29,6 +30,8 @@ struct RwiWebhookSender { headers: std::collections::HashMap, allowed_events: Vec, client: reqwest::Client, + retries: u32, + track_queue_latency: bool, } impl RwiWebhookSender { @@ -40,6 +43,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), } } @@ -56,14 +61,63 @@ 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. + // 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", + "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, ) -> 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 +134,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)] @@ -106,12 +166,31 @@ 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); - 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 } +/// 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, @@ -130,9 +209,25 @@ 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) => { @@ -178,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!( @@ -198,6 +293,8 @@ async fn run_rwi_webhook_handler( entry.call_id.as_str() }; if success { + metrics::counter!("rwi_events_pushed_total", "event_type" => event_type) + .increment(1); info!( url = %record.url, event_type, @@ -207,6 +304,11 @@ async fn run_rwi_webhook_handler( "RWI webhook delivered" ); } else { + metrics::counter!( + "rwi_events_push_failed_total", + "event_type" => event_type + ) + .increment(1); warn!( url = %record.url, event_type, @@ -220,6 +322,11 @@ async fn run_rwi_webhook_handler( } 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, so the body must be visible at @@ -247,6 +354,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", @@ -260,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)] @@ -318,8 +427,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(), @@ -350,8 +461,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(); @@ -438,10 +551,11 @@ mod tests { events: vec![], headers: None, timeout_ms: Some(5000), + retries: Some(2), }); 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)); @@ -473,12 +587,13 @@ mod tests { events: vec![], headers: None, timeout_ms: Some(5000), + retries: Some(2), }); 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)); } 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. ///