feat(rwi): dedicated webhook runtime, event metrics, push retry and configurable queue (+ call_ringing fix)

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 |

docs/rwi_events_reference.md

@@ -43,6 +43,12 @@ Authorization: Bearer <token>
 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\<String\> | [](全部) | 事件类型白名单;为空转发全部事件 |
+| `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. 信封格式

docs/rwi_events_reference_en.md

@@ -43,6 +43,12 @@ Or via query parameter: `GET /rwi/v1?token=<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\<String\> | [] (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` |
 
 ---
 

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);
         }

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")

src/config.rs

@@ -888,6 +888,17 @@ pub struct LocatorWebhookConfig {
     pub events: Vec<String>,
     pub headers: Option<HashMap<String, String>>,
     pub timeout_ms: Option<u64>,
+    /// 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<u32>,
+    /// 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<bool>,
 }
 
 /// 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<String>,
     pub ami_path: Option<String>,
     pub rwi_path: Option<String>,
@@ -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(),
         }
     }
 }

src/proxy/proxy_call/media_state.rs

@@ -13,6 +13,9 @@ pub struct MediaState {
     pub callee_offer_cached_webrtc: Option<bool>,
     pub answer: Option<String>,
     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<String>,
     pub bridge: Option<MediaBridge>,
 }
@@ -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,
         }

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;

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());
     }