radixdb_executor/procedural/
call.rs

1use std::sync::atomic::{AtomicU64, Ordering};
2use std::sync::Arc;
3
4use radixdb_catalog::{
5    ArgumentMode, CatalogDataType, CatalogGeneration, CatalogObject, CatalogPayload, ObjectId,
6    ObjectKind, ResourcePolicy, RoutineDefinition, RoutineResult,
7};
8use radixdb_core::{DataType, Error, Result, Row, Value};
9use radixdb_procedural::{
10    BudgetOwner, BudgetSnapshot, CancellationProbe, Diagnostic, ExecutionOutcome, Interpreter,
11    PrincipalContext, PrincipalHost, ProceduralResult, RuntimeValue, SqlRowSink, VerifiedProgram,
12};
13use radixdb_sql::{
14    walk_expression_tree_mut, CallArgumentSyntax, CallStatement, Expression, Parser, Precedence,
15};
16use radixdb_storage::traits::Engine;
17use radixdb_storage::traits::{MemoryResult, QueryResult};
18
19use crate::catalog::DdlTransaction;
20use crate::context::ExecutionContext;
21use crate::expression::ExpressionEval;
22use crate::mutation::host::ActiveTransaction;
23use crate::Executor;
24
25use super::error::{cleanup_failed, map_executor_error};
26use super::host::ExecutorProceduralHost;
27use super::load_published_routine;
28use super::transaction_visible_catalog;
29use super::value::scalar_value;
30
31static NEXT_CALL_BOUNDARY_ID: AtomicU64 = AtomicU64::new(1);
32
33/// Transaction-aware result staging supplied by the protocol/API boundary.
34///
35/// Rows may be spooled incrementally, but cannot become visible to the caller
36/// until `publish` is invoked after the call statement commits or releases its
37/// internal savepoint. This avoids an executor-owned materialize-all default.
38pub trait ProceduralResultStage {
39    fn begin(&mut self) -> ProceduralResult<()>;
40    fn stage_row(&mut self, row: Vec<RuntimeValue>) -> ProceduralResult<()>;
41    fn publish(&mut self);
42    fn discard(&mut self);
43}
44
45#[derive(Debug, Clone, PartialEq)]
46pub struct ProceduralCallOutcome {
47    execution: ExecutionOutcome,
48    budget: BudgetSnapshot,
49}
50
51impl ProceduralCallOutcome {
52    pub const fn execution(&self) -> &ExecutionOutcome {
53        &self.execution
54    }
55
56    pub const fn budget(&self) -> BudgetSnapshot {
57        self.budget
58    }
59}
60
61#[doc(hidden)]
62pub enum CallBoundary {
63    OwnedTransaction,
64    CallerSavepoint(String),
65}
66
67struct ResultStageAdapter<'a>(&'a mut dyn ProceduralResultStage);
68
69impl SqlRowSink for ResultStageAdapter<'_> {
70    fn push_row(&mut self, row: Vec<RuntimeValue>) -> ProceduralResult<()> {
71        self.0.stage_row(row)
72    }
73}
74
75#[derive(Default)]
76pub(super) struct CallStatementStage {
77    rows: Vec<Row>,
78    published: bool,
79}
80
81impl ProceduralResultStage for CallStatementStage {
82    fn begin(&mut self) -> ProceduralResult<()> {
83        self.rows.clear();
84        self.published = false;
85        Ok(())
86    }
87
88    fn stage_row(&mut self, row: Vec<RuntimeValue>) -> ProceduralResult<()> {
89        self.rows.push(Row::from_values(
90            row.iter()
91                .map(scalar_value)
92                .collect::<ProceduralResult<Vec<_>>>()?,
93        ));
94        Ok(())
95    }
96
97    fn publish(&mut self) {
98        self.published = true;
99    }
100
101    fn discard(&mut self) {
102        self.rows.clear();
103        self.published = false;
104    }
105}
106
107struct ExternalCallCandidate<'a> {
108    object: &'a CatalogObject,
109    definition: &'a RoutineDefinition,
110    positions: Vec<Option<usize>>,
111    cost: u32,
112}
113
114impl Executor {
115    pub(crate) fn execute_call_statement(
116        &self,
117        statement: &CallStatement,
118        context: &ExecutionContext,
119    ) -> Result<Box<dyn QueryResult>> {
120        // Dispatch owns the shared DDL fence for the outer CALL. Reuse that
121        // ownership for every procedural SQL leaf instead of recursively
122        // acquiring the read side after a DDL writer may have queued.
123        let executor = self.fork_with_owned_ddl_fence();
124        let boundary = executor.begin_procedural_boundary()?;
125        match execute_call_statement_inside_boundary(&executor, statement, context) {
126            Ok(result) => match executor.complete_procedural_boundary(&boundary) {
127                Ok(()) => Ok(result),
128                Err(error) => {
129                    let _ = executor.abort_procedural_boundary(&boundary);
130                    Err(error)
131                }
132            },
133            Err(error) => {
134                let _ = executor.abort_procedural_boundary(&boundary);
135                Err(error)
136            }
137        }
138    }
139
140    /// Resolve and execute one durable Procedure from the transaction-visible
141    /// catalog generation. Source and typed metadata are verified after the
142    /// call boundary pins that generation.
143    pub fn execute_procedure(
144        &self,
145        routine: ObjectId,
146        arguments: Vec<RuntimeValue>,
147        context: &ExecutionContext,
148        principals: PrincipalContext,
149        results: &mut dyn ProceduralResultStage,
150    ) -> ProceduralResult<ProceduralCallOutcome> {
151        if !self.ddl_fence_already_held {
152            let _fence = self.engine.acquire_ddl_statement_fence(false);
153            return self
154                .fork_with_owned_ddl_fence()
155                .execute_procedure(routine, arguments, context, principals, results);
156        }
157        validate_principals(principals)?;
158        results.begin()?;
159        let boundary = match self.begin_procedural_boundary() {
160            Ok(boundary) => boundary,
161            Err(error) => {
162                results.discard();
163                return Err(map_executor_error(error));
164            }
165        };
166        self.execute_procedure_inside_boundary(
167            routine, arguments, context, principals, results, &boundary,
168        )
169    }
170
171    pub(super) fn execute_procedure_inside_boundary(
172        &self,
173        routine: ObjectId,
174        arguments: Vec<RuntimeValue>,
175        context: &ExecutionContext,
176        principals: PrincipalContext,
177        results: &mut dyn ProceduralResultStage,
178        boundary: &CallBoundary,
179    ) -> ProceduralResult<ProceduralCallOutcome> {
180        crate::authorization::authorize_routine_invocation(
181            self,
182            principals.session_principal,
183            principals.effective_principal,
184            routine,
185        )
186        .map_err(map_executor_error)?;
187        let published = match load_published_routine(self, routine, ObjectKind::Procedure) {
188            Ok(published) => published,
189            Err(error) => {
190                results.discard();
191                return Err(self.abort_after_setup_error(boundary, map_executor_error(error)));
192            }
193        };
194        let budget = match context.procedural_budget() {
195            Some(budget) => budget.clone(),
196            None => match call_budget(context, published.resource_policy) {
197                Ok(budget) => budget,
198                Err(error) => {
199                    results.discard();
200                    return Err(self.abort_after_setup_error(boundary, error));
201                }
202            },
203        };
204        self.execute_inside_boundary(
205            &published.program,
206            arguments,
207            context,
208            principals,
209            (published.security == radixdb_catalog::SecurityMode::Definer)
210                .then_some(published.owner),
211            &budget,
212            results,
213            boundary,
214        )
215    }
216
217    /// Execute one verified procedural program as one atomic engine operation.
218    pub fn execute_procedural_program(
219        &self,
220        program: &VerifiedProgram,
221        arguments: Vec<RuntimeValue>,
222        context: &ExecutionContext,
223        principals: PrincipalContext,
224        policy: ResourcePolicy,
225        results: &mut dyn ProceduralResultStage,
226    ) -> ProceduralResult<ProceduralCallOutcome> {
227        if !self.ddl_fence_already_held {
228            let _fence = self.engine.acquire_ddl_statement_fence(false);
229            return self.fork_with_owned_ddl_fence().execute_procedural_program(
230                program, arguments, context, principals, policy, results,
231            );
232        }
233        validate_principals(principals)?;
234        let budget = call_budget(context, policy)?;
235        budget.check_boundary()?;
236        results.begin()?;
237
238        let boundary = match self.begin_procedural_boundary() {
239            Ok(boundary) => boundary,
240            Err(error) => {
241                results.discard();
242                return Err(map_executor_error(error));
243            }
244        };
245        self.execute_inside_boundary(
246            program, arguments, context, principals, None, &budget, results, &boundary,
247        )
248    }
249
250    #[allow(clippy::too_many_arguments)]
251    fn execute_inside_boundary(
252        &self,
253        program: &VerifiedProgram,
254        arguments: Vec<RuntimeValue>,
255        context: &ExecutionContext,
256        principals: PrincipalContext,
257        definer: Option<ObjectId>,
258        budget: &BudgetOwner,
259        results: &mut dyn ProceduralResultStage,
260        boundary: &CallBoundary,
261    ) -> ProceduralResult<ProceduralCallOutcome> {
262        let execution = {
263            let mut host = ExecutorProceduralHost::new(self, context, principals, None);
264            let mut sink = ResultStageAdapter(results);
265            if let Some(owner) = definer {
266                host.push_definer(owner).and_then(|()| {
267                    let execution = Interpreter
268                        .execute_with_result_sink(program, arguments, &mut host, budget, &mut sink);
269                    let cleanup = host.pop_definer();
270                    match (execution, cleanup) {
271                        (Ok(outcome), Ok(())) => Ok(outcome),
272                        (Err(primary), Ok(())) => Err(primary),
273                        (Ok(_), Err(cleanup)) => Err(cleanup),
274                        (Err(primary), Err(cleanup)) => {
275                            Err(primary.with_detail("definer_cleanup_error", cleanup.to_string()))
276                        }
277                    }
278                })
279            } else {
280                Interpreter
281                    .execute_with_result_sink(program, arguments, &mut host, budget, &mut sink)
282            }
283        };
284
285        match execution {
286            Ok(execution) => {
287                if let Err(error) = self.complete_procedural_boundary(boundary) {
288                    results.discard();
289                    let primary = map_executor_error(error);
290                    return Err(match self.abort_procedural_boundary(boundary) {
291                        Ok(()) => primary,
292                        Err(cleanup) => cleanup_failed(primary, cleanup),
293                    });
294                }
295                results.publish();
296                Ok(ProceduralCallOutcome {
297                    execution,
298                    budget: budget.snapshot(),
299                })
300            }
301            Err(primary) => {
302                results.discard();
303                Err(match self.abort_procedural_boundary(boundary) {
304                    Ok(()) => primary,
305                    Err(cleanup) => cleanup_failed(primary, cleanup),
306                })
307            }
308        }
309    }
310
311    pub(super) fn abort_after_setup_error(
312        &self,
313        boundary: &CallBoundary,
314        primary: Diagnostic,
315    ) -> Diagnostic {
316        match self.abort_procedural_boundary(boundary) {
317            Ok(()) => primary,
318            Err(cleanup) => cleanup_failed(primary, cleanup),
319        }
320    }
321
322    pub(crate) fn begin_procedural_boundary(&self) -> radixdb_core::Result<CallBoundary> {
323        let call_id = NEXT_CALL_BOUNDARY_ID.fetch_add(1, Ordering::Relaxed);
324        let mut active = self.active_transaction.lock().unwrap();
325        if let Some(state) = active.as_mut() {
326            let name = format!("\0radixdb-call-{call_id}");
327            state.create_savepoint(&name)?;
328            return Ok(CallBoundary::CallerSavepoint(name));
329        }
330
331        let mut transaction = self.engine.begin_transaction()?;
332        let catalog = match self.engine.pin_catalog() {
333            Ok(catalog) => DdlTransaction::begin_shared(catalog),
334            Err(error) => {
335                return match transaction.rollback() {
336                    Ok(()) => Err(error),
337                    Err(cleanup) => Err(radixdb_core::Error::internal(format!(
338                        "cannot pin procedural catalog: {error}; transaction rollback also failed: {cleanup}"
339                    ))),
340                };
341            }
342        };
343        *active = Some(ActiveTransaction::new(transaction, catalog));
344        Ok(CallBoundary::OwnedTransaction)
345    }
346
347    pub(crate) fn complete_procedural_boundary(
348        &self,
349        boundary: &CallBoundary,
350    ) -> radixdb_core::Result<()> {
351        match boundary {
352            CallBoundary::OwnedTransaction => self.commit_installed_transaction(),
353            CallBoundary::CallerSavepoint(name) => self.release_active_savepoint(name),
354        }
355    }
356
357    pub(crate) fn abort_procedural_boundary(
358        &self,
359        boundary: &CallBoundary,
360    ) -> radixdb_core::Result<()> {
361        match boundary {
362            CallBoundary::OwnedTransaction => {
363                if self.has_active_transaction() {
364                    self.rollback_installed_transaction()
365                } else {
366                    Ok(())
367                }
368            }
369            CallBoundary::CallerSavepoint(name) => {
370                self.rollback_active_to_savepoint(name)?;
371                self.release_active_savepoint(name)
372            }
373        }
374    }
375}
376
377fn execute_call_statement_inside_boundary(
378    executor: &Executor,
379    statement: &CallStatement,
380    context: &ExecutionContext,
381) -> Result<Box<dyn QueryResult>> {
382    let supplied = statement
383        .arguments
384        .iter()
385        .map(|argument| {
386            ExpressionEval::compile(&argument.value, &[])?
387                .with_context(context)
388                .eval_slice(&Row::new())
389                .map(|value| (argument, value))
390        })
391        .collect::<Result<Vec<_>>>()?;
392    let (catalog, _) = transaction_visible_catalog(executor)?;
393    let candidate = resolve_external_call(catalog.as_ref(), statement, &supplied)?;
394    let arguments = bind_external_arguments(executor, context, &candidate, &supplied)?;
395    let columns = external_call_columns(candidate.definition);
396    let result_model = candidate.definition.result().clone();
397    let mut stage = CallStatementStage::default();
398    let outcome = executor
399        .execute_procedure(
400            candidate.object.id(),
401            arguments,
402            context,
403            PrincipalContext {
404                session_principal: context.principal_id(),
405                invoker_principal: context.effective_principal_id(),
406                effective_principal: context.effective_principal_id(),
407            },
408            &mut stage,
409        )
410        .map_err(procedural_call_error)?;
411    if !stage.published {
412        return Err(Error::internal(
413            "procedure result stage was not published after successful call",
414        ));
415    }
416    let rows = match result_model {
417        RoutineResult::Table(_) => stage.rows,
418        RoutineResult::Void if columns.is_empty() => Vec::new(),
419        RoutineResult::Void => vec![Row::from_values(
420            outcome
421                .execution()
422                .output_values
423                .iter()
424                .map(scalar_value)
425                .collect::<ProceduralResult<Vec<_>>>()
426                .map_err(procedural_call_error)?,
427        )],
428        RoutineResult::Scalar { .. } | RoutineResult::Trigger => {
429            return Err(Error::internal(
430                "procedure catalog object has a function-only result contract",
431            ));
432        }
433    };
434    Ok(Box::new(MemoryResult::with_rows(columns, rows)))
435}
436
437fn resolve_external_call<'a>(
438    catalog: &'a CatalogGeneration,
439    statement: &CallStatement,
440    supplied: &[(&CallArgumentSyntax, Value)],
441) -> Result<ExternalCallCandidate<'a>> {
442    let (namespace, routine_name) = external_call_name(catalog, statement)?;
443    let mut candidates = Vec::new();
444    for object in catalog.objects_of_kind(ObjectKind::Procedure) {
445        if object.namespace_id() != Some(namespace)
446            || !object
447                .name()
448                .normalized()
449                .as_str()
450                .eq_ignore_ascii_case(routine_name)
451        {
452            continue;
453        }
454        let CatalogPayload::Procedure(payload) = object.payload() else {
455            unreachable!("catalog kind/payload invariant")
456        };
457        if let Some((positions, cost)) = external_candidate(payload.definition(), supplied) {
458            candidates.push(ExternalCallCandidate {
459                object,
460                definition: payload.definition(),
461                positions,
462                cost,
463            });
464        }
465    }
466    let minimum = candidates
467        .iter()
468        .map(|candidate| candidate.cost)
469        .min()
470        .ok_or_else(|| {
471            Error::invalid_argument(format!(
472                "no procedure overload matches call {}",
473                statement.routine
474            ))
475        })?;
476    candidates.retain(|candidate| candidate.cost == minimum);
477    if candidates.len() != 1 {
478        return Err(Error::invalid_argument(format!(
479            "procedure call {} has multiple equal-cost overloads",
480            statement.routine
481        )));
482    }
483    Ok(candidates.remove(0))
484}
485
486pub(super) fn bind_job_call(
487    executor: &Executor,
488    catalog: &CatalogGeneration,
489    routine: &radixdb_sql::ObjectName,
490    arguments: &[CallArgumentSyntax],
491    context: &ExecutionContext,
492) -> Result<(ObjectId, Vec<(CatalogDataType, Value)>)> {
493    let supplied = arguments
494        .iter()
495        .map(|argument| {
496            ExpressionEval::compile(&argument.value, &[])?
497                .with_context(context)
498                .eval_slice(&Row::new())
499                .map(|value| (argument, value))
500        })
501        .collect::<Result<Vec<_>>>()?;
502    let statement = CallStatement {
503        token: routine
504            .components
505            .last()
506            .ok_or_else(|| Error::invalid_argument("job procedure name is empty"))?
507            .token
508            .clone(),
509        routine: routine.clone(),
510        arguments: arguments.to_vec(),
511    };
512    let candidate = resolve_external_call(catalog, &statement, &supplied)?;
513    let procedure_id = candidate.object.id();
514    let input_types = candidate
515        .definition
516        .arguments()
517        .iter()
518        .filter(|argument| argument.mode() != ArgumentMode::Out)
519        .map(|argument| argument.data_type())
520        .collect::<Vec<_>>();
521    let values = bind_external_arguments(executor, context, &candidate, &supplied)?
522        .iter()
523        .map(scalar_value)
524        .collect::<ProceduralResult<Vec<_>>>()
525        .map_err(procedural_call_error)?;
526    debug_assert_eq!(input_types.len(), values.len());
527    Ok((procedure_id, input_types.into_iter().zip(values).collect()))
528}
529
530fn external_call_name<'a>(
531    catalog: &CatalogGeneration,
532    statement: &'a CallStatement,
533) -> Result<(ObjectId, &'a str)> {
534    let (name, namespace) = statement
535        .routine
536        .components
537        .split_last()
538        .ok_or_else(|| Error::invalid_argument("procedure name is empty"))?;
539    if namespace.is_empty() {
540        return Ok((ObjectId::BOOTSTRAP_NAMESPACE, name.value.as_str()));
541    }
542    let namespace = crate::catalog::resolve_namespace_path(
543        catalog,
544        namespace.iter().map(|component| component.value.as_str()),
545    )?;
546    Ok((namespace, name.value.as_str()))
547}
548
549fn external_candidate(
550    definition: &RoutineDefinition,
551    supplied: &[(&CallArgumentSyntax, Value)],
552) -> Option<(Vec<Option<usize>>, u32)> {
553    let input_count = definition
554        .arguments()
555        .iter()
556        .filter(|argument| argument.mode() != ArgumentMode::Out)
557        .count();
558    if supplied.len() > input_count {
559        return None;
560    }
561    let mut positions = vec![None; definition.arguments().len()];
562    let input_positions = definition
563        .arguments()
564        .iter()
565        .enumerate()
566        .filter_map(|(index, argument)| (argument.mode() != ArgumentMode::Out).then_some(index))
567        .collect::<Vec<_>>();
568    let mut positional = 0;
569    for (supplied_index, (argument, _)) in supplied.iter().enumerate() {
570        let declared_index = if let Some(name) = &argument.name {
571            definition.arguments().iter().position(|declared| {
572                declared.mode() != ArgumentMode::Out
573                    && declared.name().normalized().as_str() == name.value_lower.as_str()
574            })?
575        } else {
576            let index = *input_positions.get(positional)?;
577            positional += 1;
578            index
579        };
580        if positions[declared_index].replace(supplied_index).is_some() {
581            return None;
582        }
583    }
584    let mut cost = 0_u32;
585    for (index, declared) in definition.arguments().iter().enumerate() {
586        if declared.mode() == ArgumentMode::Out {
587            continue;
588        }
589        let Some(supplied_index) = positions[index] else {
590            if declared.mode() == ArgumentMode::In && declared.default_sql().is_some() {
591                continue;
592            }
593            return None;
594        };
595        let value = &supplied[supplied_index].1;
596        if value.is_null() {
597            if !declared.nullable() {
598                return None;
599            }
600            if value.data_type() != DataType::Null
601                && value.data_type() != declared.data_type().logical_type()
602            {
603                return None;
604            }
605        } else if value.data_type() == declared.data_type().logical_type() {
606        } else if matches!(
607            (value.data_type(), declared.data_type().logical_type()),
608            (DataType::Integer, DataType::Decimal) | (DataType::Date, DataType::Timestamp)
609        ) {
610            cost = cost.saturating_add(1);
611        } else {
612            return None;
613        }
614    }
615    Some((positions, cost))
616}
617
618fn bind_external_arguments(
619    executor: &Executor,
620    context: &ExecutionContext,
621    candidate: &ExternalCallCandidate<'_>,
622    supplied: &[(&CallArgumentSyntax, Value)],
623) -> Result<Vec<RuntimeValue>> {
624    let mut arguments = Vec::new();
625    let mut earlier = std::collections::BTreeMap::new();
626    for (index, declared) in candidate.definition.arguments().iter().enumerate() {
627        if declared.mode() == ArgumentMode::Out {
628            continue;
629        }
630        let value = if let Some(supplied_index) = candidate.positions[index] {
631            supplied[supplied_index]
632                .1
633                .try_coerce_to_type(declared.data_type().logical_type())?
634        } else {
635            let default = declared.default_sql().ok_or_else(|| {
636                Error::invalid_argument(format!(
637                    "required procedure argument '{}' is missing",
638                    declared.name().display().as_str()
639                ))
640            })?;
641            evaluate_external_default(executor, context, default.as_str(), &earlier)?
642                .try_coerce_to_type(declared.data_type().logical_type())?
643        };
644        if value.is_null() && !declared.nullable() {
645            return Err(Error::invalid_argument(format!(
646                "procedure argument '{}' is NOT NULL",
647                declared.name().display().as_str()
648            )));
649        }
650        earlier.insert(
651            declared.name().normalized().as_str().to_owned(),
652            value.clone(),
653        );
654        arguments.push(RuntimeValue::scalar(value));
655    }
656    Ok(arguments)
657}
658
659fn evaluate_external_default(
660    _executor: &Executor,
661    context: &ExecutionContext,
662    source: &str,
663    earlier: &std::collections::BTreeMap<String, Value>,
664) -> Result<Value> {
665    let mut parser = Parser::new(source);
666    let mut expression = parser
667        .parse_expression(Precedence::Lowest)
668        .ok_or_else(|| Error::parse("procedure default is not an expression"))?;
669    if let Some(error) = parser.errors().first() {
670        return Err(Error::parse(error.to_string()));
671    }
672    walk_expression_tree_mut(&mut expression, &mut |node| {
673        let Expression::Identifier(identifier) = node else {
674            return;
675        };
676        if let Some(value) = earlier.get(identifier.value_lower()) {
677            *node = Expression::BoundValue(Box::new(value.clone()));
678        }
679    });
680    ExpressionEval::compile(&expression, &[])?
681        .with_context(context)
682        .eval_slice(&Row::new())
683}
684
685fn external_call_columns(definition: &RoutineDefinition) -> Vec<String> {
686    match definition.result() {
687        RoutineResult::Table(columns) => columns
688            .iter()
689            .map(|column| column.name().display().as_str().to_owned())
690            .collect(),
691        RoutineResult::Void => definition
692            .arguments()
693            .iter()
694            .filter(|argument| argument.mode() != ArgumentMode::In)
695            .map(|argument| argument.name().display().as_str().to_owned())
696            .collect(),
697        RoutineResult::Scalar { .. } | RoutineResult::Trigger => Vec::new(),
698    }
699}
700
701pub(super) fn procedural_call_error(error: Diagnostic) -> Error {
702    if error.category() == radixdb_procedural::DiagnosticCategory::Security {
703        Error::authorization_denied(error.to_string())
704    } else {
705        Error::invalid_argument(error.to_string())
706    }
707}
708
709pub(super) fn call_budget(
710    context: &ExecutionContext,
711    mut policy: ResourcePolicy,
712) -> ProceduralResult<BudgetOwner> {
713    if context.timeout_ms() > 0 {
714        policy.deadline_ms = policy.deadline_ms.min(context.timeout_ms());
715    }
716    let cancellation: Arc<dyn CancellationProbe> = Arc::new(context.cancellation_handle());
717    BudgetOwner::with_parent_cancellation(policy, cancellation)
718}
719
720fn validate_principals(principals: PrincipalContext) -> ProceduralResult<()> {
721    let all = [
722        principals.session_principal,
723        principals.invoker_principal,
724        principals.effective_principal,
725    ];
726    if all.contains(&ObjectId::BOOTSTRAP_NAMESPACE) {
727        return Err(Diagnostic::new(
728            radixdb_procedural::DiagnosticKind::SecurityObjectDenied,
729            "namespace identity cannot execute as a principal",
730        ));
731    }
732    Ok(())
733}
radixdb_executor/procedural/call.rs

radixdb_executor/procedural/
call.rs
