use super::*;
fn runtime_pool_lock(runtime: &RedDBRuntime) -> std::sync::MutexGuard<'_, PoolState> {
runtime
.inner
.pool
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
}
impl RedDBRuntime {
pub fn acquire(&self) -> RedDBResult<RuntimeConnection> {
self.acquire_connection(true)
}
pub(crate) fn acquire_wire_connection(&self) -> RedDBResult<RuntimeConnection> {
self.acquire_connection(false)
}
fn acquire_connection(&self, uses_pool_slot: bool) -> RedDBResult<RuntimeConnection> {
let mut pool = self
.inner
.pool
.lock()
.map_err(|e| RedDBError::Internal(format!("connection pool lock poisoned: {e}")))?;
if uses_pool_slot && pool.pooled_active >= self.inner.pool_config.max_connections {
return Err(RedDBError::Internal(
"connection pool exhausted".to_string(),
));
}
let id = if let Some(id) = pool.idle.pop() {
id
} else {
let id = pool.next_id;
pool.next_id += 1;
id
};
pool.active += 1;
if uses_pool_slot {
pool.pooled_active += 1;
}
pool.total_checkouts += 1;
drop(pool);
self.inner.node_load_telemetry.record_connect();
Ok(RuntimeConnection {
background_tasks: self.background_tasks.clone(),
id,
uses_pool_slot,
inner: Arc::clone(&self.inner),
})
}
pub fn checkpoint(&self) -> RedDBResult<()> {
self.checkpoint_local()?;
if let Err(err) = self.assert_remote_write_allowed("checkpoint") {
tracing::warn!(
target: "reddb::serverless::lease",
error = %err,
"checkpoint: skipping remote upload — lease not held"
);
return Ok(());
}
self.inner
.db
.upload_to_remote_backend()
.map_err(|err| RedDBError::Engine(err.to_string()))
}
pub(super) fn checkpoint_local(&self) -> RedDBResult<()> {
let started = std::time::Instant::now();
self.seal_hypertable_chunks_for_checkpoint(
self.inner.checkpoint_columnar_emission_budget_chunks,
)?;
let store = self.inner.db.store();
let publication = store.vault_publication_lock.lock();
self.prepare_paged_counts()?;
self.persist_probabilistic_snapshots()?;
self.inner.db.flush_local_only().map_err(|err| {
let msg = err.to_string();
crate::telemetry::operator_event::OperatorEvent::CheckpointFailed {
lsn: 0,
error: msg.clone(),
}
.emit_global();
crate::telemetry::operator_event::OperatorEvent::WalFsyncFailed {
path: "<flush_local_only>".to_string(),
error: msg.clone(),
}
.emit_global();
RedDBError::Engine(msg)
})?;
drop(publication);
let checkpoint_lsn = self.cdc_current_lsn();
self.inner
.checkpoint_projection_stats
.record_checkpoint(checkpoint_lsn, started.elapsed().as_millis() as u64);
Ok(())
}
pub(crate) fn assert_remote_write_allowed(&self, action: &str) -> RedDBResult<()> {
if self.inner.db.remote_backend.is_none() {
return Ok(());
}
match self.inner.write_gate.lease_state() {
crate::runtime::write_gate::LeaseGateState::NotHeld => {
self.inner.audit_log.record(
action,
"system",
"remote_backend",
"err: writer lease not held",
crate::json::Value::Null,
);
Err(RedDBError::ReadOnly(format!(
"writer lease not held — {action} blocked (serverless fence)"
)))
}
_ => Ok(()),
}
}
pub fn run_maintenance(&self) -> RedDBResult<()> {
self.inner
.db
.run_maintenance()
.map_err(|err| RedDBError::Internal(err.to_string()))
}
pub fn scan_collection(
&self,
collection: &str,
cursor: Option<ScanCursor>,
limit: usize,
) -> RedDBResult<ScanPage> {
use reddb_rql::ast::PolicyAction;
self.check_collection_privilege(PolicyAction::Select, collection)?;
let store = self.inner.db.store();
let manager = store
.get_collection(collection)
.ok_or_else(|| RedDBError::NotFound(collection.to_string()))?;
let offset = cursor.map(|cursor| cursor.offset).unwrap_or(0);
let limit = limit.max(1);
let read_bytes = usize::try_from(self.memory_budget().resolved_bytes / 16)
.unwrap_or(usize::MAX)
.clamp(1, 16 * 1024 * 1024);
let (total, items) = if crate::runtime::execution_context::current_auth_identity().is_some()
&& crate::runtime::rls_injection::rls_is_enabled(self, collection)
{
manager.try_page_in_id_order_where(offset, limit, read_bytes, |entity| {
crate::runtime::rls_injection::entity_passes_rls_action(
self,
collection,
PolicyAction::Select,
entity,
)
})?
} else {
manager.try_page_in_id_order(offset, limit, read_bytes)?
};
let end = total.min(offset.saturating_add(limit));
let next = (end < total).then_some(ScanCursor { offset: end });
Ok(ScanPage {
collection: collection.to_string(),
items,
next,
total,
})
}
pub fn get_collection_entity(
&self,
collection: &str,
id: crate::storage::EntityId,
) -> RedDBResult<Option<crate::storage::UnifiedEntity>> {
use reddb_rql::ast::PolicyAction;
self.check_collection_privilege(PolicyAction::Select, collection)?;
let entity = self.inner.db.store().get(collection, id);
Ok(entity.filter(|entity| {
crate::runtime::rls_injection::entity_passes_rls_action(
self,
collection,
PolicyAction::Select,
entity,
)
}))
}
pub fn catalog(&self) -> CatalogModelSnapshot {
self.inner.db.catalog_model_snapshot()
}
pub fn try_catalog(&self) -> RedDBResult<CatalogModelSnapshot> {
if self
.inner
.db
.store()
.config()
.resident_memory_bytes_max
.is_none()
{
return Ok(self.catalog());
}
let read_bytes = usize::try_from(self.memory_budget().resolved_bytes / 16)
.unwrap_or(usize::MAX)
.clamp(1, 16 * 1024 * 1024);
let _scratch = self
.try_reserve_query_buffer(read_bytes * 2)
.ok_or_else(|| {
self.memory_accounting().record_admission_denied();
RedDBError::MaterializationLimitExceeded {
executor: "catalog",
limit: usize::try_from(self.memory_budget().resolved_bytes)
.unwrap_or(usize::MAX),
current: usize::try_from(self.memory_accounting().total_used_bytes())
.unwrap_or(usize::MAX)
.saturating_add(1),
}
})?;
Ok(self.inner.db.try_catalog_model_snapshot(read_bytes)?)
}
pub fn catalog_consistency_report(&self) -> crate::catalog::CatalogConsistencyReport {
self.inner.db.catalog_consistency_report()
}
pub fn catalog_attention_summary(&self) -> CatalogAttentionSummary {
crate::catalog::attention_summary(&self.catalog())
}
pub fn collection_attention(&self) -> Vec<CollectionDescriptor> {
crate::catalog::collection_attention(&self.catalog())
}
pub fn index_attention(&self) -> Vec<CatalogIndexStatus> {
crate::catalog::index_attention(&self.catalog())
}
pub fn graph_projection_attention(&self) -> Vec<CatalogGraphProjectionStatus> {
crate::catalog::graph_projection_attention(&self.catalog())
}
pub fn analytics_job_attention(&self) -> Vec<CatalogAnalyticsJobStatus> {
crate::catalog::analytics_job_attention(&self.catalog())
}
pub fn stats(&self) -> RuntimeStats {
let pool = runtime_pool_lock(self);
RuntimeStats {
active_connections: pool.active,
idle_connections: pool.idle.len(),
total_checkouts: pool.total_checkouts,
paged_mode: self.inner.db.is_paged(),
started_at_unix_ms: self.inner.started_at_unix_ms,
store: self.inner.db.stats(),
system: SystemInfo::collect(),
result_blob_cache: self.inner.result_blob_cache.stats(),
kv: self.inner.kv_stats.snapshot(),
metrics_ingest: self.inner.metrics_ingest_stats.snapshot(),
}
}
}
#[cfg(test)]
mod security_boundary_tests {
use super::*;
use crate::application::entity::{DeleteEntityInput, PatchEntityInput};
use crate::application::ports::RuntimeEntityPort;
use crate::auth::Role;
use crate::runtime::execution_context::{with_snapshot_bundle, SnapshotBundle};
use crate::storage::{EntityData, UnifiedEntity};
use reddb_types::Value;
fn text_field<'a>(entity: &'a UnifiedEntity, name: &str) -> Option<&'a str> {
let EntityData::Row(row) = &entity.data else {
return None;
};
let value = row
.named
.as_ref()
.and_then(|named| named.get(name))
.or_else(|| {
row.schema.as_ref().and_then(|schema| {
schema
.iter()
.position(|column| column == name)
.and_then(|index| row.columns.get(index))
})
})?;
match value {
Value::Text(value) => Some(value.as_ref()),
_ => None,
}
}
#[test]
fn typed_entity_boundary_filters_before_pagination_and_hides_foreign_ids() {
let runtime = RedDBRuntime::in_memory().expect("runtime");
runtime
.execute_query(
"CREATE TABLE docs (id INT, tenant_id TEXT, body TEXT) TENANT BY (tenant_id)",
)
.expect("tenant table");
runtime
.execute_query(
"INSERT INTO docs (id, tenant_id, body) VALUES \
(1, 'acme', 'first'), (2, 'globex', 'hidden'), (3, 'acme', 'second')",
)
.expect("seed platform rows");
let unfiltered = runtime
.scan_collection("docs", None, 10)
.expect("platform scan");
let foreign_id = unfiltered
.items
.iter()
.find(|entity| text_field(entity, "tenant_id") == Some("globex"))
.map(|entity| entity.id)
.expect("foreign row");
let bundle = SnapshotBundle {
snapshot: None,
auth: Some(("alice".to_string(), Role::Write)),
tenant: Some("acme".to_string()),
};
with_snapshot_bundle(&bundle, || {
let first = runtime
.scan_collection("docs", None, 1)
.expect("first visible page");
assert_eq!(first.total, 2);
assert_eq!(first.items.len(), 1);
assert_eq!(text_field(&first.items[0], "tenant_id"), Some("acme"));
assert_eq!(first.next, Some(ScanCursor { offset: 1 }));
let second = runtime
.scan_collection("docs", first.next, 1)
.expect("second visible page");
assert_eq!(second.total, 2);
assert_eq!(second.items.len(), 1);
assert_eq!(text_field(&second.items[0], "tenant_id"), Some("acme"));
assert_eq!(second.next, None);
assert!(runtime
.get_collection_entity("docs", foreign_id)
.expect("typed get")
.is_none());
assert!(matches!(
runtime.patch_entity(PatchEntityInput {
collection: "docs".to_string(),
id: foreign_id,
payload: crate::json::Value::Object(Default::default()),
operations: Vec::new(),
}),
Err(crate::RedDBError::NotFound(_))
));
let deleted = runtime
.delete_entity(DeleteEntityInput {
collection: "docs".to_string(),
id: foreign_id,
})
.expect("typed delete");
assert!(!deleted.deleted);
});
assert!(runtime.db().store().get("docs", foreign_id).is_some());
}
#[test]
fn typed_entity_boundary_uses_model_specific_rls_policies() {
let runtime = RedDBRuntime::in_memory().expect("runtime");
for query in [
"CREATE GRAPH links",
"INSERT INTO links NODE (label,node_type,visible) VALUES ('shown','Item',true)",
"INSERT INTO links NODE (label,node_type,visible) VALUES ('hidden','Item',false)",
"CREATE POLICY visible_nodes ON NODES OF links FOR SELECT USING (properties.visible=true)",
"ALTER TABLE links ENABLE ROW LEVEL SECURITY",
"CREATE DOCUMENT papers",
"INSERT INTO papers DOCUMENT VALUES ({name:'shown',visible:true})",
"INSERT INTO papers DOCUMENT VALUES ({name:'hidden',visible:false})",
"CREATE POLICY visible_documents ON DOCUMENTS OF papers FOR SELECT USING (visible=true)",
"ALTER TABLE papers ENABLE ROW LEVEL SECURITY",
] {
runtime.execute_query(query).expect(query);
}
let bundle = SnapshotBundle {
snapshot: None,
auth: Some(("alice".to_string(), Role::Read)),
tenant: None,
};
with_snapshot_bundle(&bundle, || {
let nodes = runtime
.scan_collection("links", None, 10)
.expect("typed graph scan");
assert_eq!(nodes.total, 1);
assert_eq!(nodes.items.len(), 1);
let documents = runtime
.scan_collection("papers", None, 10)
.expect("typed document scan");
assert_eq!(documents.total, 1);
assert_eq!(documents.items.len(), 1);
});
}
}