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 /// Mutable scans inspect at most `batch_size` physical positions across
72 /// collections and resume from their previous position on the next tick.
73 pub fn sweep_retention_tick(&self, batch_size: usize) {
74 if batch_size == 0 {
75 return;
76 }
77 let now_ms = std::time::SystemTime::now()
78 .duration_since(std::time::UNIX_EPOCH)
79 .map(|d| d.as_millis() as u64)
80 .unwrap_or(0);
81
82 let store = self.inner.db.store();
83 let mut collections = store.list_collections();
84 collections.sort();
85 if let Some(next) = self.inner.retention_sweeper.read().next_collection() {
86 if let Some(start) = collections.iter().position(|name| name == next) {
87 collections.rotate_left(start);
88 }
89 }
90 let mut remaining = batch_size;
91 let mut append_only_checked = false;
92 for (index, name) in collections.iter().enumerate() {
93 let Some(contract) = self.inner.db.collection_contract(name) else {
94 continue;
95 };
96 let Some(retention_ms) = contract.retention_duration_ms else {
97 continue;
98 };
99 if contract.append_only {
100 if !append_only_checked {
101 append_only_checked = true;
102 if let Err(err) = self.retire_expired_append_only_segments(now_ms, batch_size) {
103 tracing::warn!(error = %err, "append-only retention sweep failed");
104 }
105 }
106 continue;
107 }
108 let Some(ts_column) =
109 crate::runtime::retention_filter::resolve_timestamp_column(&contract)
110 else {
111 continue;
112 };
113 if store.get_collection(name).is_none() {
114 continue;
115 }
116 if remaining == 0 {
117 continue;
118 }
119 let cutoff = super::retention_sweeper::cutoff_ms(now_ms, retention_ms);
120
121 let candidates = self
122 .inner
123 .retention_sweeper
124 .write()
125 .scan_tick(&store, name, remaining);
126 let (candidates, scanned) = match candidates {
127 Ok(result) => result,
128 Err(_) => {
129 self.inner
130 .retention_sweeper
131 .write()
132 .record_tick(name, 0, 1, now_ms);
133 // The failed scan may already have consumed its budget.
134 // Retry other collections on a later tick.
135 remaining = 0;
136 self.inner
137 .retention_sweeper
138 .write()
139 .resume_at(collections[(index + 1) % collections.len()].clone());
140 continue;
141 }
142 };
143 remaining -= scanned;
144 if remaining == 0 && !collections.is_empty() {
145 self.inner
146 .retention_sweeper
147 .write()
148 .resume_at(collections[(index + 1) % collections.len()].clone());
149 }
150 if candidates.is_empty() {
151 self.inner
152 .retention_sweeper
153 .write()
154 .record_tick(name, 0, 1, now_ms);
155 continue;
156 }
157
158 let stmt = format!("DELETE FROM {} WHERE {} < {}", name, ts_column, cutoff);
159 let result = super::retention_sweeper::with_delete_candidates(
160 candidates,
161 name,
162 &ts_column,
163 || self.execute_query(&stmt),
164 );
165 let (deleted, pending) = match result {
166 Ok(r) => (r.affected_rows, u64::from(scanned > 0)),
167 Err(_) => (0, 1),
168 };
169
170 self.inner
171 .retention_sweeper
172 .write()
173 .record_tick(name, deleted, pending, now_ms);
174 }
175 }
176
177 pub fn refresh_due_materialized_views(&self) {
178 let due = {
179 let mut cache = self.inner.materialized_views.write();
180 cache.claim_due_at(std::time::Instant::now())
181 };
182 for name in due {
183 // Round-trip through `execute_query` (rather than the
184 // expression-only `execute_query_expr` entry, which
185 // explicitly rejects DDL/maintenance statements). Failures
186 // are captured inside the RefreshMaterializedView handler
187 // via `record_refresh_failure`; the scheduler ignores the
188 // Result so one bad view doesn't halt the loop.
189 let stmt = format!("REFRESH MATERIALIZED VIEW {}", name);
190 let _ = self.execute_query(&stmt);
191 }
192 }
193}