feat(cdr): export push and queue pipeline metrics

docs/config/06-media-recording.md

@@ -227,6 +227,19 @@ The `[callrecord]` section is **optional**. It adds a secondary raw-CDR sink on
 
 `max_concurrent` controls how many post-call CDR save/upload/hook tasks may run at once. The default is `64`; values below `1` are clamped to `1`.
 
+Common pipeline options (independent of the storage type):
+
+| Key | Default | Description |
+|---|---|---|
+| `channel_capacity` | 2048 | Bounded queue length between call producers and the CDR manager. Producers drop (and count `cdr_records_dropped_total`) when full. |
+| `batch_size` | 64 | Max records per manager batch. |
+| `track_queue_latency` | false | Record the `cdr_queue_latency_seconds` histogram (queueing wait only — save/push time excluded). |
+
+The pipeline exports Prometheus metrics (`cdr_records_enqueued_total`,
+`cdr_records_pushed_total`, `cdr_records_push_failed_total`,
+`cdr_records_dropped_total`, `cdr_queue_size`, `cdr_queue_current`,
+`cdr_queue_latency_seconds`) — see [observability.md](../observability.md#call-record-cdr-pipeline).
+
 ### Database
 Writes CDR JSON to a separate database table (default: `call_records`).
 

docs/observability.md

@@ -109,6 +109,31 @@ All metrics emitted by RustPBX, organized by category:
 | `rustpbx_call_duration_seconds` | Histogram | `direction` | Wall-clock time from INVITE to BYE |
 | `rustpbx_call_talk_time_seconds` | Histogram | `direction` | Talk time (only for answered calls) |
 
+#### Call Record (CDR) Pipeline
+
+Metrics for the call-record pipeline: producers enqueue finished CDRs into a
+bounded queue (`[callrecord] channel_capacity`), and a single manager task
+drains the queue in batches and hands them to the configured saver
+(`http`, `database`, `local`, `s3`).
+
+| Metric | Type | Labels | Description |
+|---|---|---|---|
+| `cdr_records_enqueued_total` | Counter | - | CDRs accepted into the queue |
+| `cdr_records_pushed_total` | Counter | - | CDRs persisted by the saver (batch success) |
+| `cdr_records_push_failed_total` | Counter | - | CDRs in batches the saver failed to persist |
+| `cdr_records_dropped_total` | Counter | - | CDRs lost because the queue was full or the manager was gone |
+| `cdr_queue_size` | Gauge | - | Configured queue capacity |
+| `cdr_queue_current` | Gauge | - | CDRs currently queued (sampled every 5 s) |
+| `cdr_queue_latency_seconds` | Histogram | - | Queueing wait (record enqueued → manager dequeued); opt-in via `[callrecord] track_queue_latency` |
+
+A healthy pipeline keeps `cdr_records_enqueued_total == cdr_records_pushed_total`,
+`cdr_records_dropped_total == 0` and `cdr_queue_current` near 0. Rising
+`cdr_queue_current` or any `cdr_records_dropped_total` increments mean the
+saver endpoint is slower than the call completion rate — scale the endpoint or
+raise `channel_capacity` (memory-bounded). `cdr_queue_latency_seconds`
+measures how long a record waits in the queue only; a slow endpoint does not
+inflate it, backlog does.
+
 #### Trunk Metrics
 
 | Metric | Type | Labels | Description |

src/callrecord/mod.rs

@@ -186,6 +186,13 @@ pub struct CallRecord {
     pub extensions: http::Extensions,
 }
 
+/// Extension key stashing the enqueue instant on a `CallRecord` so the
+/// manager can measure queueing wait (enqueued → dequeued) for the
+/// opt-in `cdr_queue_latency_seconds` histogram without changing the
+/// channel item type. Never serialized.
+#[derive(Debug, Clone, Copy)]
+pub struct RecordEnqueuedAt(pub Instant);
+
 impl Clone for CallRecord {
     fn clone(&self) -> Self {
         Self {
@@ -538,6 +545,8 @@ pub struct CallRecordManager {
     pub batch_size: usize,
     pub sender: CallRecordSender,
     pub stats: Arc<CallRecordStats>,
+    channel_capacity: usize,
+    track_queue_latency: bool,
     cancel_token: CancellationToken,
     receiver: CallRecordReceiver,
     saver: Box<dyn CallRecordSaver>,
@@ -628,6 +637,7 @@ impl CallRecordManagerBuilder {
             .unwrap_or(DEFAULT_CALL_RECORD_BATCH_SIZE)
             .max(1);
         let (sender, receiver) = tokio::sync::mpsc::channel(channel_capacity);
+        let track_queue_latency = config.as_ref().is_some_and(|c| c.track_queue_latency);
         let saver: Box<dyn CallRecordSaver> = match config.map(|c| c.storage) {
             // No [callrecord] section → default: database → rustpbx_call_records
             None => {
@@ -763,6 +773,8 @@ impl CallRecordManagerBuilder {
         Ok(CallRecordManager {
             batch_size,
             stats: Arc::new(CallRecordStats::new()),
+            channel_capacity,
+            track_queue_latency,
             cancel_token,
             sender,
             receiver,
@@ -1328,7 +1340,17 @@ impl CallRecordManager {
     pub async fn serve(&mut self) {
         let token = self.cancel_token.clone();
         let batch_size = self.batch_size.max(1);
-        info!(batch_size, "CallRecordManager serving");
+        let track_queue_latency = self.track_queue_latency;
+        info!(
+            batch_size,
+            channel_capacity = self.channel_capacity,
+            track_queue_latency,
+            "CallRecordManager serving"
+        );
+        crate::metrics::cdr::set_queue_size(self.channel_capacity);
+        crate::metrics::cdr::set_queue_current(0);
+        let mut gauge_interval = tokio::time::interval(Duration::from_secs(5));
+        gauge_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
         let receiver = &mut self.receiver;
         let saver = self.saver.as_ref();
         let hooks = &self.hooks;
@@ -1350,6 +1372,17 @@ impl CallRecordManager {
                         continue;
                     }
 
+                    if track_queue_latency {
+                        let now = Instant::now();
+                        for record in records.iter_mut() {
+                            if let Some(enqueued_at) = record.extensions.remove::<RecordEnqueuedAt>() {
+                                crate::metrics::cdr::queue_latency_seconds(
+                                    now.duration_since(enqueued_at.0).as_secs_f64(),
+                                );
+                            }
+                        }
+                    }
+
                     let started_at = Instant::now();
                     for hook in hooks {
                         if let Err(e) = hook.on_record_enrich(&mut records).await {
@@ -1359,6 +1392,7 @@ impl CallRecordManager {
 
                     match saver.save(&records).await {
                         Ok(file_names) => {
+                            crate::metrics::cdr::pushed(records.len() as u64);
                             if file_names.len() != records.len() {
                                 warn!(
                                     batch_size = records.len(),
@@ -1376,6 +1410,7 @@ impl CallRecordManager {
                             );
                         }
                         Err(err) => {
+                            crate::metrics::cdr::push_failed(records.len() as u64);
                             warn!(batch_size = records.len(), "Failed to save call record batch: {}", err);
                         }
                     }
@@ -1386,6 +1421,9 @@ impl CallRecordManager {
                         }
                     }
                 }
+                _ = gauge_interval.tick(), if !shutting_down => {
+                    crate::metrics::cdr::set_queue_current(receiver.len());
+                }
                 _ = token.cancelled(), if !shutting_down => {
                     shutting_down = true;
                     info!(pending = receiver.len(), "CallRecordManager received shutdown");

src/callrecord/tests.rs

@@ -377,6 +377,7 @@ async fn test_database_saver_without_url_needs_main_db() {
         .with_config(CallRecordConfig {
             channel_capacity: 2048,
             batch_size: 64,
+            track_queue_latency: false,
             storage: CallRecordStorageConfig::Database {
                 database_url: None,
                 table_name: "custom_table".to_string(),
@@ -424,6 +425,7 @@ async fn test_local_config_without_main_db_ok() {
         .with_config(CallRecordConfig {
             channel_capacity: 2048,
             batch_size: 4,
+            track_queue_latency: false,
             storage: CallRecordStorageConfig::Local { root: root.clone() },
         })
         .build()
@@ -450,6 +452,7 @@ async fn test_builder_uses_configured_channel_and_batch_settings() {
         .with_config(CallRecordConfig {
             channel_capacity: 17,
             batch_size: 3,
+            track_queue_latency: false,
             storage: CallRecordStorageConfig::Local {
                 root: tmp.path().to_string_lossy().into_owned(),
             },
@@ -470,6 +473,7 @@ async fn test_http_config_without_main_db_ok() {
         .with_config(CallRecordConfig {
             channel_capacity: 2048,
             batch_size: 4,
+            track_queue_latency: false,
             storage: CallRecordStorageConfig::Http {
                 url: "http://127.0.0.1:1/cdr".to_string(),
                 headers: None,
@@ -494,6 +498,7 @@ async fn test_s3_config_without_main_db_ok() {
         .with_config(CallRecordConfig {
             channel_capacity: 2048,
             batch_size: 4,
+            track_queue_latency: false,
             storage: CallRecordStorageConfig::S3 {
                 vendor: crate::config::S3Vendor::Minio,
                 bucket: "test-bucket".to_string(),
@@ -523,6 +528,7 @@ async fn test_database_with_url_without_main_db_ok() {
         .with_config(CallRecordConfig {
             channel_capacity: 2048,
             batch_size: 4,
+            track_queue_latency: false,
             storage: CallRecordStorageConfig::Database {
                 database_url: Some("sqlite::memory:".to_string()),
                 table_name: "custom_cdr".to_string(),
@@ -556,6 +562,7 @@ async fn test_local_saver_does_not_write_to_db() {
         .with_config(CallRecordConfig {
             channel_capacity: 2048,
             batch_size: 4,
+            track_queue_latency: false,
             storage: CallRecordStorageConfig::Local { root: root.clone() },
         })
         .build()

src/config.rs

@@ -641,6 +641,11 @@ pub struct CallRecordConfig {
     /// Maximum number of records passed to hooks and the saver at once.
     #[serde(default = "default_call_record_batch_size")]
     pub batch_size: usize,
+    /// Record the queueing-wait histogram `cdr_queue_latency_seconds`
+    /// (record enqueued → manager dequeued). The save/push time itself is
+    /// excluded; disabled by default.
+    #[serde(default)]
+    pub track_queue_latency: bool,
     #[serde(flatten)]
     pub storage: CallRecordStorageConfig,
 }
@@ -1769,6 +1774,7 @@ impl Default for CallRecordConfig {
         Self {
             channel_capacity: default_call_record_channel_capacity(),
             batch_size: default_call_record_batch_size(),
+            track_queue_latency: false,
             storage: CallRecordStorageConfig::Local {
                 #[cfg(target_os = "windows")]
                 root: "./config/cdr".to_string(),

src/metrics.rs

@@ -768,6 +768,43 @@ pub mod cc {
     }
 }
 
+pub mod cdr {
+    /// Record accepted into the bounded queue (producer → manager).
+    pub fn enqueued() {
+        metrics::counter!("cdr_records_enqueued_total").increment(1);
+    }
+
+    /// Record lost because the queue was full or the manager was gone.
+    pub fn dropped() {
+        metrics::counter!("cdr_records_dropped_total").increment(1);
+    }
+
+    /// Records handed to the saver and persisted successfully.
+    pub fn pushed(n: u64) {
+        metrics::counter!("cdr_records_pushed_total").increment(n);
+    }
+
+    /// Records in a batch the saver failed to persist.
+    pub fn push_failed(n: u64) {
+        metrics::counter!("cdr_records_push_failed_total").increment(n);
+    }
+
+    pub fn set_queue_size(capacity: usize) {
+        metrics::gauge!("cdr_queue_size").set(capacity as f64);
+    }
+
+    pub fn set_queue_current(n: usize) {
+        metrics::gauge!("cdr_queue_current").set(n as f64);
+    }
+
+    /// Queueing wait (record enqueued → manager dequeued). Excludes the
+    /// save/push time itself; a slow endpoint does NOT inflate this —
+    /// queue backlog does. Opt-in via `[callrecord] track_queue_latency`.
+    pub fn queue_latency_seconds(duration_secs: f64) {
+        metrics::histogram!("cdr_queue_latency_seconds").record(duration_secs);
+    }
+}
+
 pub fn init_static_gauges() {
     let version = crate::version::get_short_version();
     metrics::gauge!("rustpbx_info", "version" => version).set(1.0);

src/proxy/proxy_call/reporter.rs

@@ -323,9 +323,20 @@ impl CallReporter {
             // Bounded channel: drop new records (with a warn log) if the
             // saver has fallen behind, instead of buffering indefinitely.
             // `try_send` is sync, so the existing synchronous emit path is
-            // preserved.
-            if let Err(tokio::sync::mpsc::error::TrySendError::Full(_)) = sender.try_send(record) {
-                tracing::warn!("call record channel full; dropping record to bound memory");
+            // preserved. The enqueue instant rides in `extensions` for the
+            // manager's opt-in queueing-latency histogram.
+            record
+                .extensions
+                .insert(crate::callrecord::RecordEnqueuedAt(std::time::Instant::now()));
+            match sender.try_send(record) {
+                Ok(()) => crate::metrics::cdr::enqueued(),
+                Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
+                    crate::metrics::cdr::dropped();
+                    tracing::warn!("call record channel full; dropping record to bound memory");
+                }
+                Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
+                    crate::metrics::cdr::dropped();
+                }
             }
         }
     }