radixdb_api/
server_runtime.rs

1// Copyright 2026 RadixDB Contributors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Narrow library contract used by the network runtime.
16//!
17//! The TCP state machine owns frames, authentication and cursors. This module
18//! prevents it from also depending on executor contexts, MVCC lifecycle enums,
19//! physical column storage or the instrumentation implementation.
20
21use std::time::Duration;
22
23use crate::ObjectId;
24use crate::{Database, NamedParams, ParamVec};
25use radixdb_core::{Error, Result};
26use radixdb_executor::context::{CancellationHandle, ExecutionContext};
27#[doc(hidden)]
28pub use radixdb_executor::procedural::{
29    Diagnostic as ServerJobDiagnostic, DiagnosticKind as ServerJobDiagnosticKind,
30    JobAttemptMetadata as ServerJobAttemptMetadata, JobAttemptOutcome as ServerJobAttemptOutcome,
31    ScheduledJobDefinition as ServerScheduledJobDefinition,
32    ScheduledJobSchedule as ServerScheduledJobSchedule,
33};
34use radixdb_storage::instrumentation::ProtocolColumnBatchFallback;
35use radixdb_storage::traits::TypedBatchFallbackReason;
36use radixdb_storage::volume::column::ColumnData;
37
38/// Opaque cancellation signal shared by a server session and its requests.
39#[derive(Clone, Debug)]
40pub struct ServerCancellation {
41    pub(crate) inner: CancellationHandle,
42}
43
44impl ServerCancellation {
45    /// Create an independent cancellation signal.
46    pub fn new() -> Self {
47        Self {
48            inner: ExecutionContext::new().cancellation_handle(),
49        }
50    }
51
52    /// Request cancellation.
53    pub fn cancel(&self) {
54        self.inner.cancel();
55    }
56
57    /// Return whether cancellation has been requested.
58    pub fn is_cancelled(&self) -> bool {
59        self.inner.is_cancelled()
60    }
61}
62
63impl Default for ServerCancellation {
64    fn default() -> Self {
65        Self::new()
66    }
67}
68
69/// Opaque request context accepted by embedded facade methods used by server.
70pub struct ServerExecutionContext {
71    pub(crate) inner: ExecutionContext,
72}
73
74impl ServerExecutionContext {
75    /// Construct a positional-parameter request context.
76    pub fn positional(params: ParamVec) -> Self {
77        Self {
78            inner: ExecutionContext::with_params(params),
79        }
80    }
81
82    /// Construct a named-parameter request context.
83    pub fn named(params: NamedParams) -> Self {
84        Self {
85            inner: ExecutionContext::with_named_params(params.into_inner()),
86        }
87    }
88
89    /// Bind the authenticated session identity and immutable protocol request
90    /// ID after user parameters have been admitted.
91    pub fn bind_request_identity(&mut self, principal_id: ObjectId, request_id: u64) -> Result<()> {
92        self.inner = self.inner.with_principal_id(principal_id);
93        self.inner.set_request_id(request_id)
94    }
95
96    /// Inherit session/server shutdown cancellation without sharing the
97    /// request-local cancellation bit.
98    pub fn bind_parent_cancellation(&mut self, parent: &ServerCancellation) {
99        self.inner.bind_parent_cancellation(&parent.inner);
100    }
101
102    /// Obtain the request-local signal registered under the protocol request ID.
103    pub fn cancellation(&self) -> ServerCancellation {
104        ServerCancellation {
105            inner: self.inner.cancellation_handle(),
106        }
107    }
108
109    pub(crate) fn inner(&self) -> &ExecutionContext {
110        &self.inner
111    }
112
113    pub(crate) fn into_inner(self) -> ExecutionContext {
114        self.inner
115    }
116}
117
118/// Stable facade view of a database engine lifecycle.
119#[derive(Clone, Debug, PartialEq, Eq)]
120pub enum DatabaseRuntimeState {
121    Closed,
122    Opening,
123    Ready,
124    Closing,
125    CloseFailed(Error),
126    Failed(Error),
127}
128
129impl Database {
130    /// Snapshot enabled durable Job definitions for the stock scheduler host.
131    #[doc(hidden)]
132    pub fn scheduled_jobs_snapshot(&self) -> Result<Vec<ServerScheduledJobDefinition>> {
133        self.with_connection_executor(|executor| executor.scheduled_jobs_snapshot())
134    }
135
136    /// Execute one scheduler-owned attempt through the ordinary procedural
137    /// transaction boundary without exposing the executor to the stock server.
138    #[doc(hidden)]
139    pub fn execute_scheduled_job_attempt(
140        &self,
141        job_id: ObjectId,
142        metadata: ServerJobAttemptMetadata,
143        cancellation: &ServerCancellation,
144    ) -> std::result::Result<ServerJobAttemptOutcome, ServerJobDiagnostic> {
145        self.with_connection_executor(|executor| {
146            let mut context = ExecutionContext::new();
147            context.bind_parent_cancellation(&cancellation.inner);
148            Ok(executor.execute_job_attempt(job_id, metadata, &context))
149        })
150        .map_err(|error| {
151            ServerJobDiagnostic::new(
152                ServerJobDiagnosticKind::RuntimeInvalidState,
153                error.to_string(),
154            )
155        })?
156    }
157
158    /// Return lifecycle state without exposing the storage engine handle.
159    pub fn runtime_state(&self) -> DatabaseRuntimeState {
160        match self.engine().lifecycle_state() {
161            radixdb_storage::mvcc::engine::EngineLifecycleState::Closed => {
162                DatabaseRuntimeState::Closed
163            }
164            radixdb_storage::mvcc::engine::EngineLifecycleState::Opening => {
165                DatabaseRuntimeState::Opening
166            }
167            radixdb_storage::mvcc::engine::EngineLifecycleState::Ready => {
168                DatabaseRuntimeState::Ready
169            }
170            radixdb_storage::mvcc::engine::EngineLifecycleState::Closing => {
171                DatabaseRuntimeState::Closing
172            }
173            radixdb_storage::mvcc::engine::EngineLifecycleState::CloseFailed(error) => {
174                DatabaseRuntimeState::CloseFailed(error)
175            }
176            radixdb_storage::mvcc::engine::EngineLifecycleState::Failed(error) => {
177                DatabaseRuntimeState::Failed(error)
178            }
179        }
180    }
181}
182
183/// Typed column value crossing the embedded facade into a transport runtime.
184pub enum ServerColumnData {
185    Int64 {
186        values: Vec<i64>,
187        nulls: Vec<bool>,
188    },
189    Float64 {
190        values: Vec<f64>,
191        nulls: Vec<bool>,
192    },
193    TimestampNanos {
194        values: Vec<i64>,
195        nulls: Vec<bool>,
196    },
197    Boolean {
198        values: Vec<bool>,
199        nulls: Vec<bool>,
200    },
201    DictionaryText {
202        ids: Vec<u32>,
203        dictionary: Vec<String>,
204        nulls: Vec<bool>,
205    },
206    Bytes {
207        data: Vec<u8>,
208        offsets: Vec<(u64, u64)>,
209        nulls: Vec<bool>,
210    },
211    JsonText {
212        data: Vec<u8>,
213        offsets: Vec<(u64, u64)>,
214        nulls: Vec<bool>,
215    },
216    External {
217        data: Vec<u8>,
218        offsets: Vec<(u64, u64)>,
219        type_ref: radixdb_core::ExternalTypeRef,
220        nulls: Vec<bool>,
221    },
222}
223
224/// One decoded, output-ordered column batch independent of storage types.
225pub struct ServerColumnBatch {
226    row_count: usize,
227    columns: Vec<ServerColumnData>,
228}
229
230impl ServerColumnBatch {
231    pub fn row_count(&self) -> usize {
232        self.row_count
233    }
234
235    pub fn into_columns(self) -> Vec<ServerColumnData> {
236        self.columns
237    }
238
239    pub(crate) fn from_storage(batch: radixdb_storage::traits::TypedColumnBatch) -> Result<Self> {
240        let row_count = batch.row_count();
241        let columns = batch
242            .into_columns()
243            .into_iter()
244            .map(server_column_from_storage)
245            .collect::<Result<Vec<_>>>()?;
246        Ok(Self { row_count, columns })
247    }
248}
249
250fn server_column_from_storage(column: ColumnData) -> Result<ServerColumnData> {
251    match column {
252        ColumnData::Int64 { values, nulls } => Ok(ServerColumnData::Int64 { values, nulls }),
253        ColumnData::Float64 { values, nulls } => Ok(ServerColumnData::Float64 { values, nulls }),
254        ColumnData::TimestampNanos { values, nulls } => {
255            Ok(ServerColumnData::TimestampNanos { values, nulls })
256        }
257        ColumnData::Boolean { values, nulls } => Ok(ServerColumnData::Boolean { values, nulls }),
258        ColumnData::Dictionary {
259            ids,
260            dictionary,
261            nulls,
262        } => Ok(ServerColumnData::DictionaryText {
263            ids,
264            dictionary: dictionary.iter().map(ToString::to_string).collect(),
265            nulls,
266        }),
267        ColumnData::Bytes {
268            data,
269            offsets,
270            ext_type: radixdb_core::DataType::Bytes,
271            nulls,
272        } => Ok(ServerColumnData::Bytes {
273            data,
274            offsets,
275            nulls,
276        }),
277        ColumnData::Bytes {
278            data,
279            offsets,
280            ext_type: radixdb_core::DataType::Json,
281            nulls,
282        } => Ok(ServerColumnData::JsonText {
283            data,
284            offsets,
285            nulls,
286        }),
287        ColumnData::Bytes { ext_type, .. } => Err(Error::NotSupported(format!(
288            "typed column protocol does not yet support {ext_type}"
289        ))),
290        ColumnData::External {
291            data,
292            offsets,
293            type_ref,
294            nulls,
295        } => Ok(ServerColumnData::External {
296            data,
297            offsets,
298            type_ref,
299            nulls,
300        }),
301    }
302}
303
304/// Coarse transport-facing reason why a typed column batch is unavailable.
305#[derive(Clone, Copy, Debug, PartialEq, Eq)]
306pub enum ServerBatchFallback {
307    RowState,
308    QueryShape,
309    StorageShape,
310    Schema,
311}
312
313impl ServerBatchFallback {
314    pub(crate) fn from_storage(reason: TypedBatchFallbackReason) -> Self {
315        match reason {
316            TypedBatchFallbackReason::RowAlreadyFetched
317            | TypedBatchFallbackReason::RowIterationStarted
318            | TypedBatchFallbackReason::Closed
319            | TypedBatchFallbackReason::PendingError => Self::RowState,
320            TypedBatchFallbackReason::RowFilter
321            | TypedBatchFallbackReason::DictionaryFilter
322            | TypedBatchFallbackReason::IndexSelection
323            | TypedBatchFallbackReason::TypedPredicate
324            | TypedBatchFallbackReason::ExactTypedFilter
325            | TypedBatchFallbackReason::FilterCoveredByTypedPredicates
326            | TypedBatchFallbackReason::RowGroupSkips
327            | TypedBatchFallbackReason::EmptyProjection
328            | TypedBatchFallbackReason::DuplicateProjection
329            | TypedBatchFallbackReason::MergedSource
330            | TypedBatchFallbackReason::MixedTypedAndRowSources
331            | TypedBatchFallbackReason::UnsupportedResultShape => Self::QueryShape,
332            TypedBatchFallbackReason::NotArtifactBacked
333            | TypedBatchFallbackReason::UnsupportedStorageType
334            | TypedBatchFallbackReason::InvalidRange => Self::StorageShape,
335            TypedBatchFallbackReason::SchemaMappingMissingColumn
336            | TypedBatchFallbackReason::UnsupportedSchemaDefault => Self::Schema,
337        }
338    }
339}
340
341/// Instrumentation adapter kept outside the network state machine.
342pub struct ServerRuntimeMetrics;
343
344impl ServerRuntimeMetrics {
345    pub fn connection_opened() {
346        radixdb_storage::instrumentation::record_server_connection_opened();
347    }
348
349    pub fn connection_closed() {
350        radixdb_storage::instrumentation::record_server_connection_closed();
351    }
352
353    pub fn execution_started() {
354        radixdb_storage::instrumentation::record_server_execution_started();
355    }
356
357    pub fn execution_finished() {
358        radixdb_storage::instrumentation::record_server_execution_finished();
359    }
360
361    pub fn replace_session_owners(
362        before_sessions: u64,
363        after_sessions: u64,
364        before_cursors: u64,
365        after_cursors: u64,
366        before_prepared: u64,
367        after_prepared: u64,
368    ) {
369        radixdb_storage::instrumentation::replace_server_session_owners(
370            before_sessions,
371            after_sessions,
372            before_cursors,
373            after_cursors,
374            before_prepared,
375            after_prepared,
376        );
377    }
378
379    pub fn flush_thread_local() {
380        radixdb_storage::instrumentation::flush_thread_local_counters();
381    }
382
383    pub fn protocol_encode(bytes: u64, elapsed: Duration) {
384        radixdb_storage::instrumentation::record_protocol_encode(bytes, elapsed);
385    }
386
387    pub fn protocol_socket_write(bytes: u64, elapsed: Duration) {
388        radixdb_storage::instrumentation::record_protocol_socket_write(bytes, elapsed);
389    }
390
391    pub fn protocol_round_trip(elapsed: Duration) {
392        radixdb_storage::instrumentation::record_protocol_round_trip(elapsed);
393    }
394
395    pub fn protocol_row_adapter(values: u64) {
396        radixdb_storage::instrumentation::record_protocol_row_adapter(values);
397    }
398
399    pub fn protocol_result_rows(rows: u64) {
400        radixdb_storage::instrumentation::record_protocol_result_rows(rows);
401    }
402
403    pub fn protocol_row_batch(rows: u64) {
404        radixdb_storage::instrumentation::record_protocol_row_batch(rows);
405    }
406
407    pub fn protocol_column_batch_fallback(reason: ServerBatchFallback) {
408        let reason = match reason {
409            ServerBatchFallback::RowState => ProtocolColumnBatchFallback::RowState,
410            ServerBatchFallback::QueryShape => ProtocolColumnBatchFallback::QueryShape,
411            ServerBatchFallback::StorageShape => ProtocolColumnBatchFallback::StorageShape,
412            ServerBatchFallback::Schema => ProtocolColumnBatchFallback::Schema,
413        };
414        radixdb_storage::instrumentation::record_protocol_column_batch_fallback(reason);
415    }
416
417    pub fn column_batch_pending_opened(rows: u64, retained_bytes: u64) {
418        radixdb_storage::instrumentation::record_protocol_column_batch_pending_opened(
419            rows,
420            retained_bytes,
421        );
422    }
423
424    pub fn column_batch_pending_completed(rows: u64, retained_bytes: u64) {
425        radixdb_storage::instrumentation::record_protocol_column_batch_pending_completed(
426            rows,
427            retained_bytes,
428        );
429    }
430
431    pub fn column_batch_pending_dropped(rows: u64, retained_bytes: u64) {
432        radixdb_storage::instrumentation::record_protocol_column_batch_pending_dropped(
433            rows,
434            retained_bytes,
435        );
436    }
437}
438
439/// Storage tuning values admitted by server configuration without exposing
440/// the physical storage module to the network runtime.
441pub struct ServerStorageContract;
442
443impl ServerStorageContract {
444    pub const DEFAULT_COPY_MAX_TRANSACTION_BYTES: usize =
445        radixdb_storage::config::DEFAULT_COPY_MAX_TRANSACTION_BYTES;
446    pub const DEFAULT_MAX_COMPACTION_JOBS: usize =
447        radixdb_storage::config::DEFAULT_MAX_COMPACTION_JOBS;
448    pub const MAX_COMPACTION_JOBS: usize = radixdb_storage::config::MAX_COMPACTION_JOBS;
449    pub const DEFAULT_STORAGE_CPU_WORKERS: usize =
450        radixdb_storage::config::DEFAULT_STORAGE_CPU_WORKERS;
451    pub const DEFAULT_PAGE_CACHE_LEVEL: u8 = radixdb_storage::config::DEFAULT_PAGE_CACHE_LEVEL;
452    pub const MAX_PAGE_CACHE_LEVEL: u8 = radixdb_storage::config::MAX_PAGE_CACHE_LEVEL;
453    pub const DEFAULT_PAGE_CACHE_MAX_BYTES: u64 =
454        radixdb_storage::config::DEFAULT_PAGE_CACHE_MAX_BYTES;
455    pub const DEFAULT_PAGE_CACHE_MEMORY_RESERVE: u64 =
456        radixdb_storage::config::DEFAULT_PAGE_CACHE_MEMORY_RESERVE;
457}
458
459/// Credential primitives admitted by server configuration without exposing
460/// the executor implementation crate to the network runtime.
461pub struct ServerCredentialContract;
462
463impl ServerCredentialContract {
464    pub fn validate_password_verifier(encoded: &str) -> Result<()> {
465        radixdb_executor::credentials::validate_password_verifier(encoded)
466    }
467
468    pub fn verify_password_verifier(encoded: &str, password: &str) -> bool {
469        radixdb_executor::credentials::verify_password_verifier(encoded, password)
470    }
471
472    pub fn hash_password_verifier(password: &str) -> Result<String> {
473        radixdb_executor::credentials::hash_password_verifier(password)
474    }
475}
476
477/// Classify transaction-lifecycle SQL without granting the server parser/AST ownership.
478pub fn sql_contains_transaction_control(sql: &str) -> bool {
479    radixdb_executor::program_contains_transaction_control(sql)
480}
481
482#[cfg(test)]
483mod tests {
484    use super::*;
485
486    #[test]
487    fn mixed_typed_sources_are_a_query_shape_fallback() {
488        assert_eq!(
489            ServerBatchFallback::from_storage(TypedBatchFallbackReason::MixedTypedAndRowSources),
490            ServerBatchFallback::QueryShape
491        );
492    }
493}
radixdb_api/server_runtime.rs

radixdb_api/
server_runtime.rs
