dwara_core/analytics/
rollup.rs

1//! Rollup cascade and retention (DW-043).
2//!
3//! Four granularities roll in a fixed cascade — raw → 1m → 5m → 1h →
4//! 1d — each stage aggregating the PREVIOUS stage's completed windows,
5//! never raw twice (except by deterministic recompute, see below):
6//!
7//! - A window is COMPLETE when its end plus a safety grace (writer
8//!   lag headroom) is in the past; the grace means a straggler record
9//!   flushed late still lands in the window it belongs to. Rows
10//!   arriving later than the grace are documented as rollup-lost
11//!   (they remain queryable in `raw` until raw retention expires).
12//! - Each granularity keeps a CURSOR (exclusive ms frontier) in
13//!   `meta`, advanced in the SAME transaction as the rows it covers:
14//!   a crash mid-cascade never double-counts.
15//! - Every aggregation is a wholesale window RECOMPUTE
16//!   (`INSERT OR REPLACE` over the window's full source range), so
17//!   re-running any window — after a crash, a restored backup, or a
18//!   cursor reset — reproduces byte-identical rows. Idempotence comes
19//!   from determinism, not from merge arithmetic (upsert-merge would
20//!   double-count on re-run).
21//! - Custom-dimension rollups aggregate the same raw range in RUST
22//!   (the source column is a JSON object; keeping JSON out of SQL
23//!   avoids depending on the bundled SQLite's JSON1 build flags).
24//!
25//! Retention deletes expired rows per granularity (raw by `ts_ms`,
26//! rollups by `window_start`) and incrementally vacuums, which is the
27//! bounded-disk half of the done-when: pages return to the file
28//! system without a full VACUUM ever blocking the writer.
29
30use std::collections::BTreeMap;
31
32use rusqlite::Connection;
33
34use super::schema::{bucket_of, BUCKET_COLS};
35
36/// The four granularities, in cascade order (index = the `gran`
37/// column value stored in the rollup tables).
38pub const GRANULARITIES_MS: [i64; 4] = [60_000, 300_000, 3_600_000, 86_400_000];
39
40/// Meta key for a granularity's cursor (`g0`..`g3`; `g0` is the
41/// raw→1m frontier).
42fn cursor_key(gran: usize) -> String {
43    format!("g{gran}_cursor_ms")
44}
45
46/// Read a cursor (0 when never set).
47pub fn cursor(conn: &Connection, gran: usize) -> i64 {
48    conn.query_row(
49        "SELECT value FROM meta WHERE key = ?1",
50        [cursor_key(gran)],
51        |r| r.get::<_, String>(0),
52    )
53    .ok()
54    .and_then(|v| v.parse().ok())
55    .unwrap_or(0)
56}
57
58fn set_cursor_tx(conn: &Connection, gran: usize, frontier: i64) -> rusqlite::Result<()> {
59    conn.execute(
60        "INSERT INTO meta (key, value) VALUES (?1, ?2)
61         ON CONFLICT(key) DO UPDATE SET value = excluded.value",
62        rusqlite::params![cursor_key(gran), frontier.to_string()],
63    )?;
64    Ok(())
65}
66
67/// Window start (floor to granularity) of a timestamp.
68pub fn window_of(ts_ms: i64, gran_ms: i64) -> i64 {
69    (ts_ms / gran_ms) * gran_ms
70}
71
72/// One raw→1m pass: aggregate raw rows in `[cursor, frontier)` into
73/// completed 1-minute windows, recompute-style. `frontier` must be the
74/// end of the last COMPLETE window (the caller computes it with the
75/// grace). Returns the number of 1m windows written.
76pub fn roll_raw_to_1m(conn: &Connection, now_ms: i64, grace_ms: i64) -> rusqlite::Result<usize> {
77    let gran_ms = GRANULARITIES_MS[0];
78    let frontier = window_of(now_ms.saturating_sub(grace_ms), gran_ms);
79    let cur = cursor(conn, 0);
80    if frontier <= cur {
81        return Ok(0);
82    }
83    let tx = conn.unchecked_transaction()?;
84    let n = roll_raw_range(&tx, cur, frontier)?;
85    set_cursor_tx(&tx, 0, frontier)?;
86    tx.commit()?;
87    Ok(n)
88}
89
90/// Aggregate raw `[lo, hi)` into `rollup_fixed`/`rollup_dim` at
91/// granularity 0, wholesale over the range (idempotent recompute).
92/// Split from [`roll_raw_to_1m`] so tests can drive exact ranges.
93pub fn roll_raw_range(conn: &Connection, lo: i64, hi: i64) -> rusqlite::Result<usize> {
94    let windows = roll_raw_fixed(conn, lo, hi)?;
95    roll_raw_dims(conn, lo, hi)?;
96    Ok(windows)
97}
98
99/// The fixed-dimension pass: pure SQL GROUP BY with per-row bucket
100/// one-hots. Each request contributes to exactly one bucket column,
101/// so SUM(b_i) is the non-cumulative bucket histogram.
102fn roll_raw_fixed(conn: &Connection, lo: i64, hi: i64) -> rusqlite::Result<usize> {
103    let before: i64 = conn.query_row(
104        "SELECT COUNT(DISTINCT (ts_ms / 60000)) FROM raw_all WHERE ts_ms >= ?1 AND ts_ms < ?2",
105        [lo, hi],
106        |r| r.get(0),
107    )?;
108    conn.execute(
109        "INSERT OR REPLACE INTO rollup_fixed (
110            gran, window_start, listener, route, upstream, consumer,
111            method, status_class, requests, errors, rate_limited, shed,
112            duration_sum_ms, duration_max_ms,
113            b0, b1, b2, b3, b4, b5, b6, b7, b8, b9, b10, b11, b12
114        )
115        SELECT 0,
116            (ts_ms / 60000) * 60000,
117            listener, route, upstream, consumer, method, status_class,
118            COUNT(*),
119            SUM(CASE WHEN status >= 500 THEN 1 ELSE 0 END),
120            SUM(rate_limited), SUM(shed),
121            SUM(duration_ms), MAX(duration_ms),
122            SUM(CASE WHEN duration_ms <= 1.0 THEN 1 ELSE 0 END),
123            SUM(CASE WHEN duration_ms > 1.0 AND duration_ms <= 2.0 THEN 1 ELSE 0 END),
124            SUM(CASE WHEN duration_ms > 2.0 AND duration_ms <= 5.0 THEN 1 ELSE 0 END),
125            SUM(CASE WHEN duration_ms > 5.0 AND duration_ms <= 10.0 THEN 1 ELSE 0 END),
126            SUM(CASE WHEN duration_ms > 10.0 AND duration_ms <= 25.0 THEN 1 ELSE 0 END),
127            SUM(CASE WHEN duration_ms > 25.0 AND duration_ms <= 50.0 THEN 1 ELSE 0 END),
128            SUM(CASE WHEN duration_ms > 50.0 AND duration_ms <= 100.0 THEN 1 ELSE 0 END),
129            SUM(CASE WHEN duration_ms > 100.0 AND duration_ms <= 250.0 THEN 1 ELSE 0 END),
130            SUM(CASE WHEN duration_ms > 250.0 AND duration_ms <= 500.0 THEN 1 ELSE 0 END),
131            SUM(CASE WHEN duration_ms > 500.0 AND duration_ms <= 1000.0 THEN 1 ELSE 0 END),
132            SUM(CASE WHEN duration_ms > 1000.0 AND duration_ms <= 2500.0 THEN 1 ELSE 0 END),
133            SUM(CASE WHEN duration_ms > 2500.0 AND duration_ms <= 5000.0 THEN 1 ELSE 0 END),
134            SUM(CASE WHEN duration_ms > 5000.0 THEN 1 ELSE 0 END)
135        FROM raw_all
136        WHERE ts_ms >= ?1 AND ts_ms < ?2
137        GROUP BY (ts_ms / 60000), listener, route, upstream, consumer,
138                 method, status_class",
139        [lo, hi],
140    )?;
141    Ok(before as usize)
142}
143
144/// The custom-dimension pass, in Rust: one scan of the range's
145/// (ts, dims, status, duration) rows, grouped by
146/// (minute, dim name, value). Dims rows do NOT split by fixed
147/// dimensions — a dimension value aggregates every request carrying
148/// it (that is the point of a custom dimension).
149fn roll_raw_dims(conn: &Connection, lo: i64, hi: i64) -> rusqlite::Result<()> {
150    #[derive(Default)]
151    struct Acc {
152        requests: i64,
153        errors: i64,
154        duration_sum_ms: f64,
155        buckets: Vec<i64>,
156    }
157    let mut groups: BTreeMap<(i64, String, String), Acc> = BTreeMap::new();
158    {
159        let mut stmt = conn.prepare(
160            "SELECT ts_ms, dims, status, duration_ms FROM raw_all
161             WHERE ts_ms >= ?1 AND ts_ms < ?2",
162        )?;
163        let mut rows = stmt.query([lo, hi])?;
164        while let Some(row) = rows.next()? {
165            let ts: i64 = row.get(0)?;
166            let dims_json: String = row.get(1)?;
167            let status: i64 = row.get(2)?;
168            let duration_ms: f64 = row.get(3)?;
169            let Ok(map) =
170                serde_json::from_str::<serde_json::Map<String, serde_json::Value>>(&dims_json)
171            else {
172                continue;
173            };
174            let minute = window_of(ts, 60_000);
175            for (dim, value) in map {
176                let Some(value) = value.as_str().map(str::to_string) else {
177                    continue;
178                };
179                let acc = groups
180                    .entry((minute, dim.clone(), value))
181                    .or_insert_with(|| Acc {
182                        buckets: vec![0; BUCKET_COLS],
183                        ..Acc::default()
184                    });
185                acc.requests += 1;
186                if status >= 500 {
187                    acc.errors += 1;
188                }
189                acc.duration_sum_ms += duration_ms;
190                acc.buckets[bucket_of(duration_ms)] += 1;
191            }
192        }
193    }
194    let mut stmt = conn.prepare(
195        "INSERT OR REPLACE INTO rollup_dim (
196            gran, window_start, dim, value, requests, errors,
197            duration_sum_ms, b0, b1, b2, b3, b4, b5, b6, b7, b8, b9,
198            b10, b11, b12
199        ) VALUES (0, ?1, ?2, ?3, ?4, ?5, ?6,
200                  ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17,
201                  ?18, ?19)",
202    )?;
203    for ((minute, dim, value), acc) in groups {
204        let mut params: Vec<Box<dyn rusqlite::ToSql>> = vec![
205            Box::new(minute),
206            Box::new(dim),
207            Box::new(value),
208            Box::new(acc.requests),
209            Box::new(acc.errors),
210            Box::new(acc.duration_sum_ms),
211        ];
212        for b in &acc.buckets {
213            params.push(Box::new(*b));
214        }
215        stmt.execute(rusqlite::params_from_iter(
216            params.iter().map(|p| p.as_ref()),
217        ))?;
218    }
219    Ok(())
220}
221
222/// One cascade stage: aggregate granularity `from`'s completed windows
223/// into granularity `to` (`from + 1`). Wholesale recompute per coarse
224/// window, cursor-guarded exactly like the raw pass.
225pub fn roll_cascade(
226    conn: &Connection,
227    from: usize,
228    now_ms: i64,
229    grace_ms: i64,
230) -> rusqlite::Result<usize> {
231    debug_assert!(from + 1 < GRANULARITIES_MS.len());
232    let coarse = GRANULARITIES_MS[from + 1];
233    let frontier = window_of(now_ms.saturating_sub(grace_ms), coarse);
234    let cur = cursor(conn, from + 1);
235    if frontier <= cur {
236        return Ok(0);
237    }
238    let tx = conn.unchecked_transaction()?;
239    let n = cascade_range(&tx, from, cur, frontier)?;
240    set_cursor_tx(&tx, from + 1, frontier)?;
241    tx.commit()?;
242    Ok(n)
243}
244
245/// Aggregate the finer table's `[lo, hi)` into the next coarser
246/// granularity, wholesale. Every stored column merges by SUM (or MAX
247/// for the max), which is exactly why the schema stores NON-cumulative
248/// bucket counts and additive sums.
249pub fn cascade_range(conn: &Connection, from: usize, lo: i64, hi: i64) -> rusqlite::Result<usize> {
250    let coarse = GRANULARITIES_MS[from + 1];
251    let before: i64 = conn.query_row(
252        "SELECT COUNT(DISTINCT (window_start / ?3)) FROM rollup_fixed
253         WHERE gran = ?1 AND window_start >= ?2 AND window_start < ?4",
254        rusqlite::params![from as i64, lo, coarse, hi],
255        |r| r.get(0),
256    )?;
257    conn.execute(
258        "INSERT OR REPLACE INTO rollup_fixed (
259            gran, window_start, listener, route, upstream, consumer,
260            method, status_class, requests, errors, rate_limited, shed,
261            duration_sum_ms, duration_max_ms,
262            b0, b1, b2, b3, b4, b5, b6, b7, b8, b9, b10, b11, b12
263        )
264        SELECT ?4,
265            (window_start / ?5) * ?5,
266            listener, route, upstream, consumer, method, status_class,
267            SUM(requests), SUM(errors), SUM(rate_limited), SUM(shed),
268            SUM(duration_sum_ms), MAX(duration_max_ms),
269            SUM(b0), SUM(b1), SUM(b2), SUM(b3), SUM(b4), SUM(b5),
270            SUM(b6), SUM(b7), SUM(b8), SUM(b9), SUM(b10), SUM(b11),
271            SUM(b12)
272        FROM rollup_fixed
273        WHERE gran = ?1 AND window_start >= ?2 AND window_start < ?3
274        GROUP BY (window_start / ?5), listener, route, upstream,
275                 consumer, method, status_class",
276        rusqlite::params![from as i64, lo, hi, (from + 1) as i64, coarse],
277    )?;
278    // Custom dims cascade the same way (their schema is the additive
279    // subset).
280    conn.execute(
281        "INSERT OR REPLACE INTO rollup_dim (
282            gran, window_start, dim, value, requests, errors,
283            duration_sum_ms, b0, b1, b2, b3, b4, b5, b6, b7, b8, b9,
284            b10, b11, b12
285        )
286        SELECT ?4,
287            (window_start / ?5) * ?5,
288            dim, value,
289            SUM(requests), SUM(errors), SUM(duration_sum_ms),
290            SUM(b0), SUM(b1), SUM(b2), SUM(b3), SUM(b4), SUM(b5),
291            SUM(b6), SUM(b7), SUM(b8), SUM(b9), SUM(b10), SUM(b11),
292            SUM(b12)
293        FROM rollup_dim
294        WHERE gran = ?1 AND window_start >= ?2 AND window_start < ?3
295        GROUP BY (window_start / ?5), dim, value",
296        rusqlite::params![from as i64, lo, hi, (from + 1) as i64, coarse],
297    )?;
298    Ok(before as usize)
299}
300
301/// Retention sweep: delete expired rows per table and incrementally
302/// vacuum up to `vacuum_pages` pages. For the `raw` table, only
303/// intra-day rows older than the retention are deleted (SCALE-09
304/// #188: daily partition tables are dropped by
305/// `partition::drop_expired_partitions`, which is O(1) DROP TABLE
306/// vs O(n) DELETE). Returns rows deleted.
307pub fn sweep_retention(
308    conn: &Connection,
309    raw_keep_ms: i64,
310    gran_keep_ms: [i64; 4],
311    now_ms: i64,
312    vacuum_pages: u32,
313) -> rusqlite::Result<usize> {
314    let mut deleted = 0usize;
315    let tx = conn.unchecked_transaction()?;
316    // Only delete from the current raw table (intra-day cleanup).
317    // Old daily partitions are dropped by partition::drop_expired_partitions.
318    deleted += tx.execute(
319        "DELETE FROM raw WHERE ts_ms < ?1",
320        [now_ms.saturating_sub(raw_keep_ms)],
321    )?;
322    for (gran, keep) in gran_keep_ms.iter().enumerate() {
323        let cutoff = now_ms.saturating_sub(*keep);
324        deleted += tx.execute(
325            "DELETE FROM rollup_fixed WHERE gran = ?1 AND window_start < ?2",
326            rusqlite::params![gran as i64, cutoff],
327        )?;
328        deleted += tx.execute(
329            "DELETE FROM rollup_dim WHERE gran = ?1 AND window_start < ?2",
330            rusqlite::params![gran as i64, cutoff],
331        )?;
332    }
333    tx.commit()?;
334    // Vacuum outside the transaction (it is its own).
335    let _ = conn.execute_batch(&format!("PRAGMA incremental_vacuum={vacuum_pages};"));
336    Ok(deleted)
337}
dwara_core/analytics/rollup.rs

dwara_core/analytics/
rollup.rs
