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 at most `batch_size` whole expired segments across
58 /// collections through the operational manifest. Records the counters that
59 /// `red.retention` exposes.
60 /// Called from the background sweeper thread; safe to invoke directly
61 /// from tests with a small batch size to drain rows deterministically.
62 /// Issue #584 slice 12.
63 ///
64 /// Deletes are issued as `DELETE FROM <collection> WHERE
65 /// <ts_column> < <cutoff>` through the standard `execute_query`
66 /// chokepoint so WAL participation and snapshot guards apply
67 /// exactly as for a user-issued DELETE — replicas replay the
68 /// sweeper's deletes via the same WAL stream with no special
69 /// handling on the replication side.
70 ///
71 /// Batching is enforced by tightening the cutoff: if more than
72 /// `batch_size` rows are expired, the cutoff is dropped to the
73 /// `batch_size`-th oldest expired timestamp + 1 so the predicate
74 /// matches roughly `batch_size` rows; the remainder is reported
75 /// as `current_rows_pending_sweep_estimate` and drained on the
76 /// next tick.
77 pub fn sweep_retention_tick(&self, batch_size: usize) {
78 if batch_size == 0 {
79 return;
80 }
81 let now_ms = std::time::SystemTime::now()
82 .duration_since(std::time::UNIX_EPOCH)
83 .map(|d| d.as_millis() as u64)
84 .unwrap_or(0);
85
86 let store = self.inner.db.store();
87 let collections = store.list_collections();
88 let mut append_only_checked = false;
89 for name in collections {
90 let Some(contract) = self.inner.db.collection_contract(&name) else {
91 continue;
92 };
93 let Some(retention_ms) = contract.retention_duration_ms else {
94 continue;
95 };
96 if contract.append_only {
97 if !append_only_checked {
98 append_only_checked = true;
99 let _ = self.retire_expired_append_only_segments(now_ms, batch_size);
100 }
101 continue;
102 }
103 let Some(ts_column) =
104 crate::runtime::retention_filter::resolve_timestamp_column(&contract)
105 else {
106 continue;
107 };
108 let Some(manager) = store.get_collection(&name) else {
109 continue;
110 };
111 let cutoff = (now_ms as i64).saturating_sub(retention_ms as i64);
112
113 // Single pass: collect expired timestamps. We keep the
114 // full Vec rather than a bounded heap because the partial
115 // sort below is the simplest correct way to find the
116 // batch-th oldest; for the slice's "1000-row default
117 // batch" target this is bounded enough for production
118 // operation, and the alternative (in-place heap of size
119 // batch+1) is a follow-up optimisation.
120 let mut expired_ts: Vec<i64> = Vec::new();
121 manager.for_each_entity(|entity| {
122 let ts = match ts_column.as_str() {
123 "created_at" => Some(entity.created_at as i64),
124 "updated_at" => Some(entity.updated_at as i64),
125 other => entity
126 .data
127 .as_row()
128 .and_then(|row| row.get_field(other))
129 .and_then(|v| match v {
130 reddb_types::Value::TimestampMs(t) => Some(*t),
131 reddb_types::Value::Timestamp(t) => Some(t.saturating_mul(1_000)),
132 reddb_types::Value::BigInt(t) => Some(*t),
133 reddb_types::Value::UnsignedInteger(t) => i64::try_from(*t).ok(),
134 reddb_types::Value::Integer(t) => Some(*t),
135 _ => None,
136 }),
137 };
138 if let Some(t) = ts {
139 if t < cutoff {
140 expired_ts.push(t);
141 }
142 }
143 true
144 });
145
146 let total_expired = expired_ts.len() as u64;
147 if total_expired == 0 {
148 self.inner
149 .retention_sweeper
150 .write()
151 .record_tick(&name, 0, 0, now_ms);
152 continue;
153 }
154
155 let (effective_cutoff, pending) = if (total_expired as usize) <= batch_size {
156 (cutoff, 0u64)
157 } else {
158 // Tighten the cutoff to the (batch_size)-th oldest
159 // expired timestamp + 1 so DELETE matches roughly
160 // `batch_size` rows.
161 expired_ts.sort_unstable();
162 let nth = expired_ts[batch_size - 1];
163 (
164 nth.saturating_add(1),
165 total_expired.saturating_sub(batch_size as u64),
166 )
167 };
168
169 let stmt = format!(
170 "DELETE FROM {} WHERE {} < {}",
171 name, ts_column, effective_cutoff
172 );
173 let deleted = match self.execute_query(&stmt) {
174 Ok(r) => r.affected_rows,
175 Err(_) => 0,
176 };
177
178 self.inner
179 .retention_sweeper
180 .write()
181 .record_tick(&name, deleted, pending, now_ms);
182 }
183 }
184
185 pub fn refresh_due_materialized_views(&self) {
186 let due = {
187 let mut cache = self.inner.materialized_views.write();
188 cache.claim_due_at(std::time::Instant::now())
189 };
190 for name in due {
191 // Round-trip through `execute_query` (rather than the
192 // expression-only `execute_query_expr` entry, which
193 // explicitly rejects DDL/maintenance statements). Failures
194 // are captured inside the RefreshMaterializedView handler
195 // via `record_refresh_failure`; the scheduler ignores the
196 // Result so one bad view doesn't halt the loop.
197 let stmt = format!("REFRESH MATERIALIZED VIEW {}", name);
198 let _ = self.execute_query(&stmt);
199 }
200 }
201}