polyc_query_model/
lib.rs

1//! `DataFusion`-free vocabulary for the versioned Query protocol.
2//!
3//! This crate contains only semantic requests and result frames.
4//!
5//! The REQUEST grants no scope. It names no realm, namespace, partition,
6//! table, source, or object. The Query service derives every one of those
7//! facts from the credential it verifies at its own boundary.
8//!
9//! The TERMINAL FRAME is the opposite by design. It carries State's complete
10//! source evidence, so `SourceSnapshot` transitively reaches State's
11//! manifest and object vocabulary: projection keys, source checkpoints, object
12//! descriptors, and namespaces. That evidence flows from the server to the
13//! caller and grants the caller nothing. Do not read the small re-export list
14//! below as a small type graph.
15
16use std::fmt;
17use std::time::Duration;
18
19pub mod evidence;
20
21pub use evidence::{
22    ATTESTATION_SIGNATURE_BYTES, ATTESTATION_SIGNER_BYTES, COMMIT_ROOT_BYTES, Classification,
23    DIGEST_BYTES, ExactObjectRef, INCARNATION_BYTES, JournalAnchor, JournalAttestation,
24    JournalSource, MAX_SOURCE_PINS, ObjectDescriptor, ProjectionKey, ProjectionManifest,
25    PublisherFence, Retention, SourceCheckpoint, SourceEvidence, SourcePin,
26};
27
28/// How a query ended, as the durable audit recorded it.
29#[derive(Debug, Clone, Copy, PartialEq, Eq)]
30pub enum QueryOutcome {
31    /// Every requested row was released.
32    Succeeded,
33    /// The query stopped, and this class says who is responsible.
34    Failed(ErrorClass),
35}
36
37/// Who is responsible for a query that did not succeed.
38#[derive(Debug, Clone, Copy, PartialEq, Eq)]
39pub enum ErrorClass {
40    /// Current authority refused this caller.
41    Denied,
42    /// The declared execution deadline expired.
43    Deadline,
44    /// The caller withdrew.
45    Cancelled,
46    /// A declared resource ceiling was reached.
47    Bounds,
48    /// A required source could not answer.
49    Unavailable,
50    /// The statement, its parameters, or the plan could not be used.
51    Malformed,
52    /// Corruption, or a fault in the serving plane or its deployment.
53    Internal,
54}
55
56/// Whether a result carries every matching row.
57#[derive(Debug, Clone, Copy, PartialEq, Eq)]
58pub enum Truncation {
59    /// Every matching row was released.
60    Complete,
61    /// Release stopped at this row count with rows still available.
62    TruncatedAt(u64),
63}
64
65/// The only protocol version this build speaks.
66pub const PROTOCOL_VERSION: u32 = 1;
67/// Largest UTF-8 SQL statement accepted by the semantic boundary.
68pub const MAX_SQL_BYTES: usize = 64 * 1024;
69/// Largest number of positional parameters in one request.
70pub const MAX_PARAMETERS: usize = 256;
71/// Largest total UTF-8 parameter payload in one request.
72pub const MAX_PARAMETER_BYTES: usize = 64 * 1024;
73/// Largest caller-requested execution duration.
74pub const MAX_TIMEOUT: Duration = Duration::from_mins(5);
75/// Largest caller-requested result row count.
76pub const MAX_ROWS: u64 = 1_000_000;
77/// Largest caller-requested released result size.
78pub const MAX_RESULT_BYTES: u64 = 64 * 1024 * 1024;
79/// Largest caller-requested result frame.
80pub const MAX_FRAME_BYTES: u64 = 4 * 1024 * 1024;
81
82/// One request parameter from the protocol's closed vocabulary.
83///
84/// `Debug` reports the kind only. A parameter value is caller content, so it
85/// never reaches a log line through this type.
86#[derive(Clone, PartialEq, Eq)]
87pub enum Parameter {
88    /// A UTF-8 string.
89    Utf8(String),
90    /// An unsigned 64-bit integer.
91    UInt64(u64),
92    /// A Boolean.
93    Boolean(bool),
94    /// An explicit SQL null.
95    Null,
96}
97
98impl fmt::Debug for Parameter {
99    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
100        let kind = match self {
101            Self::Utf8(_) => "utf8",
102            Self::UInt64(_) => "uint64",
103            Self::Boolean(_) => "boolean",
104            Self::Null => "null",
105        };
106        formatter.write_str(kind)
107    }
108}
109
110/// Projection consistency requested by a caller.
111#[derive(Debug, Clone, Copy, PartialEq, Eq)]
112pub enum Consistency {
113    /// Read the currently published projection.
114    Projected,
115    /// Refuse unless the projection covers at least this journal position.
116    ///
117    /// The position names no partition or source. A result may pin several
118    /// independent sources, so this scalar has no single referent across them.
119    /// The gate that recognizes this posture refuses it before any artifact
120    /// read; a later chunk owns both the target vector and its wait semantics.
121    RequireProjectedThrough(u64),
122}
123
124/// Caller-requested ceilings. A deployment may only narrow them.
125#[derive(Debug, Clone, Copy, PartialEq, Eq)]
126pub struct RequestedBounds {
127    timeout: Duration,
128    rows: u64,
129    result_bytes: u64,
130    frame_bytes: u64,
131}
132
133impl RequestedBounds {
134    /// Validates all four caller ceilings.
135    ///
136    /// # Errors
137    ///
138    /// Returns [`ModelError::Bounds`] when a ceiling is zero or exceeds the
139    /// protocol's compile-time maximum.
140    pub fn try_new(
141        timeout: Duration,
142        rows: u64,
143        result_bytes: u64,
144        frame_bytes: u64,
145    ) -> Result<Self, ModelError> {
146        if timeout.is_zero() || timeout > MAX_TIMEOUT {
147            return Err(ModelError::Bounds("timeout_nanos"));
148        }
149        if rows == 0 || rows > MAX_ROWS {
150            return Err(ModelError::Bounds("rows"));
151        }
152        if result_bytes == 0 || result_bytes > MAX_RESULT_BYTES {
153            return Err(ModelError::Bounds("result_bytes"));
154        }
155        if frame_bytes == 0 || frame_bytes > MAX_FRAME_BYTES {
156            return Err(ModelError::Bounds("frame_bytes"));
157        }
158        Ok(Self {
159            timeout,
160            rows,
161            result_bytes,
162            frame_bytes,
163        })
164    }
165
166    /// Returns the execution timeout ceiling.
167    #[must_use]
168    pub const fn timeout(self) -> Duration {
169        self.timeout
170    }
171
172    /// Returns the row ceiling.
173    #[must_use]
174    pub const fn rows(self) -> u64 {
175        self.rows
176    }
177
178    /// Returns the released result byte ceiling.
179    #[must_use]
180    pub const fn result_bytes(self) -> u64 {
181        self.result_bytes
182    }
183
184    /// Returns the per-frame byte ceiling.
185    #[must_use]
186    pub const fn frame_bytes(self) -> u64 {
187        self.frame_bytes
188    }
189}
190
191/// A version-independent semantic Query request.
192///
193/// `Debug` reports the statement's byte length and the parameter count. The
194/// statement text and every parameter value are caller content, so neither
195/// reaches a log line through this type.
196#[derive(Clone, PartialEq, Eq)]
197pub struct QueryRequest {
198    sql: String,
199    parameters: Vec<Parameter>,
200    consistency: Consistency,
201    bounds: RequestedBounds,
202}
203
204impl QueryRequest {
205    /// Validates a complete request.
206    ///
207    /// # Errors
208    ///
209    /// Returns a typed refusal for empty or over-bound SQL and parameters.
210    pub fn try_new(
211        sql: String,
212        parameters: Vec<Parameter>,
213        consistency: Consistency,
214        bounds: RequestedBounds,
215    ) -> Result<Self, ModelError> {
216        if sql.is_empty() || sql.len() > MAX_SQL_BYTES {
217            return Err(ModelError::Bounds("sql"));
218        }
219        if parameters.len() > MAX_PARAMETERS {
220            return Err(ModelError::Bounds("parameters"));
221        }
222        let parameter_bytes = parameters.iter().try_fold(0_usize, |total, parameter| {
223            let bytes = match parameter {
224                Parameter::Utf8(value) => value.len(),
225                Parameter::UInt64(_) | Parameter::Boolean(_) | Parameter::Null => 0,
226            };
227            total
228                .checked_add(bytes)
229                .ok_or(ModelError::Bounds("parameters"))
230        })?;
231        if parameter_bytes > MAX_PARAMETER_BYTES {
232            return Err(ModelError::Bounds("parameters"));
233        }
234        Ok(Self {
235            sql,
236            parameters,
237            consistency,
238            bounds,
239        })
240    }
241
242    /// Returns the SQL text.
243    #[must_use]
244    pub fn sql(&self) -> &str {
245        &self.sql
246    }
247
248    /// Returns positional parameters.
249    #[must_use]
250    pub fn parameters(&self) -> &[Parameter] {
251        &self.parameters
252    }
253
254    /// Returns requested consistency.
255    #[must_use]
256    pub const fn consistency(&self) -> Consistency {
257        self.consistency
258    }
259
260    /// Returns caller-requested bounds.
261    #[must_use]
262    pub const fn bounds(&self) -> RequestedBounds {
263        self.bounds
264    }
265}
266
267impl fmt::Debug for QueryRequest {
268    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
269        formatter
270            .debug_struct("QueryRequest")
271            .field("sql_bytes", &self.sql.len())
272            .field("parameters", &self.parameters.len())
273            .field("consistency", &self.consistency)
274            .field("bounds", &self.bounds)
275            .finish()
276    }
277}
278
279/// One Arrow IPC schema frame.
280///
281/// `Debug` reports the encoded length only. The bytes describe result columns.
282#[derive(Clone, PartialEq, Eq)]
283pub struct SchemaFrame {
284    arrow_ipc: Vec<u8>,
285}
286
287impl SchemaFrame {
288    /// Validates one opaque Arrow IPC schema frame.
289    ///
290    /// # Errors
291    ///
292    /// Returns [`ModelError::Bounds`] for an empty or over-bound frame.
293    pub fn try_new(arrow_ipc: Vec<u8>) -> Result<Self, ModelError> {
294        validate_frame_bytes(&arrow_ipc)?;
295        Ok(Self { arrow_ipc })
296    }
297
298    /// Returns the opaque Arrow IPC schema bytes.
299    #[must_use]
300    pub fn arrow_ipc(&self) -> &[u8] {
301        &self.arrow_ipc
302    }
303}
304
305impl fmt::Debug for SchemaFrame {
306    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
307        formatter
308            .debug_struct("SchemaFrame")
309            .field("arrow_ipc_bytes", &self.arrow_ipc.len())
310            .finish()
311    }
312}
313
314/// One ordered Arrow IPC data frame.
315///
316/// `Debug` reports the sequence, row count, and encoded length. The bytes are
317/// released rows, so they never reach a log line through this type.
318#[derive(Clone, PartialEq, Eq)]
319pub struct DataFrame {
320    sequence: u64,
321    rows: u64,
322    arrow_ipc: Vec<u8>,
323}
324
325impl DataFrame {
326    /// Validates one ordered opaque Arrow IPC data frame.
327    ///
328    /// # Errors
329    ///
330    /// Returns [`ModelError::Bounds`] for an empty or over-bound frame.
331    pub fn try_new(sequence: u64, rows: u64, arrow_ipc: Vec<u8>) -> Result<Self, ModelError> {
332        validate_frame_bytes(&arrow_ipc)?;
333        Ok(Self {
334            sequence,
335            rows,
336            arrow_ipc,
337        })
338    }
339
340    /// Returns the zero-based frame sequence.
341    #[must_use]
342    pub const fn sequence(&self) -> u64 {
343        self.sequence
344    }
345
346    /// Returns the number of rows encoded in the frame.
347    #[must_use]
348    pub const fn rows(&self) -> u64 {
349        self.rows
350    }
351
352    /// Returns opaque Arrow IPC record-batch bytes.
353    #[must_use]
354    pub fn arrow_ipc(&self) -> &[u8] {
355        &self.arrow_ipc
356    }
357}
358
359impl fmt::Debug for DataFrame {
360    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
361        formatter
362            .debug_struct("DataFrame")
363            .field("sequence", &self.sequence)
364            .field("rows", &self.rows)
365            .field("arrow_ipc_bytes", &self.arrow_ipc.len())
366            .finish()
367    }
368}
369
370/// Final audited result facts.
371///
372/// `Debug` reports counts and outcome only. The source evidence names object
373/// keys, namespaces, partitions, signer keys, and signatures, so it never
374/// reaches a log line through this type.
375#[derive(Clone, PartialEq, Eq)]
376pub struct TerminalFrame {
377    outcome: QueryOutcome,
378    duration: Duration,
379    rows: u64,
380    result_bytes: u64,
381    truncation: Truncation,
382    source: SourceEvidence,
383}
384
385impl TerminalFrame {
386    /// Records the exact durable completion facts returned by Query.
387    #[must_use]
388    pub const fn new(
389        outcome: QueryOutcome,
390        duration: Duration,
391        rows: u64,
392        result_bytes: u64,
393        truncation: Truncation,
394        source: SourceEvidence,
395    ) -> Self {
396        Self {
397            outcome,
398            duration,
399            rows,
400            result_bytes,
401            truncation,
402            source,
403        }
404    }
405
406    /// Returns the durable completion outcome.
407    #[must_use]
408    pub const fn outcome(&self) -> QueryOutcome {
409        self.outcome
410    }
411
412    /// Returns measured execution duration.
413    #[must_use]
414    pub const fn duration(&self) -> Duration {
415        self.duration
416    }
417
418    /// Returns total rows released across data frames.
419    #[must_use]
420    pub const fn rows(&self) -> u64 {
421        self.rows
422    }
423
424    /// Returns total opaque data bytes released across data frames.
425    #[must_use]
426    pub const fn result_bytes(&self) -> u64 {
427        self.result_bytes
428    }
429
430    /// Returns whether the result was complete or truncated.
431    #[must_use]
432    pub const fn truncation(&self) -> Truncation {
433        self.truncation
434    }
435
436    /// Returns exact canonical source premises recorded by Query audit.
437    #[must_use]
438    pub const fn source(&self) -> &SourceEvidence {
439        &self.source
440    }
441}
442
443impl fmt::Debug for TerminalFrame {
444    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
445        formatter
446            .debug_struct("TerminalFrame")
447            .field("outcome", &self.outcome)
448            .field("duration", &self.duration)
449            .field("rows", &self.rows)
450            .field("result_bytes", &self.result_bytes)
451            .field("truncation", &self.truncation)
452            .field("source_pins", &self.source.pins().len())
453            .finish()
454    }
455}
456
457fn validate_frame_bytes(bytes: &[u8]) -> Result<(), ModelError> {
458    if bytes.is_empty() || u64::try_from(bytes.len()).unwrap_or(u64::MAX) > MAX_FRAME_BYTES {
459        return Err(ModelError::Bounds("arrow_ipc"));
460    }
461    Ok(())
462}
463
464/// One version-independent result stream frame.
465///
466/// Every variant's own `Debug` reports shape only, so this derive carries no
467/// statement, parameter, row byte, or source identifier.
468#[derive(Debug, Clone, PartialEq, Eq)]
469pub enum ResultFrame {
470    /// The required first frame.
471    Schema(SchemaFrame),
472    /// An ordered data frame.
473    Data(DataFrame),
474    /// The required final frame.
475    Terminal(TerminalFrame),
476}
477
478/// Semantic request validation failure.
479#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
480pub enum ModelError {
481    /// One field is empty or outside its closed bound.
482    #[error("query field `{0}` is outside its protocol bound")]
483    Bounds(&'static str),
484    /// A canonically ordered vector is unsorted or names one source twice.
485    #[error("query field `{0}` is not in strict canonical order")]
486    Order(&'static str),
487}
488
489#[cfg(test)]
490mod tests {
491    use super::*;
492
493    fn bounds() -> RequestedBounds {
494        RequestedBounds::try_new(Duration::from_secs(1), 10, 1024, 512).unwrap()
495    }
496
497    #[test]
498    fn request_refuses_empty_and_overbound_content() {
499        assert_eq!(
500            QueryRequest::try_new(String::new(), vec![], Consistency::Projected, bounds()),
501            Err(ModelError::Bounds("sql"))
502        );
503        assert_eq!(
504            QueryRequest::try_new(
505                "select ?".into(),
506                vec![Parameter::Utf8("x".repeat(MAX_PARAMETER_BYTES + 1))],
507                Consistency::Projected,
508                bounds(),
509            ),
510            Err(ModelError::Bounds("parameters"))
511        );
512    }
513
514    #[test]
515    fn bounds_refuse_zero_and_crossed_frame_limits() {
516        assert_eq!(
517            RequestedBounds::try_new(Duration::ZERO, 1, 1, 1),
518            Err(ModelError::Bounds("timeout_nanos"))
519        );
520        assert_eq!(
521            RequestedBounds::try_new(Duration::from_secs(1), 1, 8, MAX_FRAME_BYTES + 1),
522            Err(ModelError::Bounds("frame_bytes"))
523        );
524    }
525
526    #[test]
527    fn debug_output_carries_no_caller_content() {
528        let request = QueryRequest::try_new(
529            "select secret_column from messages".into(),
530            vec![Parameter::Utf8("tenant-secret".into())],
531            Consistency::Projected,
532            bounds(),
533        )
534        .unwrap();
535        let rendered = format!("{request:?}");
536        assert!(!rendered.contains("secret_column"), "{rendered}");
537        assert!(!rendered.contains("tenant-secret"), "{rendered}");
538        assert!(rendered.contains("sql_bytes"), "{rendered}");
539
540        let data = DataFrame::try_new(0, 2, vec![7, 8, 9]).unwrap();
541        let rendered = format!("{data:?}");
542        assert!(!rendered.contains('7'), "{rendered}");
543        assert!(rendered.contains("arrow_ipc_bytes: 3"), "{rendered}");
544    }
545}
polyc_query_model/lib.rs

polyc_query_model/
lib.rs
