nodedb_lite/engine/timeseries/
query_routing.rs

1//! Timeseries query routing: Local, Cloud, or Hybrid execution.
2//!
3//! Lite holds its own agent's data. Origin holds aggregated data from
4//! all agents. Query routing determines where execution happens.
5//!
6//! - **Local** (default): Lite columnar engine scans local partitions only.
7//!   No network needed. No `__source` tag — Lite only has its own data.
8//!
9//! - **Cloud**: Lite forwards SQL to Origin via sync WebSocket. Origin
10//!   auto-filters by `__source=<this_lite_id>` unless the query explicitly
11//!   asks for fleet-wide data (e.g., `GROUP BY __source`).
12//!
13//! - **Hybrid**: Local data + pre-computed fleet aggregates from Origin
14//!   via shape subscriptions. Returns local + stale shape data with
15//!   `shape_staleness_ms` indicator. No per-query round-trip for fleet data.
16
17use std::collections::HashMap;
18
19use serde::{Deserialize, Serialize};
20
21use nodedb_types::timeseries::{SeriesId, TimeRange};
22
23/// Query execution scope.
24#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
25pub enum QueryScope {
26    /// Execute locally on Lite engine only. Default. No network.
27    #[default]
28    Local,
29    /// Forward to Origin via sync WebSocket. Requires connectivity.
30    Cloud,
31    /// Local data + cached fleet aggregates from shape subscriptions.
32    /// Works offline with stale fleet data.
33    Hybrid,
34}
35
36/// A timeseries shape subscription: Origin pushes pre-computed
37/// downsampled aggregates to Lite at a fixed interval.
38///
39/// Example: `fleet_avg_cpu` at 5m resolution — Lite receives the
40/// fleet-wide average CPU every 5 minutes without per-query round-trips.
41#[derive(Debug, Clone, Serialize, Deserialize)]
42pub struct TimeseriesShape {
43    /// Shape identifier.
44    pub shape_id: String,
45    /// Collection on Origin.
46    pub collection: String,
47    /// Metric to aggregate.
48    pub metric: String,
49    /// Tags to group by (empty = aggregate all devices).
50    pub group_by: Vec<String>,
51    /// Aggregation function: "avg", "sum", "min", "max", "count".
52    pub aggregate: String,
53    /// Bucket interval in milliseconds (e.g., 300_000 = 5 minutes).
54    pub interval_ms: u64,
55}
56
57/// Cached shape data received from Origin via sync.
58///
59/// Stored locally as read-only series, updated by each shape push.
60#[derive(Debug, Clone)]
61pub struct CachedShapeData {
62    pub shape: TimeseriesShape,
63    /// Cached aggregated buckets: `(bucket_start_ms, value)`.
64    pub buckets: Vec<(i64, f64)>,
65    /// When this cache was last updated (epoch ms).
66    pub last_updated_ms: u64,
67    /// Max timestamp in cached data.
68    pub max_ts: i64,
69}
70
71impl CachedShapeData {
72    /// How stale the cached data is, relative to `now_ms`.
73    pub fn staleness_ms(&self, now_ms: u64) -> u64 {
74        now_ms.saturating_sub(self.last_updated_ms)
75    }
76}
77
78/// Manages timeseries shape subscriptions and cached fleet data on Lite.
79#[derive(Debug, Default)]
80pub struct TimeseriesShapeManager {
81    /// Active shape subscriptions: `shape_id → cached data`.
82    shapes: HashMap<String, CachedShapeData>,
83}
84
85impl TimeseriesShapeManager {
86    pub fn new() -> Self {
87        Self::default()
88    }
89
90    /// Register a shape subscription.
91    pub fn subscribe(&mut self, shape: TimeseriesShape) {
92        let shape_id = shape.shape_id.clone();
93        self.shapes.insert(
94            shape_id,
95            CachedShapeData {
96                shape,
97                buckets: Vec::new(),
98                last_updated_ms: 0,
99                max_ts: 0,
100            },
101        );
102    }
103
104    /// Unsubscribe from a shape.
105    pub fn unsubscribe(&mut self, shape_id: &str) {
106        self.shapes.remove(shape_id);
107    }
108
109    /// Update cached data from an Origin shape push.
110    pub fn update(&mut self, shape_id: &str, buckets: Vec<(i64, f64)>, now_ms: u64) {
111        if let Some(cached) = self.shapes.get_mut(shape_id) {
112            cached.max_ts = buckets.iter().map(|(ts, _)| *ts).max().unwrap_or(0);
113            cached.buckets = buckets;
114            cached.last_updated_ms = now_ms;
115        }
116    }
117
118    /// Query cached fleet data for a shape.
119    ///
120    /// Returns `(buckets, staleness_ms)`. Staleness = 0 means fresh data.
121    /// If the shape doesn't exist, returns empty with max staleness.
122    pub fn query(&self, shape_id: &str, range: &TimeRange, now_ms: u64) -> (Vec<(i64, f64)>, u64) {
123        match self.shapes.get(shape_id) {
124            Some(cached) => {
125                let filtered: Vec<(i64, f64)> = cached
126                    .buckets
127                    .iter()
128                    .filter(|(ts, _)| range.contains(*ts))
129                    .copied()
130                    .collect();
131                (filtered, cached.staleness_ms(now_ms))
132            }
133            None => (Vec::new(), u64::MAX),
134        }
135    }
136
137    /// List active shape subscriptions.
138    pub fn active_shapes(&self) -> Vec<&TimeseriesShape> {
139        self.shapes.values().map(|c| &c.shape).collect()
140    }
141
142    /// Number of active subscriptions.
143    pub fn len(&self) -> usize {
144        self.shapes.len()
145    }
146
147    pub fn is_empty(&self) -> bool {
148        self.shapes.is_empty()
149    }
150
151    /// Export for persistence (redb serialization).
152    pub fn export(&self) -> Vec<(String, CachedShapeData)> {
153        self.shapes
154            .iter()
155            .map(|(k, v)| (k.clone(), v.clone()))
156            .collect()
157    }
158
159    /// Import from persistence.
160    pub fn import(&mut self, entries: Vec<(String, CachedShapeData)>) {
161        for (k, v) in entries {
162            self.shapes.insert(k, v);
163        }
164    }
165}
166
167/// Result of a hybrid query: local data + fleet shape data.
168#[derive(Debug)]
169pub struct HybridQueryResult {
170    /// Local scan results: `(timestamp, value, series_id)`.
171    pub local: Vec<(i64, f64, SeriesId)>,
172    /// Fleet aggregate from shape cache: `(bucket_start, value)`.
173    pub fleet: Vec<(i64, f64)>,
174    /// Staleness of the fleet data in milliseconds.
175    /// 0 = fresh. u64::MAX = no shape subscription.
176    pub shape_staleness_ms: u64,
177    /// Whether local data was available (always true for Local scope).
178    pub local_available: bool,
179}
180
181/// Parameters for a routed timeseries query.
182pub struct RoutedQueryParams<'a> {
183    pub scope: QueryScope,
184    pub collection: &'a str,
185    pub range: &'a TimeRange,
186    pub bucket_ms: Option<i64>,
187    pub shape_id: Option<&'a str>,
188    pub now_ms: u64,
189}
190
191/// Execute a timeseries query with routing based on `QueryScope`.
192///
193/// - `Local`: scans the local `TimeseriesEngine`.
194/// - `Cloud`: forwards SQL to Origin via `NodeDbRemote` (pgwire). Requires
195///   an active pgwire connection — NOT the sync WebSocket. Query and sync
196///   are separate channels (production pattern).
197/// - `Hybrid`: local scan + cached shape data. No per-query network hop.
198///   Fleet data degrades gracefully with staleness indicator.
199pub fn execute_routed_query(
200    params: &RoutedQueryParams<'_>,
201    engine: &super::engine::TimeseriesEngine,
202    shape_mgr: &TimeseriesShapeManager,
203) -> HybridQueryResult {
204    let RoutedQueryParams {
205        scope,
206        collection,
207        range,
208        bucket_ms,
209        shape_id,
210        now_ms,
211    } = params;
212    let _ = bucket_ms; // Reserved for future aggregation pushdown.
213
214    let local = match scope {
215        QueryScope::Local | QueryScope::Hybrid => engine.scan(collection, range),
216        QueryScope::Cloud => {
217            // Cloud-only: no local scan. Caller must forward SQL
218            // via NodeDbRemote::execute_sql() over pgwire separately.
219            Vec::new()
220        }
221    };
222
223    let (fleet, shape_staleness_ms) = match scope {
224        QueryScope::Hybrid => {
225            if let Some(sid) = shape_id {
226                shape_mgr.query(sid, range, *now_ms)
227            } else {
228                (Vec::new(), u64::MAX)
229            }
230        }
231        _ => (Vec::new(), u64::MAX),
232    };
233
234    HybridQueryResult {
235        local_available: !matches!(scope, QueryScope::Cloud),
236        local,
237        fleet,
238        shape_staleness_ms,
239    }
240}
241
242#[cfg(test)]
243mod tests {
244    use super::*;
245
246    fn make_shape(id: &str) -> TimeseriesShape {
247        TimeseriesShape {
248            shape_id: id.into(),
249            collection: "metrics".into(),
250            metric: "cpu_usage".into(),
251            group_by: vec![],
252            aggregate: "avg".into(),
253            interval_ms: 300_000,
254        }
255    }
256
257    #[test]
258    fn subscribe_and_query() {
259        let mut mgr = TimeseriesShapeManager::new();
260        mgr.subscribe(make_shape("fleet_cpu"));
261        assert_eq!(mgr.len(), 1);
262
263        // No data yet — empty result.
264        let (buckets, staleness) = mgr.query("fleet_cpu", &TimeRange::new(0, 1_000_000), 1000);
265        assert!(buckets.is_empty());
266        assert_eq!(staleness, 1000); // 1000 - 0 = 1000ms stale
267
268        // Update with data.
269        mgr.update(
270            "fleet_cpu",
271            vec![(300_000, 45.0), (600_000, 52.0), (900_000, 48.0)],
272            1_000_000,
273        );
274
275        let (buckets, staleness) = mgr.query("fleet_cpu", &TimeRange::new(0, 1_000_000), 1_000_000);
276        assert_eq!(buckets.len(), 3);
277        assert_eq!(staleness, 0); // Just updated.
278    }
279
280    #[test]
281    fn unsubscribe_removes() {
282        let mut mgr = TimeseriesShapeManager::new();
283        mgr.subscribe(make_shape("s1"));
284        assert_eq!(mgr.len(), 1);
285        mgr.unsubscribe("s1");
286        assert_eq!(mgr.len(), 0);
287    }
288
289    #[test]
290    fn query_range_filtering() {
291        let mut mgr = TimeseriesShapeManager::new();
292        mgr.subscribe(make_shape("s1"));
293        mgr.update(
294            "s1",
295            vec![(100, 1.0), (200, 2.0), (300, 3.0), (400, 4.0)],
296            500,
297        );
298
299        // Query range [200, 300].
300        let (buckets, _) = mgr.query("s1", &TimeRange::new(200, 300), 500);
301        assert_eq!(buckets.len(), 2);
302        assert_eq!(buckets[0], (200, 2.0));
303        assert_eq!(buckets[1], (300, 3.0));
304    }
305
306    #[test]
307    fn missing_shape_returns_max_staleness() {
308        let mgr = TimeseriesShapeManager::new();
309        let (buckets, staleness) = mgr.query("nonexistent", &TimeRange::new(0, 1000), 1000);
310        assert!(buckets.is_empty());
311        assert_eq!(staleness, u64::MAX);
312    }
313
314    #[test]
315    fn export_import_roundtrip() {
316        let mut mgr = TimeseriesShapeManager::new();
317        mgr.subscribe(make_shape("s1"));
318        mgr.update("s1", vec![(100, 42.0)], 200);
319
320        let exported = mgr.export();
321        let mut mgr2 = TimeseriesShapeManager::new();
322        mgr2.import(exported);
323        assert_eq!(mgr2.len(), 1);
324
325        let (buckets, _) = mgr2.query("s1", &TimeRange::new(0, 1000), 200);
326        assert_eq!(buckets.len(), 1);
327    }
328
329    #[test]
330    fn query_scope_default_is_local() {
331        assert_eq!(QueryScope::default(), QueryScope::Local);
332    }
333}
nodedb_lite/engine/timeseries/query_routing.rs

nodedb_lite/engine/timeseries/
query_routing.rs
