feldera · Karakatiza666 · Jul 16, 2026 · Jul 16, 2026 · mythical-fred · Jul 16, 2026
diff --git a/crates/adapters/src/controller.rs b/crates/adapters/src/controller.rs
@@ -5470,13 +5470,18 @@ impl StatisticsThread {
             };
 
             // Update time series..
+            let latency = controller_status.latency_percentiles();
             let sample = SampleStatistics {
                 time: Utc::now(),
                 total_processed_records: controller_status
                     .global_metrics
                     .num_total_processed_records(),
                 memory_bytes: process_rss_bytes().unwrap_or_default(),
                 storage_bytes,
+                processing_latency_p50_micros: latency.processing_p50,
+                processing_latency_p99_micros: latency.processing_p99,
+                completion_latency_p50_micros: latency.completion_p50,
+                completion_latency_p99_micros: latency.completion_p99,
             };
             let mut time_series = controller_status.time_series.lock().unwrap();
             if time_series.len() >= 60 {

diff --git a/crates/adapters/src/controller/stats.rs b/crates/adapters/src/controller/stats.rs
@@ -751,6 +751,47 @@ impl ControllerStatus {
         self.inputs.read_recursive()
     }
 
+    /// Latency percentiles (microseconds) across input connectors, for the
+    /// time-series graph. Each connector contributes its own median latency,
+    /// we then take p50 and p99 across connectors, for both
+    /// processing (ingest to processed) and completion (ingest to all outputs).
+    /// `None` when no connector has samples.
+    pub fn latency_percentiles(&self) -> LatencyPercentiles {
+        let inputs = self.input_status();
+        let mut processing = Vec::with_capacity(inputs.len());
+        let mut completion = Vec::with_capacity(inputs.len());
+        for ep in inputs.values() {
+            if let Some(median) = ep
+                .metrics
+                .processing_latency_micros_histogram
+                .lock()
+                .unwrap()
+                .snapshot()
+                .quantile(0.5)
+            {
+                processing.push(median);
+            }
+            if let Some(median) = ep
+                .metrics
+                .completion_latency_micros_histogram
+                .lock()
+                .unwrap()
+                .snapshot()
+                .quantile(0.5)
+            {
+                completion.push(median);
+            }
+        }
+        processing.sort_unstable();
+        completion.sort_unstable();
+        LatencyPercentiles {
+            processing_p50: percentile_of_sorted(&processing, 0.5),
+            processing_p99: percentile_of_sorted(&processing, 0.99),
+            completion_p50: percentile_of_sorted(&completion, 0.5),
+            completion_p99: percentile_of_sorted(&completion, 0.99),
+        }
+    }
+
     /// Register connector-specific metrics for an input endpoint.
     pub fn set_input_custom_metrics(
         &self,
@@ -1525,6 +1566,25 @@ impl InputEndpointMetrics {
     }
 }
 
+/// Latency percentiles for the time series, in microseconds. See
+/// [`ControllerStatus::latency_percentiles`].
+#[derive(Clone, Copy, Debug, Default)]
+pub struct LatencyPercentiles {
+    pub processing_p50: Option<u64>,
+    pub processing_p99: Option<u64>,
+    pub completion_p50: Option<u64>,
+    pub completion_p99: Option<u64>,
+}
+
+/// Returns `q`-quantile of an ascending-sorted slice (ceil-rank). `None` if empty.
+fn percentile_of_sorted(sorted: &[u64], q: f64) -> Option<u64> {
+    if sorted.is_empty() {
+        return None;
+    }
+    let rank = ((q.clamp(0.0, 1.0) * sorted.len() as f64).ceil() as usize).clamp(1, sorted.len());
+    Some(sorted[rank - 1])
+}
+
 #[derive(Debug)]
 pub struct StepResults {
     pub amt: BufferSize,
@@ -2800,3 +2860,63 @@ impl OutputEndpointStatus {
         self.metrics.total_processed_steps.load(Ordering::Acquire)
     }
 }
+
+#[cfg(test)]
+mod latency_tests {
+    use super::percentile_of_sorted;
+
+    #[test]
+    fn percentile_empty() {
+        assert_eq!(percentile_of_sorted(&[], 0.5), None);
+        assert_eq!(percentile_of_sorted(&[], 0.99), None);
+    }
+
+    #[test]
+    fn percentile_single() {
+        assert_eq!(percentile_of_sorted(&[42], 0.5), Some(42));
+        assert_eq!(percentile_of_sorted(&[42], 0.99), Some(42));
+    }
+
+    #[test]
+    fn percentile_uniform() {
+        let v: Vec<u64> = (1..=100).collect();
+        // ceil(0.5*100)=50 -> 1-based rank 50 -> value 50.
+        assert_eq!(percentile_of_sorted(&v, 0.5), Some(50));
+        // ceil(0.99*100)=99 -> value 99.
+        assert_eq!(percentile_of_sorted(&v, 0.99), Some(99));
+    }
+
+    #[test]
+    fn percentile_few_slow_connectors() {
+        // 97 fast connectors at 1ms, 3 slow at 500ms (values in micros).
+        let mut v = vec![1_000u64; 97];
+        v.extend([500_000, 500_000, 500_000]);
+        v.sort_unstable();
+        // p50 stays with the fast majority.
+        assert_eq!(percentile_of_sorted(&v, 0.5), Some(1_000));
+        // p99 (rank 99) reaches the slow tail: surfaces the laggards.
+        assert_eq!(percentile_of_sorted(&v, 0.99), Some(500_000));
+    }
+
+    #[test]
+    fn percentile_broad_degradation() {
+        // 80 fast at 1ms, 20 slow at 100ms: ~20% degraded.
+        let mut v = vec![1_000u64; 80];
+        v.extend(std::iter::repeat(100_000).take(20));
+        v.sort_unstable();
+        // Both p50 (typical) stays fast and p99 catches the slow band.
+        assert_eq!(percentile_of_sorted(&v, 0.5), Some(1_000));
+        assert_eq!(percentile_of_sorted(&v, 0.99), Some(100_000));
+    }
+
+    #[test]
+    fn percentile_single_lone_outlier_escapes_p99() {
+        // Documents the known limit: one slow connector of 100 sits above p99.
+        let mut v = vec![1_000u64; 99];
+        v.push(900_000);
+        v.sort_unstable();
+        // p99 (rank 99) is still the fast majority; only max would catch it.
+        assert_eq!(percentile_of_sorted(&v, 0.99), Some(1_000));
+        assert_eq!(percentile_of_sorted(&v, 1.0), Some(900_000));
+    }
+}
diff --git a/crates/feldera-types/src/time_series.rs b/crates/feldera-types/src/time_series.rs
@@ -34,4 +34,22 @@ pub struct SampleStatistics {
     /// Storage usage in bytes.
     #[serde(rename = "s")]
     pub storage_bytes: u64,
+
+    /// Processing latency (ingest to circuit-processed), microseconds:
+    /// p50 across connectors of each connector's median. Absent without samples.
+    #[serde(rename = "pp50", skip_serializing_if = "Option::is_none", default)]
+    pub processing_latency_p50_micros: Option<u64>,
+
+    /// Processing latency, microseconds: p99 across connectors.
+    #[serde(rename = "pp99", skip_serializing_if = "Option::is_none", default)]
+    pub processing_latency_p99_micros: Option<u64>,
+
+    /// Completion latency (ingest to all outputs pushed), microseconds:
+    /// p50 across connectors.
+    #[serde(rename = "cp50", skip_serializing_if = "Option::is_none", default)]
+    pub completion_latency_p50_micros: Option<u64>,
+
+    /// Completion latency, microseconds: p99 across connectors.
+    #[serde(rename = "cp99", skip_serializing_if = "Option::is_none", default)]
+    pub completion_latency_p99_micros: Option<u64>,
 }
diff --git a/crates/storage/src/histogram.rs b/crates/storage/src/histogram.rs
@@ -115,6 +115,25 @@ impl ExponentialHistogramSnapshot {
     pub fn sum(&self) -> u64 {
         self.sum
     }
+
+    /// Approximate `q`-quantile (`q` in `0.0..=1.0`), as the lower bound of the
+    /// containing bucket. `None` if empty.
+    pub fn quantile(&self, q: f64) -> Option<u64> {
+        let total: u64 = self.buckets.iter().sum();
+        if total == 0 {
+            return None;
+        }
+        let rank = ((q.clamp(0.0, 1.0) * total as f64).ceil() as u64).clamp(1, total);
+        let mut cumulative = 0u64;
+        for (index, count) in self.buckets.iter().enumerate() {
+            cumulative += *count;
+            if cumulative >= rank {
+                return Some(*bucket_to_range(index).start());
+            }
+        }
+        // Unreachable: `cumulative` reaches `total >= rank` in the loop.
+        Some(*bucket_to_range(N_BUCKETS - 1).start())
+    }
 }
 
 pub struct Bucket {
@@ -278,7 +297,52 @@ impl SlidingHistogram {
 
 #[cfg(test)]
 mod test {
-    use crate::histogram::{N_BUCKETS, bucket_to_range, number_to_bucket};
+    use crate::histogram::{ExponentialHistogram, N_BUCKETS, bucket_to_range, number_to_bucket};
+
+    #[test]
+    fn quantile_empty() {
+        let hist = ExponentialHistogram::new();
+        assert_eq!(hist.snapshot().quantile(0.5), None);
+        assert_eq!(hist.snapshot().quantile(0.99), None);
+    }
+
+    #[test]
+    fn quantile_single_sample() {
+        let hist = ExponentialHistogram::new();
+        hist.record(42u64);
+        // 42 lands in the bucket for [40, 49]; the lower bound is 40.
+        assert_eq!(hist.snapshot().quantile(0.0), Some(40));
+        assert_eq!(hist.snapshot().quantile(0.5), Some(40));
+        assert_eq!(hist.snapshot().quantile(1.0), Some(40));
+    }
+
+    #[test]
+    fn quantile_all_same_bucket() {
+        let hist = ExponentialHistogram::new();
+        for _ in 0..100 {
+            hist.record(5u64);
+        }
+        // Values 0..=9 each occupy their own bucket, so 5 is exact.
+        assert_eq!(hist.snapshot().quantile(0.5), Some(5));
+        assert_eq!(hist.snapshot().quantile(0.99), Some(5));
+    }
+
+    #[test]
+    fn quantile_known_distribution() {
+        let hist = ExponentialHistogram::new();
+        // 99 fast samples (bucket [0,0]) and 1 slow sample (bucket [900,999]).
+        for _ in 0..99 {
+            hist.record(0u64);
+        }
+        hist.record(950u64);
+        let snap = hist.snapshot();
+        // p50 stays with the fast majority.
+        assert_eq!(snap.quantile(0.5), Some(0));
+        // p99 (ceil(0.99*100)=99th value) is still the last fast sample.
+        assert_eq!(snap.quantile(0.99), Some(0));
+        // p100 reaches the lone slow sample: bucket [900, 999], lower bound 900.
+        assert_eq!(snap.quantile(1.0), Some(900));
+    }
 
     #[test]
     fn buckets() {