mongreldb_protocol/
services.rs

1//! Protocol service definitions (spec section 10.4, S1D-003).
2//!
3//! The seven services below are the server's contract with its protocol
4//! adapters (native RPC, HTTP/JSON, Kit, MySQL wire): adapters translate
5//! wire requests into the canonical model of [`crate::request`] and call
6//! these traits. The server also implements the generated gRPC services in
7//! `mongreldb_server::native`.
8//!
9//! # Errors
10//!
11//! Every method returns [`CategoryError`]
12//! ([`mongreldb_types::errors::CategoryError`]), the structural taxonomy of
13//! spec section 9.7 that every language binding maps. Programmatic handling
14//! keys off the category (or its stable code), never the message.
15//!
16//! # Async shape: object-safe boxed futures
17//!
18//! The traits use hand-written boxed futures ([`BoxFuture`]) instead of
19//! native `async fn` in traits. Native async-fn-in-trait (stable, and usable
20//! on this workspace's Rust 1.88) desugars to RPITIT, which is NOT
21//! object-safe: `Arc<dyn QueryService>` would be impossible, forcing every
22//! adapter to monomorphize around concrete service types. The adapters
23//! dispatch services dynamically, so object safety is required — and this
24//! crate's dependency set is frozen, so the `async-trait` crate is not
25//! available to bridge the gap. The cost is one heap allocation per call,
26//! acceptable at the protocol boundary where a call is already a network
27//! request. The same choice gives an object-safe [`ArrowFrameStream`]
28//! without a `futures` dependency.
29
30use core::pin::Pin;
31use std::future::Future;
32
33use mongreldb_types::errors::CategoryError;
34use mongreldb_types::ids::{DatabaseId, QueryId, SchemaVersion, TransactionId};
35
36use crate::prepared::PreparedStatementBinding;
37use crate::request::{AuthenticatedIdentity, ExecuteRequest, IsolationLevel, SessionId};
38use crate::session::Session;
39
40/// The boxed future every service method returns: object-safe, `Send`, and
41/// resolving to `Result<T, CategoryError>`. See the module-level
42/// documentation for why this is not native `async fn` in traits.
43pub type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = Result<T, CategoryError>> + Send + 'a>>;
44
45/// Credentials presented at session open.
46///
47/// Variants are never reordered and discriminants never reused (spec
48/// section 4.10); new credential kinds are only appended.
49#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
50pub enum Credentials {
51    /// Catalog username + password, verified against the catalog's Argon2id
52    /// password hashes.
53    Password {
54        /// Case-sensitive username.
55        username: String,
56        /// Cleartext password, verified and immediately discarded; the wire
57        /// form is protected by TLS 1.3 (S1D-002).
58        password: String,
59    },
60}
61
62/// The buffered result of a non-streaming [`QueryService::execute`].
63#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
64pub struct ExecuteResponse {
65    /// The query execution that produced this response.
66    pub query_id: QueryId,
67    /// Rows affected (for DML); zero for queries.
68    pub rows_affected: u64,
69    /// Result frames: Arrow IPC byte chunks, the buffered form of the
70    /// [`ArrowFrameStream`] contract. Empty for commands without a result
71    /// set. The protocol crate fixes framing; the server encodes real Arrow
72    /// record batches into these frames.
73    pub frames: Vec<Vec<u8>>,
74}
75
76/// The execution phase of a query (S1D-006).
77///
78/// Variants are never reordered and discriminants never reused (spec
79/// section 4.10).
80#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
81pub enum QueryPhase {
82    /// Waiting for admission (queue wait counts toward the deadline).
83    Queued,
84    /// Parsing and planning.
85    Planning,
86    /// Executing.
87    Executing,
88    /// Serializing result frames (counts toward the deadline).
89    Serializing,
90    /// Finished successfully (durable outcome).
91    Completed,
92    /// Finished with a failure (durable outcome); see
93    /// [`QueryStatus::error`].
94    Failed,
95    /// Cancelled by the caller or by deadline expiry (durable outcome).
96    Cancelled,
97}
98
99/// The status of one query execution, as returned by
100/// [`QueryService::get_query_status`].
101#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
102pub struct QueryStatus {
103    /// The query this status describes.
104    pub query_id: QueryId,
105    /// Current (or final) phase.
106    pub phase: QueryPhase,
107    /// The failure that ended the query, present iff
108    /// [`QueryPhase::Failed`].
109    pub error: Option<CategoryError>,
110}
111
112/// One column of a [`TableSchema`].
113#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
114pub struct ColumnSchema {
115    /// Column name.
116    pub name: String,
117    /// Canonical type name (e.g. `INT64`, `TEXT`).
118    pub data_type: String,
119    /// Whether the column accepts `NULL`.
120    pub nullable: bool,
121}
122
123/// The schema of one table, as returned by
124/// [`CatalogService::get_schema`].
125#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
126pub struct TableSchema {
127    /// Table name.
128    pub table: String,
129    /// Current schema version of the table; clients pin prepared statements
130    /// to it (S1D-005).
131    pub schema_version: SchemaVersion,
132    /// Columns in declaration order.
133    pub columns: Vec<ColumnSchema>,
134}
135
136/// The serving state reported by [`HealthService::status`].
137#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
138pub struct HealthStatus {
139    /// Whether the server is accepting requests.
140    pub serving: bool,
141    /// Optional human-readable detail (e.g. why the server is not serving).
142    pub detail: Option<String>,
143}
144
145/// A pull stream of Arrow IPC byte frames, as returned by
146/// [`QueryService::execute_stream`].
147///
148/// Each frame is one Arrow IPC byte chunk carried on the wire as the
149/// payload of a [`crate::envelope::ProtocolEnvelope`] (spec section 4.10);
150/// the protocol crate fixes the framing contract; the server supplies real
151/// Arrow record-batch encoding. `Ok(Some(frame))` yields
152/// one chunk, `Ok(None)` ends the stream, and `Err` is terminal (the stream
153/// yields nothing further).
154pub trait ArrowFrameStream: Send {
155    /// Pulls the next frame; see the trait documentation for the contract.
156    fn next_frame(&mut self) -> BoxFuture<'_, Option<Vec<u8>>>;
157}
158
159impl ArrowFrameStream for std::vec::IntoIter<Vec<u8>> {
160    fn next_frame(&mut self) -> BoxFuture<'_, Option<Vec<u8>>> {
161        let frame = self.next();
162        Box::pin(async move { Ok(frame) })
163    }
164}
165
166/// Authentication (S1D-003): turns credentials into an
167/// [`AuthenticatedIdentity`]. Failures fail closed as
168/// [`mongreldb_types::errors::ErrorCategory::Unauthenticated`].
169pub trait AuthService: Send + Sync {
170    /// Verifies credentials and returns the authenticated identity.
171    fn authenticate<'a>(
172        &'a self,
173        credentials: &'a Credentials,
174    ) -> BoxFuture<'a, AuthenticatedIdentity>;
175}
176
177/// Session lifecycle (S1D-003, S1D-004).
178pub trait SessionService: Send + Sync {
179    /// Opens a session for an authenticated principal on a database.
180    fn open_session(
181        &self,
182        principal: AuthenticatedIdentity,
183        database_id: DatabaseId,
184    ) -> BoxFuture<'_, Session>;
185
186    /// Closes a session, rolling back any active transaction and dropping
187    /// its prepared statements.
188    fn close_session(&self, session_id: SessionId) -> BoxFuture<'_, ()>;
189}
190
191/// Query preparation, execution, streaming, and cancellation (S1D-003).
192pub trait QueryService: Send + Sync {
193    /// Prepares a SQL statement on a session, returning the binding record
194    /// (S1D-005) the executor validates before every execution.
195    fn prepare(
196        &self,
197        session_id: SessionId,
198        sql: String,
199    ) -> BoxFuture<'_, PreparedStatementBinding>;
200
201    /// Executes a canonical request, buffering the result.
202    fn execute(&self, request: ExecuteRequest) -> BoxFuture<'_, ExecuteResponse>;
203
204    /// Executes a canonical request, streaming the result as Arrow IPC byte
205    /// frames (see [`ArrowFrameStream`]).
206    fn execute_stream(&self, request: ExecuteRequest) -> BoxFuture<'_, Box<dyn ArrowFrameStream>>;
207
208    /// Cancels a running query (S1D-006). Cancelling an unknown query fails;
209    /// a finished query keeps its durable outcome (spec section 4.7), which
210    /// [`QueryService::get_query_status`] reports.
211    fn cancel_query(&self, query_id: QueryId) -> BoxFuture<'_, ()>;
212
213    /// Returns the current status of a query, including its durable outcome
214    /// once finished.
215    fn get_query_status(&self, query_id: QueryId) -> BoxFuture<'_, QueryStatus>;
216}
217
218/// Explicit transaction control on a session (S1D-003).
219pub trait TransactionService: Send + Sync {
220    /// Begins a transaction at the requested isolation level.
221    fn begin(
222        &self,
223        session_id: SessionId,
224        isolation: IsolationLevel,
225    ) -> BoxFuture<'_, TransactionId>;
226
227    /// Commits the session's active transaction. Commit failures carry the
228    /// transaction categories of the taxonomy (e.g.
229    /// [`mongreldb_types::errors::ErrorCategory::TransactionConflict`],
230    /// [`mongreldb_types::errors::ErrorCategory::CommitOutcomeUnknown`]); an
231    /// ambiguous outcome is only replayed with a durable idempotency key
232    /// (spec section 11.7).
233    fn commit(&self, session_id: SessionId) -> BoxFuture<'_, ()>;
234
235    /// Rolls back the session's active transaction.
236    fn rollback(&self, session_id: SessionId) -> BoxFuture<'_, ()>;
237}
238
239/// Catalog reads (S1D-003).
240pub trait CatalogService: Send + Sync {
241    /// Returns the current schema of one table.
242    fn get_schema(&self, database_id: DatabaseId, table: String) -> BoxFuture<'_, TableSchema>;
243}
244
245/// Administrative operations (S1D-003). The request's command must be
246/// [`crate::request::ExecuteCommand::Admin`]; non-admin principals fail as
247/// [`mongreldb_types::errors::ErrorCategory::PermissionDenied`].
248pub trait AdminService: Send + Sync {
249    /// Executes the admin command of a canonical request.
250    fn execute_admin(&self, request: ExecuteRequest) -> BoxFuture<'_, ()>;
251}
252
253/// Liveness and readiness (S1D-003).
254pub trait HealthService: Send + Sync {
255    /// Returns the current serving state.
256    fn status(&self) -> BoxFuture<'_, HealthStatus>;
257}
258
259#[cfg(test)]
260mod tests {
261    use super::*;
262    use crate::request::{ExecuteCommand, ResultLimits};
263    use crate::test_support::{assert_serde_round_trip, block_on};
264    use mongreldb_types::errors::ErrorCategory;
265    use std::sync::Arc;
266
267    #[test]
268    fn dto_serde_round_trips() {
269        assert_serde_round_trip(&Credentials::Password {
270            username: "alice".to_owned(),
271            password: "s3cret".to_owned(),
272        });
273        assert_serde_round_trip(&ExecuteResponse {
274            query_id: QueryId::new_random(),
275            rows_affected: 17,
276            frames: vec![b"arrow-ipc-bytes".to_vec(), vec![]],
277        });
278        for phase in [
279            QueryPhase::Queued,
280            QueryPhase::Planning,
281            QueryPhase::Executing,
282            QueryPhase::Serializing,
283            QueryPhase::Completed,
284            QueryPhase::Failed,
285            QueryPhase::Cancelled,
286        ] {
287            assert_serde_round_trip(&phase);
288        }
289        assert_serde_round_trip(&QueryStatus {
290            query_id: QueryId::new_random(),
291            phase: QueryPhase::Failed,
292            error: Some(CategoryError::new(
293                ErrorCategory::DeadlineExceeded,
294                "deadline expired during execution",
295            )),
296        });
297        assert_serde_round_trip(&QueryStatus {
298            query_id: QueryId::new_random(),
299            phase: QueryPhase::Completed,
300            error: None,
301        });
302        assert_serde_round_trip(&ColumnSchema {
303            name: "tenant".to_owned(),
304            data_type: "INT64".to_owned(),
305            nullable: false,
306        });
307        assert_serde_round_trip(&TableSchema {
308            table: "events".to_owned(),
309            schema_version: SchemaVersion::new(10),
310            columns: vec![
311                ColumnSchema {
312                    name: "tenant".to_owned(),
313                    data_type: "INT64".to_owned(),
314                    nullable: false,
315                },
316                ColumnSchema {
317                    name: "payload".to_owned(),
318                    data_type: "TEXT".to_owned(),
319                    nullable: true,
320                },
321            ],
322        });
323        assert_serde_round_trip(&HealthStatus {
324            serving: true,
325            detail: None,
326        });
327        assert_serde_round_trip(&HealthStatus {
328            serving: false,
329            detail: Some("draining".to_owned()),
330        });
331    }
332
333    /// A stub auth service that always fails closed.
334    struct StubAuth;
335
336    impl AuthService for StubAuth {
337        fn authenticate<'a>(
338            &'a self,
339            credentials: &'a Credentials,
340        ) -> BoxFuture<'a, AuthenticatedIdentity> {
341            Box::pin(async move {
342                let Credentials::Password { username, .. } = credentials;
343                Err(CategoryError::new(
344                    ErrorCategory::Unauthenticated,
345                    format!("invalid credentials for {username:?}"),
346                ))
347            })
348        }
349    }
350
351    #[test]
352    fn category_error_propagates_through_dyn_dispatch() {
353        let service: Arc<dyn AuthService> = Arc::new(StubAuth);
354        let credentials = Credentials::Password {
355            username: "alice".to_owned(),
356            password: "wrong".to_owned(),
357        };
358        let error = block_on(service.authenticate(&credentials)).unwrap_err();
359        // The structural shape survives dynamic dispatch: category, stable
360        // code, and message.
361        assert_eq!(error.category, ErrorCategory::Unauthenticated);
362        assert_eq!(error.code(), 19);
363        assert_eq!(error.message, "invalid credentials for \"alice\"");
364        assert!(!error.category.is_retryable());
365        assert_eq!(
366            error.to_string(),
367            "unauthenticated: invalid credentials for \"alice\""
368        );
369    }
370
371    struct StubQuery;
372
373    impl QueryService for StubQuery {
374        fn prepare(
375            &self,
376            _session_id: SessionId,
377            sql: String,
378        ) -> BoxFuture<'_, PreparedStatementBinding> {
379            Box::pin(async move {
380                Err(CategoryError::new(
381                    ErrorCategory::SchemaVersionMismatch,
382                    format!("cannot prepare {sql:?}: stale catalog"),
383                ))
384            })
385        }
386
387        fn execute(&self, request: ExecuteRequest) -> BoxFuture<'_, ExecuteResponse> {
388            Box::pin(async move {
389                Ok(ExecuteResponse {
390                    query_id: request.query_id,
391                    rows_affected: 0,
392                    frames: vec![b"frame-1".to_vec(), b"frame-2".to_vec()],
393                })
394            })
395        }
396
397        fn execute_stream(
398            &self,
399            _request: ExecuteRequest,
400        ) -> BoxFuture<'_, Box<dyn ArrowFrameStream>> {
401            Box::pin(async move {
402                let stream: Box<dyn ArrowFrameStream> =
403                    Box::new(vec![b"frame-1".to_vec(), b"frame-2".to_vec()].into_iter());
404                Ok(stream)
405            })
406        }
407
408        fn cancel_query(&self, query_id: QueryId) -> BoxFuture<'_, ()> {
409            Box::pin(async move {
410                Err(CategoryError::new(
411                    ErrorCategory::Cancelled,
412                    format!("query {query_id} is not running"),
413                ))
414            })
415        }
416
417        fn get_query_status(&self, query_id: QueryId) -> BoxFuture<'_, QueryStatus> {
418            Box::pin(async move {
419                Ok(QueryStatus {
420                    query_id,
421                    phase: QueryPhase::Completed,
422                    error: None,
423                })
424            })
425        }
426    }
427
428    fn sample_request() -> ExecuteRequest {
429        ExecuteRequest {
430            request_id: [0x99; 16],
431            query_id: QueryId::new_random(),
432            session_id: Some(SessionId::from_bytes([0x88; 16])),
433            database_id: DatabaseId::new_random(),
434            principal: AuthenticatedIdentity::Credentialless,
435            command: ExecuteCommand::Sql {
436                text: "SELECT 1".to_owned(),
437                params: vec![],
438            },
439            deadline_unix_micros: None,
440            result_limits: ResultLimits::default(),
441            resource_group: None,
442            idempotency_key: None,
443        }
444    }
445
446    #[test]
447    fn query_service_methods_work_through_dyn_dispatch() {
448        let service: Arc<dyn QueryService> = Arc::new(StubQuery);
449        let request = sample_request();
450
451        let response = block_on(service.execute(request.clone())).unwrap();
452        assert_eq!(response.query_id, request.query_id);
453        assert_eq!(response.frames.len(), 2);
454
455        let mut stream = block_on(service.execute_stream(request)).unwrap();
456        assert_eq!(
457            block_on(stream.next_frame()).unwrap(),
458            Some(b"frame-1".to_vec())
459        );
460        assert_eq!(
461            block_on(stream.next_frame()).unwrap(),
462            Some(b"frame-2".to_vec())
463        );
464        assert_eq!(block_on(stream.next_frame()).unwrap(), None);
465
466        let status = block_on(service.get_query_status(response.query_id)).unwrap();
467        assert_eq!(status.phase, QueryPhase::Completed);
468
469        let error = block_on(service.cancel_query(response.query_id)).unwrap_err();
470        assert_eq!(error.category, ErrorCategory::Cancelled);
471
472        let error = block_on(service.prepare(SessionId::ZERO, "SELECT 1".to_owned())).unwrap_err();
473        assert_eq!(error.category, ErrorCategory::SchemaVersionMismatch);
474    }
475
476    #[test]
477    fn vec_into_iter_stream_yields_all_frames_then_ends() {
478        let mut stream: Box<dyn ArrowFrameStream> = Box::new(Vec::<Vec<u8>>::new().into_iter());
479        assert_eq!(block_on(stream.next_frame()).unwrap(), None);
480    }
481}
mongreldb_protocol/services.rs

mongreldb_protocol/
services.rs
