rustev_contract/
acquisition.rs

1//! Acquisition wire documents and pure page validation (spec 036, 3.2 and 3.3).
2use serde::{Deserialize, Deserializer, Serialize};
3use serde_json::Value;
4use std::collections::BTreeSet;
5
6use crate::definition::{ProvenanceClass, TypeDecl};
7use crate::grade::Grade;
8use crate::ids::{AcquisitionRequestId, SourceId};
9use crate::source::{Reliability, valid_source_name};
10use crate::{Decimal, Document, DurationMs, Identified, Timestamp, schema};
11
12#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
13#[serde(deny_unknown_fields)]
14pub struct Operation {
15    #[serde(deserialize_with = "name")]
16    pub name: String,
17    pub cancellation: Cancellation,
18    #[serde(deserialize_with = "request_fields")]
19    #[schemars(length(max = 32))]
20    pub request: Vec<RequestField>,
21    #[serde(deserialize_with = "bound::<_, 1, 65536>")]
22    #[schemars(range(min = 1, max = 65536))]
23    pub max_request_bytes: u64,
24    #[serde(deserialize_with = "item_fields")]
25    #[schemars(length(min = 1, max = 64))]
26    pub item: Vec<ItemField>,
27    #[serde(deserialize_with = "bound::<_, 1, 10000>")]
28    #[schemars(range(min = 1, max = 10000))]
29    pub max_page_items: u64,
30    #[serde(deserialize_with = "bound::<_, 1, 4194304>")]
31    #[schemars(range(min = 1, max = 4194304))]
32    pub max_page_bytes: u64,
33    pub paging: Paging,
34    pub cost: OperationCost,
35    pub idempotent: bool,
36    pub cache: Cache,
37}
38
39#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
40#[serde(rename_all = "snake_case", deny_unknown_fields)]
41pub enum Cancellation {
42    Acknowledges,
43    None,
44}
45
46#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
47#[serde(deny_unknown_fields)]
48pub struct RequestField {
49    pub name: String,
50    #[serde(rename = "type", deserialize_with = "request_type")]
51    pub ty: TypeDecl,
52    pub required: bool,
53}
54
55#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
56#[serde(deny_unknown_fields)]
57pub struct ItemField {
58    pub name: String,
59    #[serde(rename = "type", deserialize_with = "item_type")]
60    pub ty: TypeDecl,
61    pub provenance: ProvenanceClass,
62    pub max_grade: Grade,
63    #[serde(deserialize_with = "reliability")]
64    pub reliability: Reliability,
65    pub freshness: Freshness,
66}
67
68impl ItemField {
69    /// As for fact declarations, unreliability lowers the effective ceiling.
70    pub fn limit(&self) -> Grade {
71        match self.reliability {
72            Reliability::Reliable => self.max_grade,
73            Reliability::KnownUnreliable { .. } => Grade::Hint,
74        }
75    }
76}
77
78#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
79#[serde(rename_all = "snake_case", deny_unknown_fields)]
80pub enum Freshness {
81    MaxAgeMs { value: u64 },
82    NotRequired,
83}
84
85#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
86#[serde(rename_all = "snake_case", deny_unknown_fields)]
87pub enum Paging {
88    None,
89    Cursor {
90        #[serde(deserialize_with = "bound::<_, 1, 1024>")]
91        #[schemars(range(min = 1, max = 1024))]
92        max_cursor_bytes: u64,
93    },
94    Offset,
95}
96
97/// Static acquisition disclosure. Existing `CostModel::Bounded` carries no
98/// maximum, while `CostBound::Estimated` requires units, so neither has
99/// this document's wire shape.
100#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
101#[serde(rename_all = "snake_case", deny_unknown_fields)]
102pub enum OperationCost {
103    Bounded { max_units: u64 },
104    Estimated,
105    Unknown,
106}
107
108#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
109#[serde(rename_all = "snake_case", deny_unknown_fields)]
110pub enum Cache {
111    NotCacheable,
112    Cacheable {
113        #[serde(deserialize_with = "bound::<_, 1, 86400000>")]
114        #[schemars(range(min = 1, max = 86400000))]
115        max_ttl_ms: u64,
116    },
117}
118
119#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
120#[serde(deny_unknown_fields)]
121pub struct AcquisitionRequest {
122    pub schema: String,
123    pub source: SourceId,
124    #[serde(deserialize_with = "name")]
125    pub operation: String,
126    #[serde(deserialize_with = "list::<_, _, 0, 32>")]
127    #[schemars(length(max = 32))]
128    pub fields: Vec<RequestValue>,
129    pub page: RequestPage,
130}
131crate::document::document!(
132    AcquisitionRequest,
133    schema::ACQUISITION_REQUEST,
134    crate::limits::DESCRIPTOR_V1,
135    AcquisitionRequestId
136);
137
138#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
139#[serde(rename_all = "snake_case", deny_unknown_fields)]
140pub enum RequestValue {
141    Set { value: Value },
142    Unset,
143}
144
145#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
146#[serde(rename_all = "snake_case", deny_unknown_fields)]
147pub enum RequestPage {
148    First,
149    Cursor { text: String },
150    Offset { integer: u64 },
151}
152
153#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
154#[serde(deny_unknown_fields)]
155pub struct AcquisitionPage {
156    pub schema: String,
157    pub request: AcquisitionRequestId,
158    pub as_of_ms: Timestamp,
159    pub items: Vec<serde_json::Map<String, Value>>,
160    pub next: NextPage,
161}
162crate::document::document!(
163    AcquisitionPage,
164    schema::ACQUISITION_PAGE,
165    crate::limits::SNAPSHOT_V1
166);
167
168#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
169#[serde(rename_all = "snake_case", deny_unknown_fields)]
170pub enum NextPage {
171    None,
172    Cursor { text: String },
173    Offset { integer: u64 },
174}
175
176/// The failed rule of spec 036, 3.3.4 (plus its per-operation byte bounds).
177#[derive(Debug, Clone, PartialEq, Eq)]
178pub enum PageError {
179    RequestIdentity,
180    ItemCount,
181    Item { index: usize },
182    NextKind,
183    Offset,
184    CursorLength,
185    RequestBytes,
186    PageBytes,
187    AsOfSkew,
188}
189
190/// Validate a decoded page without I/O or a clock. Bytes are canonical wire
191/// bytes; callers still bound transport buffers and use `Document::parse`.
192/// `next: none` exhausts any paging mode (3.6.2).
193pub fn validate_page(
194    page: &AcquisitionPage,
195    request: &AcquisitionRequest,
196    operation: &Operation,
197    evaluation_time: Timestamp,
198    skew: DurationMs,
199) -> Result<(), PageError> {
200    if request.id().ok().as_ref() != Some(&page.request) {
201        return Err(PageError::RequestIdentity);
202    }
203    if page.items.len() as u64 > operation.max_page_items {
204        return Err(PageError::ItemCount);
205    }
206    for (index, item) in page.items.iter().enumerate() {
207        if item.len() != operation.item.len()
208            || !operation
209                .item
210                .iter()
211                .all(|f| item.get(&f.name).is_some_and(|v| matches_type(&f.ty, v)))
212        {
213            return Err(PageError::Item { index });
214        }
215    }
216    match (&operation.paging, &page.next) {
217        (_, NextPage::None) => {}
218        (Paging::Cursor { max_cursor_bytes }, NextPage::Cursor { text }) => {
219            if text.len() as u64 > *max_cursor_bytes {
220                return Err(PageError::CursorLength);
221            }
222        }
223        (Paging::Offset, NextPage::Offset { integer }) => {
224            let offset = match request.page {
225                RequestPage::First => 0,
226                RequestPage::Offset { integer } => integer,
227                _ => return Err(PageError::NextKind),
228            };
229            if offset.checked_add(page.items.len() as u64) != Some(*integer) {
230                return Err(PageError::Offset);
231            }
232        }
233        _ => return Err(PageError::NextKind),
234    }
235    // Signed, wider arithmetic keeps the bound exact for pre-epoch times
236    // and for a `u64::MAX` skew: no cast can wrap.
237    if i128::from(page.as_of_ms.as_ms()) > i128::from(evaluation_time.as_ms()) + i128::from(skew.0)
238    {
239        return Err(PageError::AsOfSkew);
240    }
241    if request
242        .canonical()
243        .map_or(true, |b| b.len() as u64 > operation.max_request_bytes)
244    {
245        return Err(PageError::RequestBytes);
246    }
247    if page
248        .canonical()
249        .map_or(true, |b| b.len() as u64 > operation.max_page_bytes)
250    {
251        return Err(PageError::PageBytes);
252    }
253    Ok(())
254}
255
256fn matches_type(ty: &TypeDecl, v: &Value) -> bool {
257    match (ty, v) {
258        (TypeDecl::Bool, Value::Bool(_)) => true,
259        (TypeDecl::Integer, Value::Number(n)) => n.as_i64().is_some(),
260        (TypeDecl::Decimal, Value::String(s)) => Decimal::parse(s).is_ok(),
261        (TypeDecl::Text { max_bytes }, Value::String(s)) => s.len() as u64 <= *max_bytes,
262        (TypeDecl::Timestamp, Value::Number(n)) => {
263            n.as_i64().is_some_and(|n| Timestamp::from_ms(n).is_ok())
264        }
265        (TypeDecl::Enum { options }, Value::String(s)) => options.contains(s),
266        (TypeDecl::List { item, max_items }, Value::Array(a)) => {
267            a.len() as u64 <= *max_items && a.iter().all(|v| matches_type(item, v))
268        }
269        (
270            TypeDecl::Record {
271                fields,
272                present_when: None,
273            },
274            Value::Object(o),
275        ) => {
276            o.len() == fields.len()
277                && fields
278                    .iter()
279                    .all(|f| o.get(&f.name).is_some_and(|v| matches_type(&f.ty, v)))
280        }
281        _ => false,
282    }
283}
284
285fn bound<'de, D: Deserializer<'de>, const MIN: u64, const MAX: u64>(d: D) -> Result<u64, D::Error> {
286    let n = u64::deserialize(d)?;
287    if !(MIN..=MAX).contains(&n) {
288        return Err(serde::de::Error::custom("acquisition bound exceeded"));
289    }
290    Ok(n)
291}
292fn list<'de, D: Deserializer<'de>, T: Deserialize<'de>, const MIN: usize, const MAX: usize>(
293    d: D,
294) -> Result<Vec<T>, D::Error> {
295    let v = Vec::<T>::deserialize(d)?;
296    if !(MIN..=MAX).contains(&v.len()) {
297        return Err(serde::de::Error::custom("acquisition field count exceeded"));
298    }
299    Ok(v)
300}
301fn name<'de, D: Deserializer<'de>>(d: D) -> Result<String, D::Error> {
302    let s = String::deserialize(d)?;
303    if !valid_source_name(&s) {
304        return Err(serde::de::Error::custom("invalid operation name"));
305    }
306    Ok(s)
307}
308fn unique<'a>(names: impl Iterator<Item = &'a str>) -> bool {
309    let mut seen = BTreeSet::new();
310    names.into_iter().all(|n| !n.is_empty() && seen.insert(n))
311}
312pub(crate) fn operations<'de, D: Deserializer<'de>>(
313    d: D,
314) -> Result<Option<Vec<Operation>>, D::Error> {
315    // Present means a list: an explicit `null` fails here, and absence never
316    // reaches this function (`#[serde(default)]`).
317    let ops = Vec::<Operation>::deserialize(d)?;
318    if !(1..=64).contains(&ops.len()) || !unique(ops.iter().map(|o| o.name.as_str())) {
319        return Err(serde::de::Error::custom(
320            "operations must have 1 to 64 unique names",
321        ));
322    }
323    Ok(Some(ops))
324}
325fn request_fields<'de, D: Deserializer<'de>>(d: D) -> Result<Vec<RequestField>, D::Error> {
326    let fields = list::<D, RequestField, 0, 32>(d)?;
327    // 3.2.2 constrains only the operation name's bytes; request and item
328    // field names are spec 002 record field names, checked where records
329    // are (as for `FieldDecl`). Here they must be non-empty and unique.
330    if fields.iter().any(|f| f.name.is_empty()) || !unique(fields.iter().map(|f| f.name.as_str())) {
331        return Err(serde::de::Error::custom("invalid request field names"));
332    }
333    Ok(fields)
334}
335fn item_fields<'de, D: Deserializer<'de>>(d: D) -> Result<Vec<ItemField>, D::Error> {
336    let fields = list::<D, ItemField, 1, 64>(d)?;
337    if fields.iter().any(|f| f.name.is_empty()) || !unique(fields.iter().map(|f| f.name.as_str())) {
338        return Err(serde::de::Error::custom("invalid item field names"));
339    }
340    Ok(fields)
341}
342fn reliability<'de, D: Deserializer<'de>>(d: D) -> Result<Reliability, D::Error> {
343    let r = Reliability::deserialize(d)?;
344    if let Reliability::KnownUnreliable { note } = &r
345        && !(1..=256).contains(&note.len())
346    {
347        return Err(serde::de::Error::custom(
348            "reliability note must have 1 to 256 bytes",
349        ));
350    }
351    Ok(r)
352}
353fn scalar(ty: &TypeDecl) -> bool {
354    matches!(
355        ty,
356        TypeDecl::Bool
357            | TypeDecl::Integer
358            | TypeDecl::Decimal
359            | TypeDecl::Text { .. }
360            | TypeDecl::Timestamp
361            | TypeDecl::Enum { .. }
362    )
363}
364fn valid_type(ty: &TypeDecl) -> bool {
365    match ty {
366        TypeDecl::Text { max_bytes } => *max_bytes > 0,
367        TypeDecl::Enum { options } => {
368            !options.is_empty() && unique(options.iter().map(String::as_str))
369        }
370        TypeDecl::List { item, .. } => valid_type(item),
371        TypeDecl::Record {
372            fields,
373            present_when: None,
374        } => {
375            !fields.is_empty()
376                && unique(fields.iter().map(|f| f.name.as_str()))
377                && fields.iter().all(|f| valid_type(&f.ty))
378        }
379        t => scalar(t),
380    }
381}
382fn item_type<'de, D: Deserializer<'de>>(d: D) -> Result<TypeDecl, D::Error> {
383    let ty = TypeDecl::deserialize(d)?;
384    if !valid_type(&ty) {
385        return Err(serde::de::Error::custom("invalid spec 002 type"));
386    }
387    Ok(ty)
388}
389fn request_type<'de, D: Deserializer<'de>>(d: D) -> Result<TypeDecl, D::Error> {
390    let ty = item_type(d)?;
391    if !(scalar(&ty) || matches!(&ty, TypeDecl::List { item, .. } if scalar(item))) {
392        return Err(serde::de::Error::custom(
393            "request type must be scalar or list of scalars",
394        ));
395    }
396    Ok(ty)
397}
rustev_contract/acquisition.rs

rustev_contract/
acquisition.rs
