reddb_server/runtime/
graph_contract.rs

1//! Runtime ownership of optional graph contracts and projected-state validation.
2use super::*;
3use reddb_engine::catalog::GraphContract;
4use reddb_engine::transaction::read::{entity_visibility_check, SnapshotContext};
5
6impl RedDBRuntime {
7    pub fn graph_contract(&self, collection: &str) -> RedDBResult<Option<GraphContract>> {
8        self.check_collection_privilege(reddb_rql::ast::PolicyAction::Select, collection)?;
9        let contract = self
10            .db()
11            .collection_contract(collection)
12            .ok_or_else(|| RedDBError::NotFound(format!("collection '{collection}'")))?;
13        Ok(contract.graph_contract)
14    }
15
16    pub(crate) fn has_graph_contract(&self, collection: &str) -> bool {
17        self.db()
18            .collection_contract_arc(collection)
19            .is_some_and(|contract| contract.graph_contract.is_some())
20    }
21
22    pub fn set_graph_contract(
23        &self,
24        collection: &str,
25        graph: Option<GraphContract>,
26    ) -> RedDBResult<()> {
27        self.check_write(super::write_gate::WriteKind::Ddl)?;
28        if let (Some(auth), Some((username, role))) = (
29            self.inner.auth_store.read().clone(),
30            super::execution_context::current_auth_identity(),
31        ) {
32            let tenant = super::execution_context::current_tenant();
33            let principal = crate::auth::UserId::from_parts(tenant.as_deref(), &username);
34            self.check_ddl_collection_privilege(
35                &auth,
36                &principal,
37                role,
38                tenant.as_deref(),
39                &username,
40                "alter",
41                collection,
42            )
43            .map_err(RedDBError::Query)?;
44        }
45        let store = self.db().store();
46        let _publication = store.vault_publication_lock.lock();
47        if self.snapshot_manager().oldest_active_xid().is_some() {
48            return Err(RedDBError::InvalidOperation(
49                "finish active transactions before changing a graph contract".into(),
50            ));
51        }
52        let mut contract = self
53            .db()
54            .collection_contract(collection)
55            .ok_or_else(|| RedDBError::NotFound(format!("collection '{collection}'")))?;
56        if contract.declared_model != crate::catalog::CollectionModel::Graph {
57            return Err(RedDBError::InvalidOperation(
58                "graph contracts require a Graph collection".into(),
59            ));
60        }
61        if let Some(graph) = &graph {
62            graph.validate()?;
63            for snapshot in self.vcs_live_graph_validation_contexts(collection)? {
64                self.validate_graph_projection(collection, graph, &snapshot, &[], &[])?;
65            }
66            // Earlier readers must reject the file instead of ignoring a contract
67            // they cannot enforce. V14 also carries branch provenance.
68            if store.format_version() < reddb_file::native_store::STORE_VERSION_V14 {
69                let previous = store.format_version();
70                store.set_format_version(reddb_file::native_store::STORE_VERSION_V14);
71                if store.pager().is_some() {
72                    if let Err(error) = store.persist() {
73                        store.set_format_version(previous);
74                        return Err(error.into());
75                    }
76                }
77            }
78        }
79        contract.graph_contract = graph;
80        contract.version = contract.version.checked_add(1).ok_or_else(|| {
81            RedDBError::InvalidOperation("collection contract version exhausted".into())
82        })?;
83        contract.updated_at_unix_ms = std::time::SystemTime::now()
84            .duration_since(std::time::UNIX_EPOCH)
85            .unwrap_or_default()
86            .as_millis();
87        self.db()
88            .save_collection_contract(contract)
89            .map_err(|error| RedDBError::Internal(error.to_string()))?;
90        self.invalidate_result_cache();
91        Ok(())
92    }
93
94    pub(crate) fn validate_graph_changes(
95        &self,
96        collection: &str,
97        added: &[&UnifiedEntity],
98        removed: &[EntityId],
99    ) -> RedDBResult<()> {
100        let Some(contract) = self.db().collection_contract_arc(collection) else {
101            return Ok(());
102        };
103        let Some(graph) = &contract.graph_contract else {
104            return Ok(());
105        };
106        let (mut snapshot, _retention) = self.native_read_snapshot()?;
107        snapshot.own_xids = self.own_transaction_xids();
108        if self.current_xid().is_none() {
109            snapshot.snapshot = self.snapshot_manager().fresh_read_snapshot();
110            if let Some(branch) = snapshot.branch.as_mut() {
111                let branch = Arc::make_mut(branch);
112                branch
113                    .epochs
114                    .insert(branch.writer_epoch, snapshot.snapshot.clone());
115            }
116        }
117        self.validate_graph_projection(collection, graph, &snapshot, added, removed)
118    }
119
120    /// Called under the publication fence, before committing the transaction.
121    /// Refresh only the writer epoch; inherited branches retain their fork view.
122    pub(crate) fn validate_graph_contracts_at_commit(&self) -> RedDBResult<()> {
123        let (mut context, _retention) = self.native_read_snapshot()?;
124        context.snapshot = self.snapshot_manager().fresh_read_snapshot();
125        context.own_xids = self.own_transaction_xids();
126        if let Some(branch) = context.branch.as_mut() {
127            let branch = Arc::make_mut(branch);
128            branch
129                .epochs
130                .insert(branch.writer_epoch, context.snapshot.clone());
131        }
132        for contract in self.db().collection_contracts() {
133            if let Some(graph) = &contract.graph_contract {
134                self.validate_graph_projection(&contract.name, graph, &context, &[], &[])?;
135            }
136        }
137        Ok(())
138    }
139
140    pub(crate) fn validate_graph_entities(
141        &self,
142        collection: &str,
143        graph: &GraphContract,
144        entities: &[&UnifiedEntity],
145    ) -> RedDBResult<()> {
146        let Some(column) = self.tenant_column(collection) else {
147            graph.validate_entities(entities)?;
148            return Ok(());
149        };
150        // Integrity is checked across all tenants, but a tenant's edge may only
151        // resolve endpoints in its own graph. RLS cannot hide invalid endpoints
152        // from admission or let another tenant satisfy an endpoint reference.
153        let mut tenants: HashMap<String, Vec<&UnifiedEntity>> = HashMap::new();
154        for &entity in entities {
155            let properties = match &entity.data {
156                EntityData::Node(node) => &node.properties,
157                EntityData::Edge(edge) => &edge.properties,
158                _ => continue,
159            };
160            let tenant = if let Some((root, tail)) = column.split_once('.') {
161                super::impl_dml_support::dotted_tenant_text(
162                    properties.get(root).unwrap_or(&Value::Null),
163                    tail,
164                )?
165            } else {
166                properties.get(&column).and_then(|value| match value {
167                    Value::Text(value) => Some(value.to_string()),
168                    _ => None,
169                })
170            }
171            .ok_or_else(|| {
172                RedDBError::Query("graph contract requires a text tenant discriminator".into())
173            })?;
174            tenants.entry(tenant).or_default().push(entity);
175        }
176        for entities in tenants.values() {
177            graph.validate_entities(entities)?;
178        }
179        Ok(())
180    }
181
182    fn validate_graph_projection(
183        &self,
184        collection: &str,
185        graph: &GraphContract,
186        snapshot: &SnapshotContext,
187        added: &[&UnifiedEntity],
188        removed: &[EntityId],
189    ) -> RedDBResult<()> {
190        let store = self.db().store();
191        let manager = store
192            .get_collection(collection)
193            .ok_or_else(|| RedDBError::NotFound(format!("collection '{collection}'")))?;
194        let (_scan_memory, read_bytes_max) = self.reserve_index_scan(&manager)?;
195        let mut rows = Vec::new();
196        let mut reservations = Vec::new();
197        super::function_budget::try_scan_result(
198            |visit| manager.try_for_each_entity(read_bytes_max, visit),
199            |entity| {
200                if !removed.contains(&entity.id)
201                    && entity_visibility_check(snapshot, entity)
202                    && matches!(
203                        entity.kind,
204                        EntityKind::GraphNode(_) | EntityKind::GraphEdge(_)
205                    )
206                {
207                    let bytes = crate::storage::unified::memory_size::entity_bytes(entity)
208                        .saturating_mul(2)
209                        .saturating_add(512);
210                    reservations.push(
211                        self.try_reserve_query_buffer_from_sample(bytes)
212                            .ok_or_else(|| {
213                                RedDBError::Query(
214                                    "graph contract validation exceeds the query memory budget"
215                                        .into(),
216                                )
217                            })?,
218                    );
219                    rows.push(entity.clone());
220                }
221                Ok(true)
222            },
223        )?;
224        let entities: Vec<_> = rows.iter().chain(added.iter().copied()).collect();
225        self.validate_graph_entities(collection, graph, &entities)?;
226        Ok(())
227    }
228}
reddb_server/runtime/graph_contract.rs

reddb_server/runtime/
graph_contract.rs
