dwara_core/analytics/
partition.rs

1//! Raw table daily partitioning and retention automation (SCALE-09,
2//! #188).
3//!
4//! The embedded analytics store's `raw` table grows unbounded under
5//! high sustained throughput between retention prunes. This module
6//! rotates the raw table daily into partition tables (`raw_YYYYMMDD`)
7//! within the same SQLite database, making retention enforcement O(1)
8//! (DROP TABLE) instead of O(n) (DELETE rows + incremental vacuum).
9//!
10//! # Design
11//!
12//! - The `raw` table is always the CURRENT day's partition. The
13//!   writer inserts into `raw` unchanged.
14//! - At daily boundaries, `maybe_rotate` renames `raw` to
15//!   `raw_YYYYMMDD` (yesterday's date) and creates a new empty `raw`
16//!   table with the full schema.
17//! - A `raw_all` view UNION ALLs the `raw` table and all
18//!   `raw_YYYYMMDD` partition tables, so the rollup and query layers
19//!   read across all partitions without per-query changes.
20//! - `drop_expired_partitions` drops `raw_YYYYMMDD` tables older than
21//!   the raw retention period (O(1) DROP TABLE).
22//! - The `raw_partitions` meta table tracks partition names and their
23//!   date boundaries for introspection.
24//!
25//! # Why same-file partitions, not separate files
26//!
27//! The issue proposed separate SQLite files. Same-file partition
28//! tables achieve the same O(1) retention and bounded raw table size
29//! while avoiding the complexity of ATTACH/DETACH, cross-file
30//! transactions, and connection management. SQLite's DROP TABLE is
31//! O(1) (the pages are simply freed to the file's free list), and
32//! incremental vacuum reclaims them. The `raw_all` view provides
33//! cross-partition querying without dynamic SQL.
34
35use rusqlite::Connection;
36
37/// The `raw_partitions` meta table (schema v11, SCALE-09 #188):
38/// tracks the partition table name and the date it covers. One row
39/// per rotated partition. The current day's `raw` table is NOT listed
40/// here (it is the active partition; it will be listed when it is
41/// rotated).
42pub const SCHEMA_V11: &str = "
43    CREATE TABLE IF NOT EXISTS raw_partitions (
44        partition_name TEXT PRIMARY KEY,
45        date_ymd       TEXT NOT NULL,
46        created_ms     INTEGER NOT NULL
47    ) WITHOUT ROWID;
48";
49
50/// Format a unix-ms timestamp as `YYYYMMDD` in UTC.
51fn ymd_from_ms(ms: i64) -> String {
52    let secs = (ms / 1000).max(0) as u64;
53    let days = secs / 86400;
54    // Civil-from-days algorithm (Howard Hinnant).
55    let z = days as i64 + 719468;
56    let era = if z >= 0 { z } else { z - 146096 } / 146097;
57    let doe = (z - era * 146097) as u64;
58    let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146096) / 365;
59    let y = yoe as i64 + era * 400;
60    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
61    let mp = (5 * doy + 2) / 153;
62    let d = doy - (153 * mp + 2) / 5 + 1;
63    let m = if mp < 10 { mp + 3 } else { mp - 9 };
64    let year = if m <= 2 { y + 1 } else { y };
65    format!("{:04}{:02}{:02}", year, m, d)
66}
67
68/// Check if the current `raw` table has rows from a previous day. If
69/// so, rename it to `raw_YYYYMMDD` (the previous day's date), create
70/// a fresh `raw` table, record the partition, and refresh the
71/// `raw_all` view. Returns the partition name if a rotation
72/// happened, `None` otherwise.
73///
74/// Safe to call at any time (from the maintenance tick). The
75/// rotation is a single transaction: rename, create, record, refresh
76/// view. A crash between the rename and the view refresh leaves the
77/// partition accessible by its explicit name; the next maintenance
78/// tick recreates the view.
79pub fn maybe_rotate(conn: &Connection, now_ms: i64) -> rusqlite::Result<Option<String>> {
80    // Check if the raw table has rows from a previous day.
81    let prev_day_boundary = day_boundary_ms(now_ms);
82    let has_prev_day: i64 = conn.query_row(
83        "SELECT COUNT(*) FROM raw WHERE ts_ms < ?1",
84        [prev_day_boundary],
85        |r| r.get(0),
86    )?;
87    if has_prev_day == 0 {
88        return Ok(None);
89    }
90    // Determine the partition name from the oldest row's date.
91    let oldest_ms: i64 = conn.query_row(
92        "SELECT MIN(ts_ms) FROM raw WHERE ts_ms < ?1",
93        [prev_day_boundary],
94        |r| r.get(0),
95    )?;
96    let partition_date = ymd_from_ms(oldest_ms);
97    let partition_name = format!("raw_{partition_date}");
98    // If the partition already exists (e.g., a re-run after a crash),
99    // merge the rows into it instead of failing.
100    let partition_exists: i64 = conn.query_row(
101        "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name=?1",
102        [&partition_name],
103        |r| r.get(0),
104    )?;
105    let tx = conn.unchecked_transaction()?;
106    if partition_exists > 0 {
107        // Merge: insert old rows from raw into the existing partition,
108        // then delete them from raw.
109        tx.execute_batch(&format!(
110            "INSERT INTO {partition_name} (ts_ms, listener, route, consumer,
111                upstream, method, status, status_class, duration_ms, attempts,
112                rate_limited, broken, shed, dims, request_id, correlation_id,
113                request_headers_redacted, auth_identity)
114            SELECT ts_ms, listener, route, consumer, upstream, method, status,
115                status_class, duration_ms, attempts, rate_limited, broken, shed,
116                dims, request_id, correlation_id, request_headers_redacted,
117                auth_identity
118            FROM raw WHERE ts_ms < {prev_day_boundary};
119            DELETE FROM raw WHERE ts_ms < {prev_day_boundary};"
120        ))?;
121    } else {
122        // Create the partition table with the full schema, move old
123        // rows into it, then delete them from raw. This keeps today's
124        // rows in the active raw table.
125        tx.execute_batch(&format!(
126            "CREATE TABLE {partition_name} AS
127             SELECT * FROM raw WHERE ts_ms < {prev_day_boundary};
128             CREATE INDEX idx_{partition_name}_ts ON {partition_name}(ts_ms);
129             CREATE INDEX idx_{partition_name}_corr
130                 ON {partition_name}(correlation_id, ts_ms);
131             DELETE FROM raw WHERE ts_ms < {prev_day_boundary};"
132        ))?;
133        tx.execute(
134            "INSERT OR REPLACE INTO raw_partitions (partition_name, date_ymd, created_ms)
135             VALUES (?1, ?2, ?3)",
136            rusqlite::params![partition_name, partition_date, now_ms],
137        )?;
138    }
139    tx.commit()?;
140    // Refresh the raw_all view outside the transaction.
141    refresh_raw_all_view(conn)?;
142    tracing::info!(
143        code = "analytics_partition_rotated",
144        partition = %partition_name,
145        "raw table rotated to partition {partition_name}"
146    );
147    Ok(Some(partition_name))
148}
149
150/// The UTC midnight boundary (unix ms) of the day containing `now_ms`.
151fn day_boundary_ms(now_ms: i64) -> i64 {
152    let day_ms = 86_400_000i64;
153    (now_ms / day_ms) * day_ms
154}
155
156/// Recreate the `raw_all` view as the UNION ALL of the `raw` table
157/// and all `raw_YYYYMMDD` partition tables. Called after each
158/// rotation and after dropping expired partitions.
159pub fn refresh_raw_all_view(conn: &Connection) -> rusqlite::Result<()> {
160    let mut names: Vec<String> = Vec::new();
161    {
162        let mut stmt =
163            conn.prepare("SELECT partition_name FROM raw_partitions ORDER BY partition_name")?;
164        let rows = stmt.query_map([], |r| r.get::<_, String>(0))?;
165        for row in rows {
166            names.push(row?);
167        }
168    }
169    let mut sql = String::from("DROP VIEW IF EXISTS raw_all; CREATE VIEW raw_all AS ");
170    sql.push_str("SELECT * FROM raw");
171    for name in &names {
172        sql.push_str(&format!(" UNION ALL SELECT * FROM {name}"));
173    }
174    conn.execute_batch(&sql)?;
175    Ok(())
176}
177
178/// Drop partition tables older than the raw retention period. Returns
179/// the number of partitions dropped. Each drop is O(1) (SQLite frees
180/// the table's pages to the file's free list; incremental vacuum
181/// reclaims them).
182pub fn drop_expired_partitions(
183    conn: &Connection,
184    raw_keep_ms: i64,
185    now_ms: i64,
186) -> rusqlite::Result<usize> {
187    let cutoff = now_ms.saturating_sub(raw_keep_ms);
188    let cutoff_ymd = ymd_from_ms(cutoff);
189    // Select expired partitions before dropping (can't mutate while
190    // iterating the same connection).
191    let mut expired: Vec<String> = Vec::new();
192    {
193        let mut stmt = conn.prepare(
194            "SELECT partition_name FROM raw_partitions
195             WHERE date_ymd < ?1 ORDER BY partition_name",
196        )?;
197        let rows = stmt.query_map([&cutoff_ymd], |r| r.get::<_, String>(0))?;
198        for row in rows {
199            expired.push(row?);
200        }
201    }
202    if expired.is_empty() {
203        return Ok(0);
204    }
205    let tx = conn.unchecked_transaction()?;
206    let mut dropped = 0usize;
207    for name in &expired {
208        tx.execute_batch(&format!("DROP TABLE IF EXISTS {name}"))?;
209        tx.execute(
210            "DELETE FROM raw_partitions WHERE partition_name = ?1",
211            [name],
212        )?;
213        dropped += 1;
214        tracing::info!(
215            code = "analytics_partition_dropped",
216            partition = %name,
217            "expired raw partition dropped"
218        );
219    }
220    tx.commit()?;
221    refresh_raw_all_view(conn)?;
222    Ok(dropped)
223}
224
225/// List all active partition names (including the current `raw`
226/// table), ordered by name. Useful for introspection and diagnostics.
227pub fn list_partitions(conn: &Connection) -> rusqlite::Result<Vec<String>> {
228    let mut names: Vec<String> = vec!["raw".to_string()];
229    let mut stmt =
230        conn.prepare("SELECT partition_name FROM raw_partitions ORDER BY partition_name")?;
231    let rows = stmt.query_map([], |r| r.get::<_, String>(0))?;
232    for row in rows {
233        names.push(row?);
234    }
235    Ok(names)
236}
237
238#[cfg(test)]
239mod tests {
240    use super::*;
241
242    fn open_test_db() -> Connection {
243        let conn = Connection::open_in_memory().unwrap();
244        conn.execute_batch(SCHEMA_V11).unwrap();
245        conn.execute_batch(
246            "CREATE TABLE raw (
247                id           INTEGER PRIMARY KEY AUTOINCREMENT,
248                ts_ms        INTEGER NOT NULL,
249                listener     TEXT NOT NULL,
250                route        TEXT NOT NULL,
251                consumer     TEXT NOT NULL,
252                upstream     TEXT NOT NULL,
253                method       TEXT NOT NULL,
254                status       INTEGER NOT NULL,
255                status_class TEXT NOT NULL,
256                duration_ms  REAL NOT NULL,
257                attempts     INTEGER NOT NULL,
258                rate_limited INTEGER NOT NULL,
259                broken       INTEGER NOT NULL,
260                shed         INTEGER NOT NULL,
261                dims         TEXT NOT NULL DEFAULT '{}',
262                request_id   TEXT NOT NULL DEFAULT '',
263                correlation_id TEXT NOT NULL DEFAULT '',
264                request_headers_redacted TEXT,
265                auth_identity TEXT
266            );
267            CREATE INDEX idx_raw_ts ON raw(ts_ms);
268            CREATE INDEX idx_raw_correlation ON raw(correlation_id, ts_ms);",
269        )
270        .unwrap();
271        conn
272    }
273
274    #[test]
275    fn ymd_from_ms_basic() {
276        // 2026-01-01 00:00:00 UTC = 1767225600 seconds = 1767225600000 ms
277        assert_eq!(ymd_from_ms(1767225600000), "20260101");
278        // 2026-08-30 09:00:00 UTC = 1725008400 seconds... let's use a known value
279        // 2026-01-02 00:00:00 UTC = 1767312000000 ms
280        assert_eq!(ymd_from_ms(1767312000000), "20260102");
281    }
282
283    #[test]
284    fn day_boundary_aligns_to_midnight() {
285        // 2026-01-01 12:00:00 UTC = 1767225600000 + 43200000 = 1767268800000
286        let noon = 1767268800000i64;
287        let boundary = day_boundary_ms(noon);
288        assert_eq!(boundary, 1767225600000); // midnight
289    }
290
291    #[test]
292    fn no_rotation_when_raw_is_empty() {
293        let conn = open_test_db();
294        let result = maybe_rotate(&conn, 1767225600000).unwrap();
295        assert!(result.is_none());
296    }
297
298    #[test]
299    fn no_rotation_when_all_rows_are_today() {
300        let conn = open_test_db();
301        let now = 1767268800000i64; // 2026-01-01 12:00 UTC
302        conn.execute(
303            "INSERT INTO raw (ts_ms, listener, route, consumer, upstream, method,
304                status, status_class, duration_ms, attempts, rate_limited, broken,
305                shed, dims) VALUES (?1, 'l', 'r', 'c', 'u', 'GET', 200, '2xx',
306                1.0, 1, 0, 0, 0, '{}')",
307            [now],
308        )
309        .unwrap();
310        let result = maybe_rotate(&conn, now).unwrap();
311        assert!(result.is_none());
312    }
313
314    #[test]
315    fn rotation_moves_old_rows_to_partition() {
316        let conn = open_test_db();
317        let day1_morning = 1767225600000i64; // 2026-01-01 00:00 UTC
318        let day1_noon = 1767268800000i64; // 2026-01-01 12:00 UTC
319        let day2_morning = 1767312000000i64; // 2026-01-02 00:00 UTC
320                                             // Insert two rows from 2026-01-01.
321        conn.execute(
322            "INSERT INTO raw (ts_ms, listener, route, consumer, upstream, method,
323                status, status_class, duration_ms, attempts, rate_limited, broken,
324                shed, dims) VALUES (?1, 'l', 'r', 'c', 'u', 'GET', 200, '2xx',
325                1.0, 1, 0, 0, 0, '{}')",
326            [day1_morning],
327        )
328        .unwrap();
329        conn.execute(
330            "INSERT INTO raw (ts_ms, listener, route, consumer, upstream, method,
331                status, status_class, duration_ms, attempts, rate_limited, broken,
332                shed, dims) VALUES (?1, 'l', 'r', 'c', 'u', 'GET', 200, '2xx',
333                1.0, 1, 0, 0, 0, '{}')",
334            [day1_noon],
335        )
336        .unwrap();
337        // Trigger rotation from 2026-01-02. Both rows from 2026-01-01
338        // are before the day boundary (2026-01-02 00:00) and move to
339        // the partition.
340        let result = maybe_rotate(&conn, day2_morning).unwrap();
341        assert_eq!(result.as_deref(), Some("raw_20260101"));
342        // The raw table should be empty (no rows from 2026-01-02).
343        let remaining: i64 = conn
344            .query_row("SELECT COUNT(*) FROM raw", [], |r| r.get(0))
345            .unwrap();
346        assert_eq!(remaining, 0);
347        // The partition should have both rows.
348        let partitioned: i64 = conn
349            .query_row("SELECT COUNT(*) FROM raw_20260101", [], |r| r.get(0))
350            .unwrap();
351        assert_eq!(partitioned, 2);
352        // The raw_all view should see both.
353        let total: i64 = conn
354            .query_row("SELECT COUNT(*) FROM raw_all", [], |r| r.get(0))
355            .unwrap();
356        assert_eq!(total, 2);
357    }
358
359    #[test]
360    fn drop_expired_partitions() {
361        let conn = open_test_db();
362        // Create a partition for 2025-12-31.
363        conn.execute_batch(
364            "CREATE TABLE raw_20251231 (ts_ms INTEGER NOT NULL);
365             INSERT INTO raw_partitions (partition_name, date_ymd, created_ms)
366             VALUES ('raw_20251231', '20251231', 1767139200000);",
367        )
368        .unwrap();
369        refresh_raw_all_view(&conn).unwrap();
370        // Now is 2026-01-02; retention is 1 day (86400000 ms).
371        // Cutoff is 2026-01-01 00:00 UTC. Partition 20251231 is before
372        // 20260101, so it should be dropped.
373        let now = 1767312000000i64;
374        let dropped = super::drop_expired_partitions(&conn, 86_400_000, now).unwrap();
375        assert_eq!(dropped, 1);
376        // The partition table should be gone.
377        let exists: i64 = conn
378            .query_row(
379                "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='raw_20251231'",
380                [],
381                |r| r.get(0),
382            )
383            .unwrap();
384        assert_eq!(exists, 0);
385    }
386
387    #[test]
388    fn list_partitions_includes_raw() {
389        let conn = open_test_db();
390        conn.execute_batch(
391            "CREATE TABLE raw_20260101 (ts_ms INTEGER NOT NULL);
392             INSERT INTO raw_partitions (partition_name, date_ymd, created_ms)
393             VALUES ('raw_20260101', '20260101', 1767225600000);",
394        )
395        .unwrap();
396        let parts = list_partitions(&conn).unwrap();
397        assert_eq!(parts, vec!["raw", "raw_20260101"]);
398    }
399}
dwara_core/analytics/partition.rs

dwara_core/analytics/
partition.rs
