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

mongreldb_protocol/
services.rs
