feldera · mihaibudiu · Aug 13, 2026 · Jul 16, 2026 · Jul 16, 2026 · Aug 10, 2026
diff --git a/bun.lock b/bun.lock
diff --git a/crates/adapters/src/controller/stats.rs b/crates/adapters/src/controller/stats.rs
@@ -1641,8 +1641,21 @@ impl InputEndpointMetrics {
             num_transport_errors: self.num_transport_errors.load(Ordering::Relaxed),
             num_parse_errors: self.num_parse_errors.load(Ordering::Relaxed),
             end_of_input: self.end_of_input.load(Ordering::Relaxed),
+            processing_latency_p99_micros: self.processing_latency_p99_micros(),
         }
     }
+
+    /// 99th percentile processing latency in microseconds over the sliding
+    /// histogram's window, or `None` if the endpoint has no samples. See
+    /// [`ExternalInputEndpointMetrics::processing_latency_p99_micros`] for what
+    /// the window covers.
+    fn processing_latency_p99_micros(&self) -> Option<u64> {
+        self.processing_latency_micros_histogram
+            .lock()
+            .ok()?
+            .snapshot()
+            .quantile(0.99)
+    }
 }
 
 // Latency histogram creation functions.
@@ -3026,3 +3039,85 @@ impl OutputEndpointStatus {
         self.metrics.total_processed_steps.load(Ordering::Acquire)
     }
 }
+
+#[cfg(test)]
+mod test {
+    use super::InputEndpointMetrics;
+
+    #[test]
+    fn latency_p99_absent_without_samples() {
+        let metrics = InputEndpointMetrics::default();
+        assert_eq!(metrics.to_api_type().processing_latency_p99_micros, None);
+    }
+
+    #[test]
+    fn latency_p99_reports_uniform_bucket() {
+        let metrics = InputEndpointMetrics::default();
+        {
+            let mut histogram = metrics.processing_latency_micros_histogram.lock().unwrap();
+            for _ in 0..100 {
+                histogram.record(1_500u64);
+            }
+        }
+        // Bucket [1000, 1999] holds 1_500, and `quantile` reports its lower bound.
+        assert_eq!(
+            metrics.to_api_type().processing_latency_p99_micros,
+            Some(1_000)
+        );
+    }
+
+    /// Guards against reporting the median, which hides the slow tail the
+    /// column exists to surface.
+    #[test]
+    fn latency_p99_reports_slow_tail_not_median() {
+        let metrics = InputEndpointMetrics::default();
+        {
+            let mut histogram = metrics.processing_latency_micros_histogram.lock().unwrap();
+            // 90 fast samples against 10 slow ones: the median sits in the fast
+            // band, rank 99 (ceil(0.99 * 100)) lands in the slow one.
+            for _ in 0..90 {
+                histogram.record(1_500u64);
+            }
+            for _ in 0..10 {
+                histogram.record(250_000u64);
+            }
+        }
+        // Bucket [200000, 299999] holds 250_000, and `quantile` reports its
+        // lower bound. A median would report 1_000 here.
+        assert_eq!(
+            metrics.to_api_type().processing_latency_p99_micros,
+            Some(200_000)
+        );
+    }
+
+    /// Documents the known limit of a percentile over few samples: one slow
+    /// batch in a hundred stays below rank 99.
+    #[test]
+    fn latency_p99_misses_lone_outlier() {
+        let metrics = InputEndpointMetrics::default();
+        {
+            let mut histogram = metrics.processing_latency_micros_histogram.lock().unwrap();
+            for _ in 0..99 {
+                histogram.record(1_500u64);
+            }
+            histogram.record(250_000u64);
+        }
+        assert_eq!(
+            metrics.to_api_type().processing_latency_p99_micros,
+            Some(1_000)
+        );
+    }
+
+    /// Guards against reporting the completion histogram, which tracks a
+    /// different span and would silently change the column's meaning.
+    #[test]
+    fn latency_p99_ignores_completion_histogram() {
+        let metrics = InputEndpointMetrics::default();
+        metrics
+            .completion_latency_micros_histogram
+            .lock()
+            .unwrap()
+            .record(750_000u64);
+        assert_eq!(metrics.to_api_type().processing_latency_p99_micros, None);
+    }
+}
diff --git a/crates/feldera-types/src/adapter_stats.rs b/crates/feldera-types/src/adapter_stats.rs
@@ -215,6 +215,23 @@ pub struct ExternalInputEndpointMetrics {
     pub num_parse_errors: u64,
     /// True if end-of-input has been signaled.
     pub end_of_input: bool,
+    /// 99th percentile processing latency (from ingesting a batch to finishing processing it) in microseconds.
+    ///
+    /// The time from ingesting a batch of records off the wire
+    /// to the circuit finishing processing them,
+    /// covering parsing, queuing, and the circuit step.
+    ///
+    /// Does not account for completion latency.
+    ///
+    /// Taken over the endpoint's sliding histogram, which holds the 10,000 most
+    /// recent samples spanning at most 10 minutes, whichever bound is reached
+    /// first. One sample is recorded per completed batch. An endpoint that stops
+    /// ingesting keeps reporting its last known latency instead of dropping to
+    /// `None`.
+    ///
+    /// `None` until the endpoint records its first sample.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub processing_latency_p99_micros: Option<u64>,
 }
 
 /// Input endpoint status information.

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() {

diff --git a/js-packages/web-console/openapi-ts.config.ts b/js-packages/web-console/openapi-ts.config.ts
@@ -1,8 +1,30 @@
 import { defineConfig } from '@hey-api/openapi-ts'
+import {
+  type CustomTypes,
+  customTypesPlugin,
+  overrideOpenapiType
+} from './src/lib/functions/common/openapi-ts'
+
+/**
+ * Hand-written types substituted for the generated ones, keyed by the custom
+ * `format` that marks a schema.
+ */
+const customTypes = {
+  microseconds: { module: '$lib/functions/common/duration', name: 'Microseconds' }
+} as const satisfies CustomTypes
 
 export default defineConfig({
   input: '../../openapi.json',
   output: './src/lib/services/manager',
+  parser: {
+    patch: {
+      schemas: {
+        InputEndpointMetrics: (schema) => {
+          overrideOpenapiType(schema, 'processing_latency_p99_micros', 'microseconds')
+        }
+      }
+    }
+  },
   plugins: [
     {
       name: '@hey-api/client-fetch',
@@ -12,6 +34,7 @@ export default defineConfig({
     {
       name: '@hey-api/sdk',
       responseStyle: 'data'
-    }
+    },
+    customTypesPlugin(customTypes)
   ]
 })
diff --git a/js-packages/web-console/package.json b/js-packages/web-console/package.json
@@ -10,7 +10,7 @@
     "@fontsource/dm-mono": "5.2.7",
     "@fortawesome/fontawesome-free": "7.2.0",
     "@hey-api/client-fetch": "0.13.1",
-    "@hey-api/openapi-ts": "0.97.3",
+    "@hey-api/openapi-ts": "0.99.0",
     "@monaco-editor/loader": "1.7.0",
     "@playwright/test": "1.58.2",
     "@poppanator/sveltekit-svg": "6.0.1",

diff --git a/js-packages/web-console/src/assets/icons/feldera-material-icons/circle-help.svg b/js-packages/web-console/src/assets/icons/feldera-material-icons/circle-help.svg