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        self.inner.probabilistic.replayed_names.lock().clear();
119        drop(publication);
120        let checkpoint_lsn = self.cdc_current_lsn();
121        self.inner
122            .checkpoint_projection_stats
123            .record_checkpoint(checkpoint_lsn, started.elapsed().as_millis() as u64);
124        Ok(())
125    }
126
127    /// Guard remote-mutating operations on the writer lease.
128    /// Returns `Ok(())` when no remote backend is configured (the
129    /// lease is irrelevant) or the lease state is `NotRequired` /
130    /// `Held`. Returns `RedDBError::ReadOnly` when the lease is
131    /// `NotHeld`, with an audit-friendly action label so the caller
132    /// can record the rejection.
133    pub(crate) fn assert_remote_write_allowed(&self, action: &str) -> RedDBResult<()> {
134        if self.inner.db.remote_backend.is_none() {
135            return Ok(());
136        }
137        match self.inner.write_gate.lease_state() {
138            crate::runtime::write_gate::LeaseGateState::NotHeld => {
139                self.inner.audit_log.record(
140                    action,
141                    "system",
142                    "remote_backend",
143                    "err: writer lease not held",
144                    crate::json::Value::Null,
145                );
146                Err(RedDBError::ReadOnly(format!(
147                    "writer lease not held — {action} blocked (serverless fence)"
148                )))
149            }
150            _ => Ok(()),
151        }
152    }
153
154    pub fn run_maintenance(&self) -> RedDBResult<()> {
155        self.inner
156            .db
157            .run_maintenance()
158            .map_err(|err| RedDBError::Internal(err.to_string()))
159    }
160
161    pub fn scan_collection(
162        &self,
163        collection: &str,
164        cursor: Option<ScanCursor>,
165        limit: usize,
166    ) -> RedDBResult<ScanPage> {
167        use reddb_rql::ast::PolicyAction;
168
169        self.check_collection_privilege(PolicyAction::Select, collection)?;
170        let store = self.inner.db.store();
171        let manager = store
172            .get_collection(collection)
173            .ok_or_else(|| RedDBError::NotFound(collection.to_string()))?;
174
175        let offset = cursor.map(|cursor| cursor.offset).unwrap_or(0);
176        let limit = limit.max(1);
177        let read_bytes = usize::try_from(self.memory_budget().resolved_bytes / 16)
178            .unwrap_or(usize::MAX)
179            .clamp(1, 16 * 1024 * 1024);
180        let (total, items) = if crate::runtime::execution_context::current_auth_identity().is_some()
181            && crate::runtime::rls_injection::rls_is_enabled(self, collection)
182        {
183            manager.try_page_in_id_order_where(offset, limit, read_bytes, |entity| {
184                crate::runtime::rls_injection::entity_passes_rls_action(
185                    self,
186                    collection,
187                    PolicyAction::Select,
188                    entity,
189                )
190            })?
191        } else {
192            manager.try_page_in_id_order(offset, limit, read_bytes)?
193        };
194        let end = total.min(offset.saturating_add(limit));
195        let next = (end < total).then_some(ScanCursor { offset: end });
196
197        Ok(ScanPage {
198            collection: collection.to_string(),
199            items,
200            next,
201            total,
202        })
203    }
204
205    pub fn get_collection_entity(
206        &self,
207        collection: &str,
208        id: crate::storage::EntityId,
209    ) -> RedDBResult<Option<crate::storage::UnifiedEntity>> {
210        use reddb_rql::ast::PolicyAction;
211
212        self.check_collection_privilege(PolicyAction::Select, collection)?;
213        let entity = self.inner.db.store().get(collection, id);
214        Ok(entity.filter(|entity| {
215            crate::runtime::rls_injection::entity_passes_rls_action(
216                self,
217                collection,
218                PolicyAction::Select,
219                entity,
220            )
221        }))
222    }
223
224    pub fn catalog(&self) -> CatalogModelSnapshot {
225        self.inner.db.catalog_model_snapshot()
226    }
227
228    /// Fallible catalog for disk-resident databases. The decode workspace shares
229    /// query admission with concurrent reads and is released on every error.
230    pub fn try_catalog(&self) -> RedDBResult<CatalogModelSnapshot> {
231        if self
232            .inner
233            .db
234            .store()
235            .config()
236            .resident_memory_bytes_max
237            .is_none()
238        {
239            return Ok(self.catalog());
240        }
241        let read_bytes = usize::try_from(self.memory_budget().resolved_bytes / 16)
242            .unwrap_or(usize::MAX)
243            .clamp(1, 16 * 1024 * 1024);
244        let _scratch = self
245            .try_reserve_query_buffer(read_bytes * 2)
246            .ok_or_else(|| {
247                self.memory_accounting().record_admission_denied();
248                RedDBError::MaterializationLimitExceeded {
249                    executor: "catalog",
250                    limit: usize::try_from(self.memory_budget().resolved_bytes)
251                        .unwrap_or(usize::MAX),
252                    current: usize::try_from(self.memory_accounting().total_used_bytes())
253                        .unwrap_or(usize::MAX)
254                        .saturating_add(1),
255                }
256            })?;
257        Ok(self.inner.db.try_catalog_model_snapshot(read_bytes)?)
258    }
259
260    /// Fallible catalog inventory scoped to known collection names. This keeps
261    /// filtered `red.collections` queries from decoding every unrelated
262    /// collection before their predicate is applied.
263    pub(crate) fn try_catalog_for_collections(
264        &self,
265        collection_names: &std::collections::HashSet<String>,
266        observe_legacy_payloads: bool,
267    ) -> RedDBResult<CatalogModelSnapshot> {
268        if self
269            .inner
270            .db
271            .store()
272            .config()
273            .resident_memory_bytes_max
274            .is_none()
275        {
276            return Ok(self.inner.db.try_catalog_model_snapshot_for_collections(
277                collection_names,
278                observe_legacy_payloads,
279                usize::MAX,
280            )?);
281        }
282        let read_bytes = usize::try_from(self.memory_budget().resolved_bytes / 16)
283            .unwrap_or(usize::MAX)
284            .clamp(1, 16 * 1024 * 1024);
285        let _scratch = self
286            .try_reserve_query_buffer(read_bytes * 2)
287            .ok_or_else(|| {
288                self.memory_accounting().record_admission_denied();
289                RedDBError::MaterializationLimitExceeded {
290                    executor: "catalog",
291                    limit: usize::try_from(self.memory_budget().resolved_bytes)
292                        .unwrap_or(usize::MAX),
293                    current: usize::try_from(self.memory_accounting().total_used_bytes())
294                        .unwrap_or(usize::MAX)
295                        .saturating_add(1),
296                }
297            })?;
298        Ok(self.inner.db.try_catalog_model_snapshot_for_collections(
299            collection_names,
300            observe_legacy_payloads,
301            read_bytes,
302        )?)
303    }
304
305    pub fn catalog_consistency_report(&self) -> crate::catalog::CatalogConsistencyReport {
306        self.inner.db.catalog_consistency_report()
307    }
308
309    pub fn catalog_attention_summary(&self) -> CatalogAttentionSummary {
310        crate::catalog::attention_summary(&self.catalog())
311    }
312
313    pub fn collection_attention(&self) -> Vec<CollectionDescriptor> {
314        crate::catalog::collection_attention(&self.catalog())
315    }
316
317    pub fn index_attention(&self) -> Vec<CatalogIndexStatus> {
318        crate::catalog::index_attention(&self.catalog())
319    }
320
321    pub fn graph_projection_attention(&self) -> Vec<CatalogGraphProjectionStatus> {
322        crate::catalog::graph_projection_attention(&self.catalog())
323    }
324
325    pub fn analytics_job_attention(&self) -> Vec<CatalogAnalyticsJobStatus> {
326        crate::catalog::analytics_job_attention(&self.catalog())
327    }
328
329    pub fn stats(&self) -> RuntimeStats {
330        let pool = runtime_pool_lock(self);
331        RuntimeStats {
332            active_connections: pool.active,
333            idle_connections: pool.idle.len(),
334            total_checkouts: pool.total_checkouts,
335            paged_mode: self.inner.db.is_paged(),
336            started_at_unix_ms: self.inner.started_at_unix_ms,
337            store: self.inner.db.stats(),
338            system: SystemInfo::collect(),
339            result_blob_cache: self.inner.result_blob_cache.stats(),
340            kv: self.inner.kv_stats.snapshot(),
341            metrics_ingest: self.inner.metrics_ingest_stats.snapshot(),
342        }
343    }
344}
345
346#[cfg(test)]
347mod security_boundary_tests {
348    use super::*;
349    use crate::application::entity::{DeleteEntityInput, PatchEntityInput};
350    use crate::application::ports::RuntimeEntityPort;
351    use crate::auth::Role;
352    use crate::runtime::execution_context::{with_snapshot_bundle, SnapshotBundle};
353    use crate::storage::{EntityData, UnifiedEntity};
354    use reddb_types::Value;
355
356    fn text_field<'a>(entity: &'a UnifiedEntity, name: &str) -> Option<&'a str> {
357        let EntityData::Row(row) = &entity.data else {
358            return None;
359        };
360        let value = row
361            .named
362            .as_ref()
363            .and_then(|named| named.get(name))
364            .or_else(|| {
365                row.schema.as_ref().and_then(|schema| {
366                    schema
367                        .iter()
368                        .position(|column| column == name)
369                        .and_then(|index| row.columns.get(index))
370                })
371            })?;
372        match value {
373            Value::Text(value) => Some(value.as_ref()),
374            _ => None,
375        }
376    }
377
378    #[test]
379    fn typed_entity_boundary_filters_before_pagination_and_hides_foreign_ids() {
380        let runtime = RedDBRuntime::in_memory().expect("runtime");
381        runtime
382            .execute_query(
383                "CREATE TABLE docs (id INT, tenant_id TEXT, body TEXT) TENANT BY (tenant_id)",
384            )
385            .expect("tenant table");
386        runtime
387            .execute_query(
388                "INSERT INTO docs (id, tenant_id, body) VALUES \
389                 (1, 'acme', 'first'), (2, 'globex', 'hidden'), (3, 'acme', 'second')",
390            )
391            .expect("seed platform rows");
392
393        let unfiltered = runtime
394            .scan_collection("docs", None, 10)
395            .expect("platform scan");
396        let foreign_id = unfiltered
397            .items
398            .iter()
399            .find(|entity| text_field(entity, "tenant_id") == Some("globex"))
400            .map(|entity| entity.id)
401            .expect("foreign row");
402
403        let bundle = SnapshotBundle {
404            snapshot: None,
405            auth: Some(("alice".to_string(), Role::Write)),
406            auth_principal: Some(crate::auth::UserId::scoped("acme", "alice")),
407            tenant: Some("acme".to_string()),
408        };
409        with_snapshot_bundle(&bundle, || {
410            let first = runtime
411                .scan_collection("docs", None, 1)
412                .expect("first visible page");
413            assert_eq!(first.total, 2);
414            assert_eq!(first.items.len(), 1);
415            assert_eq!(text_field(&first.items[0], "tenant_id"), Some("acme"));
416            assert_eq!(first.next, Some(ScanCursor { offset: 1 }));
417
418            let second = runtime
419                .scan_collection("docs", first.next, 1)
420                .expect("second visible page");
421            assert_eq!(second.total, 2);
422            assert_eq!(second.items.len(), 1);
423            assert_eq!(text_field(&second.items[0], "tenant_id"), Some("acme"));
424            assert_eq!(second.next, None);
425
426            assert!(runtime
427                .get_collection_entity("docs", foreign_id)
428                .expect("typed get")
429                .is_none());
430            assert!(matches!(
431                runtime.patch_entity(PatchEntityInput {
432                    collection: "docs".to_string(),
433                    id: foreign_id,
434                    payload: crate::json::Value::Object(Default::default()),
435                    operations: Vec::new(),
436                }),
437                Err(crate::RedDBError::NotFound(_))
438            ));
439            let deleted = runtime
440                .delete_entity(DeleteEntityInput {
441                    collection: "docs".to_string(),
442                    id: foreign_id,
443                })
444                .expect("typed delete");
445            assert!(!deleted.deleted);
446        });
447
448        assert!(runtime.db().store().get("docs", foreign_id).is_some());
449    }
450
451    #[test]
452    fn typed_entity_boundary_uses_model_specific_rls_policies() {
453        let runtime = RedDBRuntime::in_memory().expect("runtime");
454        for query in [
455            "CREATE GRAPH links",
456            "INSERT INTO links NODE (label,node_type,visible) VALUES ('shown','Item',true)",
457            "INSERT INTO links NODE (label,node_type,visible) VALUES ('hidden','Item',false)",
458            "CREATE POLICY visible_nodes ON NODES OF links FOR SELECT USING (properties.visible=true)",
459            "ALTER TABLE links ENABLE ROW LEVEL SECURITY",
460            "CREATE DOCUMENT papers",
461            "INSERT INTO papers DOCUMENT VALUES ({name:'shown',visible:true})",
462            "INSERT INTO papers DOCUMENT VALUES ({name:'hidden',visible:false})",
463            "CREATE POLICY visible_documents ON DOCUMENTS OF papers FOR SELECT USING (visible=true)",
464            "ALTER TABLE papers ENABLE ROW LEVEL SECURITY",
465        ] {
466            runtime.execute_query(query).expect(query);
467        }
468
469        let bundle = SnapshotBundle {
470            snapshot: None,
471            auth: Some(("alice".to_string(), Role::Read)),
472            auth_principal: Some(crate::auth::UserId::platform("alice")),
473            tenant: None,
474        };
475        with_snapshot_bundle(&bundle, || {
476            let nodes = runtime
477                .scan_collection("links", None, 10)
478                .expect("typed graph scan");
479            assert_eq!(nodes.total, 1);
480            assert_eq!(nodes.items.len(), 1);
481
482            let documents = runtime
483                .scan_collection("papers", None, 10)
484                .expect("typed document scan");
485            assert_eq!(documents.total, 1);
486            assert_eq!(documents.items.len(), 1);
487        });
488    }
489}
reddb_server/runtime/impl_catalog_accessors.rs

reddb_server/runtime/
impl_catalog_accessors.rs
