reddb_server/runtime/impl_materialized_view.rs
1//! Runtime materialized-view refresh and retention sweep.
2//!
3//! Extracted verbatim from `impl_core.rs` (impl_core slice 6/10, issue #1627).
4//! Houses the background-tick surface for scheduled view refresh and retention
5//! sweeps:
6//!
7//! - `materialized_view_metadata` — snapshot of every registered view's
8//! runtime state (feeds `red.materialized_views`).
9//! - `retention_sweeper_snapshot` — snapshot of every sweeper's state
10//! (feeds `red.retention`).
11//! - `sweep_retention_tick` — one tick of the retention sweeper.
12//! - `refresh_due_materialized_views` — claim and refresh views due for
13//! a scheduled refresh.
14use super::*;
15
16impl RedDBRuntime {
17 /// Snapshot of every registered materialized view's runtime
18 /// state — feeds the `red.materialized_views` virtual table.
19 /// Issue #583 slice 10.
20 pub fn materialized_view_metadata(
21 &self,
22 ) -> Vec<crate::storage::cache::result::MaterializedViewMetadata> {
23 // Issue #595 slice 9c — `current_row_count` is now scraped
24 // live from the backing collection rather than read from the
25 // cache slot. Mirrors the slice-10 invariant on
26 // `queue_pending_gauge` in #527: the live store is the source
27 // of truth, the cache slot only carries last-refresh telemetry
28 // (timing, error, refresh cadence).
29 let store = self.inner.db.store();
30 let mut entries = self.inner.materialized_views.read().metadata();
31 for entry in &mut entries {
32 if let Some(manager) = store.get_collection(&entry.name) {
33 entry.current_row_count = manager.count() as u64;
34 }
35 }
36 entries
37 }
38
39 /// Drive scheduled refreshes for materialized views with a
40 /// `REFRESH EVERY <duration>` clause. Called from the background
41 /// scheduler thread (and from unit tests with a fake clock via
42 /// `claim_due_at`). Each invocation atomically claims the set of
43 /// due views (so two concurrent ticks never double-fire the same
44 /// view) and runs each refresh through the standard execution
45 /// path — failures are captured in `last_error` and the prior
46 /// content stays intact. Issue #583 slice 10.
47 /// Snapshot of every tracked retention sweeper state — feeds the
48 /// sweeper observability columns on `red.retention`.
49 pub(crate) fn retention_sweeper_snapshot(
50 &self,
51 ) -> Vec<(String, crate::runtime::retention_sweeper::SweeperState)> {
52 self.inner.retention_sweeper.read().snapshot()
53 }
54
55 /// Drive one tick of the retention sweeper. Mutable collections
56 /// physically delete at most `batch_size` expired rows; append-only
57 /// collections retire whole expired segments through the operational
58 /// manifest. Records the counters that `red.retention` exposes.
59 /// Called from the background sweeper thread; safe to invoke directly
60 /// from tests with a small batch size to drain rows deterministically.
61 /// Issue #584 slice 12.
62 ///
63 /// Deletes are issued as `DELETE FROM <collection> WHERE
64 /// <ts_column> < <cutoff>` through the standard `execute_query`
65 /// chokepoint so WAL participation and snapshot guards apply
66 /// exactly as for a user-issued DELETE — replicas replay the
67 /// sweeper's deletes via the same WAL stream with no special
68 /// handling on the replication side.
69 ///
70 /// Batching is enforced by tightening the cutoff: if more than
71 /// `batch_size` rows are expired, the cutoff is dropped to the
72 /// `batch_size`-th oldest expired timestamp + 1 so the predicate
73 /// matches roughly `batch_size` rows; the remainder is reported
74 /// as `current_rows_pending_sweep_estimate` and drained on the
75 /// next tick.
76 pub fn sweep_retention_tick(&self, batch_size: usize) {
77 if batch_size == 0 {
78 return;
79 }
80 let now_ms = std::time::SystemTime::now()
81 .duration_since(std::time::UNIX_EPOCH)
82 .map(|d| d.as_millis() as u64)
83 .unwrap_or(0);
84
85 let store = self.inner.db.store();
86 let collections = store.list_collections();
87 for name in collections {
88 let Some(contract) = self.inner.db.collection_contract(&name) else {
89 continue;
90 };
91 let Some(retention_ms) = contract.retention_duration_ms else {
92 continue;
93 };
94 if contract.append_only {
95 let _ = self.retire_expired_append_only_segments(now_ms);
96 continue;
97 }
98 let Some(ts_column) =
99 crate::runtime::retention_filter::resolve_timestamp_column(&contract)
100 else {
101 continue;
102 };
103 let Some(manager) = store.get_collection(&name) else {
104 continue;
105 };
106 let cutoff = (now_ms as i64).saturating_sub(retention_ms as i64);
107
108 // Single pass: collect expired timestamps. We keep the
109 // full Vec rather than a bounded heap because the partial
110 // sort below is the simplest correct way to find the
111 // batch-th oldest; for the slice's "1000-row default
112 // batch" target this is bounded enough for production
113 // operation, and the alternative (in-place heap of size
114 // batch+1) is a follow-up optimisation.
115 let mut expired_ts: Vec<i64> = Vec::new();
116 manager.for_each_entity(|entity| {
117 let ts = match ts_column.as_str() {
118 "created_at" => Some(entity.created_at as i64),
119 "updated_at" => Some(entity.updated_at as i64),
120 other => entity
121 .data
122 .as_row()
123 .and_then(|row| row.get_field(other))
124 .and_then(|v| match v {
125 crate::storage::schema::Value::TimestampMs(t) => Some(*t),
126 crate::storage::schema::Value::Timestamp(t) => {
127 Some(t.saturating_mul(1_000))
128 }
129 crate::storage::schema::Value::BigInt(t) => Some(*t),
130 crate::storage::schema::Value::UnsignedInteger(t) => {
131 i64::try_from(*t).ok()
132 }
133 crate::storage::schema::Value::Integer(t) => Some(*t),
134 _ => None,
135 }),
136 };
137 if let Some(t) = ts {
138 if t < cutoff {
139 expired_ts.push(t);
140 }
141 }
142 true
143 });
144
145 let total_expired = expired_ts.len() as u64;
146 if total_expired == 0 {
147 self.inner
148 .retention_sweeper
149 .write()
150 .record_tick(&name, 0, 0, now_ms);
151 continue;
152 }
153
154 let (effective_cutoff, pending) = if (total_expired as usize) <= batch_size {
155 (cutoff, 0u64)
156 } else {
157 // Tighten the cutoff to the (batch_size)-th oldest
158 // expired timestamp + 1 so DELETE matches roughly
159 // `batch_size` rows.
160 expired_ts.sort_unstable();
161 let nth = expired_ts[batch_size - 1];
162 (
163 nth.saturating_add(1),
164 total_expired.saturating_sub(batch_size as u64),
165 )
166 };
167
168 let stmt = format!(
169 "DELETE FROM {} WHERE {} < {}",
170 name, ts_column, effective_cutoff
171 );
172 let deleted = match self.execute_query(&stmt) {
173 Ok(r) => r.affected_rows,
174 Err(_) => 0,
175 };
176
177 self.inner
178 .retention_sweeper
179 .write()
180 .record_tick(&name, deleted, pending, now_ms);
181 }
182 }
183
184 pub fn refresh_due_materialized_views(&self) {
185 let due = {
186 let mut cache = self.inner.materialized_views.write();
187 cache.claim_due_at(std::time::Instant::now())
188 };
189 for name in due {
190 // Round-trip through `execute_query` (rather than the
191 // prepared-statement `execute_query_expr` fast path, which
192 // explicitly rejects DDL/maintenance statements). Failures
193 // are captured inside the RefreshMaterializedView handler
194 // via `record_refresh_failure`; the scheduler ignores the
195 // Result so one bad view doesn't halt the loop.
196 let stmt = format!("REFRESH MATERIALIZED VIEW {}", name);
197 let _ = self.execute_query(&stmt);
198 }
199 }
200}