reddb_server/runtime/
impl_catalog_accessors.rs

1//! Runtime catalog / stats / maintenance accessors.
2//!
3//! Extracted verbatim from `impl_core.rs` (impl_core slice 8/10, issue #1629).
4//! Houses connection `acquire`, `checkpoint`, the remote-write assertion,
5//! `run_maintenance`, `scan_collection`, and the catalog /
6//! attention-summary / `stats` readers.
7use super::*;
8
9fn runtime_pool_lock(runtime: &RedDBRuntime) -> std::sync::MutexGuard<'_, PoolState> {
10    runtime
11        .inner
12        .pool
13        .lock()
14        .unwrap_or_else(|poisoned| poisoned.into_inner())
15}
16
17impl RedDBRuntime {
18    pub fn acquire(&self) -> RedDBResult<RuntimeConnection> {
19        self.acquire_connection(true)
20    }
21
22    // Transport admission already bounds wire sessions. Share connection IDs
23    // and rollback cleanup without imposing the embedded pool's separate cap.
24    pub(crate) fn acquire_wire_connection(&self) -> RedDBResult<RuntimeConnection> {
25        self.acquire_connection(false)
26    }
27
28    fn acquire_connection(&self, uses_pool_slot: bool) -> RedDBResult<RuntimeConnection> {
29        let mut pool = self
30            .inner
31            .pool
32            .lock()
33            .map_err(|e| RedDBError::Internal(format!("connection pool lock poisoned: {e}")))?;
34        if uses_pool_slot && pool.pooled_active >= self.inner.pool_config.max_connections {
35            return Err(RedDBError::Internal(
36                "connection pool exhausted".to_string(),
37            ));
38        }
39
40        let id = if let Some(id) = pool.idle.pop() {
41            id
42        } else {
43            let id = pool.next_id;
44            pool.next_id += 1;
45            id
46        };
47        pool.active += 1;
48        if uses_pool_slot {
49            pool.pooled_active += 1;
50        }
51        pool.total_checkouts += 1;
52        drop(pool);
53
54        // Issue #1245 — record the connection acquisition after releasing
55        // the pool lock so the lock hold time is unchanged.
56        self.inner.node_load_telemetry.record_connect();
57
58        Ok(RuntimeConnection {
59            background_tasks: self.background_tasks.clone(),
60            id,
61            uses_pool_slot,
62            inner: Arc::clone(&self.inner),
63        })
64    }
65
66    pub fn checkpoint(&self) -> RedDBResult<()> {
67        // Local fsync always allowed — losing the lease shouldn't
68        // prevent us from durably persisting what's already in memory.
69        // The remote upload is the side-effect that risks clobbering a
70        // peer's state, so it's behind the lease gate.
71        self.checkpoint_local()?;
72        if let Err(err) = self.assert_remote_write_allowed("checkpoint") {
73            tracing::warn!(
74                target: "reddb::serverless::lease",
75                error = %err,
76                "checkpoint: skipping remote upload — lease not held"
77            );
78            return Ok(());
79        }
80        self.inner
81            .db
82            .upload_to_remote_backend()
83            .map_err(|err| RedDBError::Engine(err.to_string()))
84    }
85
86    pub(super) fn checkpoint_local(&self) -> RedDBResult<()> {
87        let started = std::time::Instant::now();
88        self.seal_hypertable_chunks_for_checkpoint(
89            self.inner.checkpoint_columnar_emission_budget_chunks,
90        )?;
91        let store = self.inner.db.store();
92        let publication = store.vault_publication_lock.lock();
93        // Collections created since startup need the compact visibility
94        // inventory before their first resident-to-disk publication too.
95        self.prepare_paged_counts()?;
96        self.persist_probabilistic_snapshots()?;
97        self.inner
98            .db
99            .flush_local_after_runtime_snapshots()
100            .map_err(|err| {
101                // Issue #205 — local flush failure is a CheckpointFailed
102                // operator-grade event. The local-flush path also covers
103                // the WAL fsync we depend on, so a failure here doubles as
104                // the WalFsyncFailed signal for the runtime entry point.
105                let msg = err.to_string();
106                crate::telemetry::operator_event::OperatorEvent::CheckpointFailed {
107                    lsn: 0,
108                    error: msg.clone(),
109                }
110                .emit_global();
111                crate::telemetry::operator_event::OperatorEvent::WalFsyncFailed {
112                    path: "<flush_local_only>".to_string(),
113                    error: msg.clone(),
114                }
115                .emit_global();
116                RedDBError::Engine(msg)
117            })?;
118        // The checkpoint now contains every mutation fenced above. Rebase the
119        // ordered live overlay while publication is still exclusive so a
120        // concurrent writer cannot be cleared after its WAL change.
121        self.index_store_ref().clear_paged_btree_overlays();
122        self.inner.probabilistic.replayed_names.lock().clear();
123        drop(publication);
124        let checkpoint_lsn = self.cdc_current_lsn();
125        self.inner
126            .checkpoint_projection_stats
127            .record_checkpoint(checkpoint_lsn, started.elapsed().as_millis() as u64);
128        Ok(())
129    }
130
131    /// Guard remote-mutating operations on the writer lease.
132    /// Returns `Ok(())` when no remote backend is configured (the
133    /// lease is irrelevant) or the lease state is `NotRequired` /
134    /// `Held`. Returns `RedDBError::ReadOnly` when the lease is
135    /// `NotHeld`, with an audit-friendly action label so the caller
136    /// can record the rejection.
137    pub(crate) fn assert_remote_write_allowed(&self, action: &str) -> RedDBResult<()> {
138        if self.inner.db.remote_backend.is_none() {
139            return Ok(());
140        }
141        match self.inner.write_gate.lease_state() {
142            crate::runtime::write_gate::LeaseGateState::NotHeld => {
143                self.inner.audit_log.record(
144                    action,
145                    "system",
146                    "remote_backend",
147                    "err: writer lease not held",
148                    crate::json::Value::Null,
149                );
150                Err(RedDBError::ReadOnly(format!(
151                    "writer lease not held — {action} blocked (serverless fence)"
152                )))
153            }
154            _ => Ok(()),
155        }
156    }
157
158    pub fn run_maintenance(&self) -> RedDBResult<()> {
159        self.inner
160            .db
161            .run_maintenance()
162            .map_err(|err| RedDBError::Internal(err.to_string()))
163    }
164
165    pub fn scan_collection(
166        &self,
167        collection: &str,
168        cursor: Option<ScanCursor>,
169        limit: usize,
170    ) -> RedDBResult<ScanPage> {
171        use reddb_rql::ast::PolicyAction;
172
173        self.check_collection_privilege(PolicyAction::Select, collection)?;
174        let store = self.inner.db.store();
175        let manager = store
176            .get_collection(collection)
177            .ok_or_else(|| RedDBError::NotFound(collection.to_string()))?;
178
179        let offset = cursor.map(|cursor| cursor.offset).unwrap_or(0);
180        let limit = limit.max(1);
181        let read_bytes = usize::try_from(self.memory_budget().resolved_bytes / 16)
182            .unwrap_or(usize::MAX)
183            .clamp(1, 16 * 1024 * 1024);
184        let (total, items) = if crate::runtime::execution_context::current_auth_identity().is_some()
185            && crate::runtime::rls_injection::rls_is_enabled(self, collection)
186        {
187            manager.try_page_in_id_order_where(offset, limit, read_bytes, |entity| {
188                crate::runtime::rls_injection::entity_passes_rls_action(
189                    self,
190                    collection,
191                    PolicyAction::Select,
192                    entity,
193                )
194            })?
195        } else {
196            manager.try_page_in_id_order(offset, limit, read_bytes)?
197        };
198        let end = total.min(offset.saturating_add(limit));
199        let next = (end < total).then_some(ScanCursor { offset: end });
200
201        Ok(ScanPage {
202            collection: collection.to_string(),
203            items,
204            next,
205            total,
206        })
207    }
208
209    pub fn get_collection_entity(
210        &self,
211        collection: &str,
212        id: crate::storage::EntityId,
213    ) -> RedDBResult<Option<crate::storage::UnifiedEntity>> {
214        use reddb_rql::ast::PolicyAction;
215
216        self.check_collection_privilege(PolicyAction::Select, collection)?;
217        let entity = self.inner.db.store().get(collection, id);
218        Ok(entity.filter(|entity| {
219            crate::runtime::rls_injection::entity_passes_rls_action(
220                self,
221                collection,
222                PolicyAction::Select,
223                entity,
224            )
225        }))
226    }
227
228    pub fn catalog(&self) -> CatalogModelSnapshot {
229        self.inner.db.catalog_model_snapshot()
230    }
231
232    /// Fallible catalog for disk-resident databases. The decode workspace shares
233    /// query admission with concurrent reads and is released on every error.
234    pub fn try_catalog(&self) -> RedDBResult<CatalogModelSnapshot> {
235        if self
236            .inner
237            .db
238            .store()
239            .config()
240            .resident_memory_bytes_max
241            .is_none()
242        {
243            return Ok(self.catalog());
244        }
245        let read_bytes = usize::try_from(self.memory_budget().resolved_bytes / 16)
246            .unwrap_or(usize::MAX)
247            .clamp(1, 16 * 1024 * 1024);
248        let _scratch = self
249            .try_reserve_query_buffer(read_bytes * 2)
250            .ok_or_else(|| {
251                self.memory_accounting().record_admission_denied();
252                RedDBError::MaterializationLimitExceeded {
253                    executor: "catalog",
254                    limit: usize::try_from(self.memory_budget().resolved_bytes)
255                        .unwrap_or(usize::MAX),
256                    current: usize::try_from(self.memory_accounting().total_used_bytes())
257                        .unwrap_or(usize::MAX)
258                        .saturating_add(1),
259                }
260            })?;
261        Ok(self.inner.db.try_catalog_model_snapshot(read_bytes)?)
262    }
263
264    /// Fallible catalog inventory scoped to known collection names. This keeps
265    /// filtered `red.collections` queries from decoding every unrelated
266    /// collection before their predicate is applied.
267    pub(crate) fn try_catalog_for_collections(
268        &self,
269        collection_names: &std::collections::HashSet<String>,
270        observe_legacy_payloads: bool,
271    ) -> RedDBResult<CatalogModelSnapshot> {
272        if self
273            .inner
274            .db
275            .store()
276            .config()
277            .resident_memory_bytes_max
278            .is_none()
279        {
280            return Ok(self.inner.db.try_catalog_model_snapshot_for_collections(
281                collection_names,
282                observe_legacy_payloads,
283                usize::MAX,
284            )?);
285        }
286        let read_bytes = usize::try_from(self.memory_budget().resolved_bytes / 16)
287            .unwrap_or(usize::MAX)
288            .clamp(1, 16 * 1024 * 1024);
289        let _scratch = self
290            .try_reserve_query_buffer(read_bytes * 2)
291            .ok_or_else(|| {
292                self.memory_accounting().record_admission_denied();
293                RedDBError::MaterializationLimitExceeded {
294                    executor: "catalog",
295                    limit: usize::try_from(self.memory_budget().resolved_bytes)
296                        .unwrap_or(usize::MAX),
297                    current: usize::try_from(self.memory_accounting().total_used_bytes())
298                        .unwrap_or(usize::MAX)
299                        .saturating_add(1),
300                }
301            })?;
302        Ok(self.inner.db.try_catalog_model_snapshot_for_collections(
303            collection_names,
304            observe_legacy_payloads,
305            read_bytes,
306        )?)
307    }
308
309    pub fn catalog_consistency_report(&self) -> crate::catalog::CatalogConsistencyReport {
310        self.inner.db.catalog_consistency_report()
311    }
312
313    pub fn catalog_attention_summary(&self) -> CatalogAttentionSummary {
314        crate::catalog::attention_summary(&self.catalog())
315    }
316
317    pub fn collection_attention(&self) -> Vec<CollectionDescriptor> {
318        crate::catalog::collection_attention(&self.catalog())
319    }
320
321    pub fn index_attention(&self) -> Vec<CatalogIndexStatus> {
322        crate::catalog::index_attention(&self.catalog())
323    }
324
325    pub fn graph_projection_attention(&self) -> Vec<CatalogGraphProjectionStatus> {
326        crate::catalog::graph_projection_attention(&self.catalog())
327    }
328
329    pub fn analytics_job_attention(&self) -> Vec<CatalogAnalyticsJobStatus> {
330        crate::catalog::analytics_job_attention(&self.catalog())
331    }
332
333    pub fn stats(&self) -> RuntimeStats {
334        let pool = runtime_pool_lock(self);
335        RuntimeStats {
336            active_connections: pool.active,
337            idle_connections: pool.idle.len(),
338            total_checkouts: pool.total_checkouts,
339            paged_mode: self.inner.db.is_paged(),
340            started_at_unix_ms: self.inner.started_at_unix_ms,
341            store: self.inner.db.stats(),
342            system: SystemInfo::collect(),
343            result_blob_cache: self.inner.result_blob_cache.stats(),
344            kv: self.inner.kv_stats.snapshot(),
345            metrics_ingest: self.inner.metrics_ingest_stats.snapshot(),
346        }
347    }
348}
349
350#[cfg(test)]
351mod security_boundary_tests {
352    use super::*;
353    use crate::application::entity::{DeleteEntityInput, PatchEntityInput};
354    use crate::application::ports::RuntimeEntityPort;
355    use crate::auth::Role;
356    use crate::runtime::execution_context::{with_snapshot_bundle, SnapshotBundle};
357    use crate::storage::{EntityData, UnifiedEntity};
358    use reddb_types::Value;
359
360    fn text_field<'a>(entity: &'a UnifiedEntity, name: &str) -> Option<&'a str> {
361        let EntityData::Row(row) = &entity.data else {
362            return None;
363        };
364        let value = row
365            .named
366            .as_ref()
367            .and_then(|named| named.get(name))
368            .or_else(|| {
369                row.schema.as_ref().and_then(|schema| {
370                    schema
371                        .iter()
372                        .position(|column| column == name)
373                        .and_then(|index| row.columns.get(index))
374                })
375            })?;
376        match value {
377            Value::Text(value) => Some(value.as_ref()),
378            _ => None,
379        }
380    }
381
382    #[test]
383    fn typed_entity_boundary_filters_before_pagination_and_hides_foreign_ids() {
384        let runtime = RedDBRuntime::in_memory().expect("runtime");
385        runtime
386            .execute_query(
387                "CREATE TABLE docs (id INT, tenant_id TEXT, body TEXT) TENANT BY (tenant_id)",
388            )
389            .expect("tenant table");
390        runtime
391            .execute_query(
392                "INSERT INTO docs (id, tenant_id, body) VALUES \
393                 (1, 'acme', 'first'), (2, 'globex', 'hidden'), (3, 'acme', 'second')",
394            )
395            .expect("seed platform rows");
396
397        let unfiltered = runtime
398            .scan_collection("docs", None, 10)
399            .expect("platform scan");
400        let foreign_id = unfiltered
401            .items
402            .iter()
403            .find(|entity| text_field(entity, "tenant_id") == Some("globex"))
404            .map(|entity| entity.id)
405            .expect("foreign row");
406
407        let bundle = SnapshotBundle {
408            snapshot: None,
409            auth: Some(("alice".to_string(), Role::Write)),
410            auth_principal: Some(crate::auth::UserId::scoped("acme", "alice")),
411            tenant: Some("acme".to_string()),
412        };
413        with_snapshot_bundle(&bundle, || {
414            let first = runtime
415                .scan_collection("docs", None, 1)
416                .expect("first visible page");
417            assert_eq!(first.total, 2);
418            assert_eq!(first.items.len(), 1);
419            assert_eq!(text_field(&first.items[0], "tenant_id"), Some("acme"));
420            assert_eq!(first.next, Some(ScanCursor { offset: 1 }));
421
422            let second = runtime
423                .scan_collection("docs", first.next, 1)
424                .expect("second visible page");
425            assert_eq!(second.total, 2);
426            assert_eq!(second.items.len(), 1);
427            assert_eq!(text_field(&second.items[0], "tenant_id"), Some("acme"));
428            assert_eq!(second.next, None);
429
430            assert!(runtime
431                .get_collection_entity("docs", foreign_id)
432                .expect("typed get")
433                .is_none());
434            assert!(matches!(
435                runtime.patch_entity(PatchEntityInput {
436                    collection: "docs".to_string(),
437                    id: foreign_id,
438                    payload: crate::json::Value::Object(Default::default()),
439                    operations: Vec::new(),
440                }),
441                Err(crate::RedDBError::NotFound(_))
442            ));
443            let deleted = runtime
444                .delete_entity(DeleteEntityInput {
445                    collection: "docs".to_string(),
446                    id: foreign_id,
447                })
448                .expect("typed delete");
449            assert!(!deleted.deleted);
450        });
451
452        assert!(runtime.db().store().get("docs", foreign_id).is_some());
453    }
454
455    #[test]
456    fn typed_entity_boundary_uses_model_specific_rls_policies() {
457        let runtime = RedDBRuntime::in_memory().expect("runtime");
458        for query in [
459            "CREATE GRAPH links",
460            "INSERT INTO links NODE (label,node_type,visible) VALUES ('shown','Item',true)",
461            "INSERT INTO links NODE (label,node_type,visible) VALUES ('hidden','Item',false)",
462            "CREATE POLICY visible_nodes ON NODES OF links FOR SELECT USING (properties.visible=true)",
463            "ALTER TABLE links ENABLE ROW LEVEL SECURITY",
464            "CREATE DOCUMENT papers",
465            "INSERT INTO papers DOCUMENT VALUES ({name:'shown',visible:true})",
466            "INSERT INTO papers DOCUMENT VALUES ({name:'hidden',visible:false})",
467            "CREATE POLICY visible_documents ON DOCUMENTS OF papers FOR SELECT USING (visible=true)",
468            "ALTER TABLE papers ENABLE ROW LEVEL SECURITY",
469        ] {
470            runtime.execute_query(query).expect(query);
471        }
472
473        let bundle = SnapshotBundle {
474            snapshot: None,
475            auth: Some(("alice".to_string(), Role::Read)),
476            auth_principal: Some(crate::auth::UserId::platform("alice")),
477            tenant: None,
478        };
479        with_snapshot_bundle(&bundle, || {
480            let nodes = runtime
481                .scan_collection("links", None, 10)
482                .expect("typed graph scan");
483            assert_eq!(nodes.total, 1);
484            assert_eq!(nodes.items.len(), 1);
485
486            let documents = runtime
487                .scan_collection("papers", None, 10)
488                .expect("typed document scan");
489            assert_eq!(documents.total, 1);
490            assert_eq!(documents.items.len(), 1);
491        });
492    }
493}
reddb_server/runtime/impl_catalog_accessors.rs

reddb_server/runtime/
impl_catalog_accessors.rs
