laser_wire/
validate.rs

1use crate::batch::BatchRequest;
2use crate::browse::RegisterSchema;
3use crate::clients::ClientMetadata;
4use crate::codes::AGDX_BATCH_CODE;
5use crate::codes::KV_LEASE_OP_VERSION;
6use crate::error::InvalidError;
7use crate::graph::{GraphEdge, GraphNode, GraphQuery, GraphUpsert, SourceRef};
8use crate::kv::{KvCasFenced, KvLease, KvLeaseRenew, KvRelease, KvScan, KvSet};
9use crate::limits::{
10    MAX_BATCH_OPS, MAX_CLIENT_METADATA, MAX_FRAME_BYTES, MAX_GRAPH_LABEL_BYTES,
11    MAX_GRAPH_NODE_LABELS, MAX_GRAPH_RESULT_ELEMENTS, MAX_GRAPH_TRAVERSE_DEPTH,
12    MAX_HOLDER_ID_BYTES, MAX_KEY_BYTES, MAX_LEASE_TTL_MICROS, MAX_MEMORY_BODY_BYTES,
13    MAX_MEMORY_ID_BYTES, MAX_METADATA_ENTRIES, MAX_METADATA_KEY_BYTES, MAX_PAGE_SIZE,
14    MAX_QUERY_NAME_BYTES, MAX_SCAN_LIMIT, MAX_SCHEMA_NAME_BYTES, MAX_SOURCE_REF_BYTES,
15    MAX_TEXT_QUERY_BYTES, MAX_VALUE_BYTES, MAX_VECTOR_DIMENSIONS, MIN_LEASE_TTL_MICROS,
16};
17use crate::memory::MemoryRecord;
18use crate::query::{TextQuery, Value, VectorQuery};
19
20/// A capped request type that enforces its own size and shape limits, so the
21/// cap logic lives once in the wire crate and every port and both servers get
22/// the identical check by construction rather than each remembering to compare
23/// against [`crate::limits`]. The SDK calls it before encoding, the servers call
24/// it after decode and before execution.
25pub trait Validate {
26    /// Reject a request that violates a pinned cap or a structural rule.
27    fn validate(&self) -> Result<(), InvalidError>;
28}
29
30/// The shared rule for caller-chosen names that flow into matching, filtering,
31/// or storage identifiers: non-empty, within `cap` bytes, and made only of
32/// ASCII letters, digits, `-`, `_`, and `.`. A strict safelist, not just a
33/// length bound, because these names get inlined into queries, filters, and
34/// rendered views.
35pub(crate) fn validate_safelisted_name(
36    label: &str,
37    value: &str,
38    cap: usize,
39) -> Result<(), InvalidError> {
40    if value.is_empty() {
41        return Err(InvalidError::new(format!("{label} must not be empty")));
42    }
43    if value.len() > cap {
44        return Err(InvalidError::new(format!(
45            "{label} is {}B, exceeds cap {cap}B",
46            value.len()
47        )));
48    }
49    if let Some(bad) = value
50        .bytes()
51        .find(|byte| !matches!(byte, b'a'..=b'z' | b'A'..=b'Z' | b'0'..=b'9' | b'-' | b'_' | b'.'))
52    {
53        return Err(InvalidError::new(format!(
54            "{label} has a disallowed byte {bad:#04x}: allowed are ASCII letters, digits, '-', '_', '.'"
55        )));
56    }
57    Ok(())
58}
59
60/// The shared rule for caller-chosen text that names a resource without a
61/// safelist: non-empty, within `cap` bytes, and free of control characters.
62pub(crate) fn validate_bounded_text(
63    label: &str,
64    value: &str,
65    cap: usize,
66) -> Result<(), InvalidError> {
67    if value.is_empty() {
68        return Err(InvalidError::new(format!("{label} must not be empty")));
69    }
70    if value.len() > cap {
71        return Err(InvalidError::new(format!(
72            "{label} is {}B, exceeds cap {cap}B",
73            value.len()
74        )));
75    }
76    if value.chars().any(char::is_control) {
77        return Err(InvalidError::new(format!(
78            "{label} contains a control character"
79        )));
80    }
81    Ok(())
82}
83
84impl Validate for BatchRequest {
85    fn validate(&self) -> Result<(), InvalidError> {
86        if self.ops.len() > MAX_BATCH_OPS {
87            return Err(InvalidError::new(format!(
88                "batch has {} ops, exceeds cap {MAX_BATCH_OPS}",
89                self.ops.len()
90            )));
91        }
92        let mut total = 0usize;
93        for item in &self.ops {
94            // A batch never nests a batch: an op that is itself a batch would let
95            // one request fan out without bound past the op cap.
96            if item.code == AGDX_BATCH_CODE {
97                return Err(InvalidError::new(
98                    "a batch may not contain a batch op".to_owned(),
99                ));
100            }
101            if item.payload.len() > MAX_VALUE_BYTES {
102                return Err(InvalidError::new(format!(
103                    "batch op payload is {}B, exceeds cap {MAX_VALUE_BYTES}B",
104                    item.payload.len()
105                )));
106            }
107            total = total.saturating_add(item.payload.len());
108        }
109        if total > MAX_FRAME_BYTES {
110            return Err(InvalidError::new(format!(
111                "batch total payload is {total}B, exceeds cap {MAX_FRAME_BYTES}B"
112            )));
113        }
114        Ok(())
115    }
116}
117
118impl Validate for KvSet {
119    fn validate(&self) -> Result<(), InvalidError> {
120        crate::kv::validate_namespace(&self.namespace)?;
121        if self.key.is_empty() {
122            return Err(InvalidError::new("key-value key is empty".to_owned()));
123        }
124        if self.key.len() > MAX_KEY_BYTES {
125            return Err(InvalidError::new(format!(
126                "key is {}B, exceeds cap {MAX_KEY_BYTES}B",
127                self.key.len()
128            )));
129        }
130        if self.value.len() > MAX_VALUE_BYTES {
131            return Err(InvalidError::new(format!(
132                "value is {}B, exceeds cap {MAX_VALUE_BYTES}B",
133                self.value.len()
134            )));
135        }
136        Ok(())
137    }
138}
139
140impl Validate for KvScan {
141    fn validate(&self) -> Result<(), InvalidError> {
142        crate::kv::validate_namespace(&self.namespace)?;
143        if self.limit > MAX_SCAN_LIMIT {
144            return Err(InvalidError::new(format!(
145                "scan limit {} exceeds cap {MAX_SCAN_LIMIT}",
146                self.limit
147            )));
148        }
149        Ok(())
150    }
151}
152
153// The fenced-lease family shares its structural rules, so they live once: the
154// exact op version (any other `v` is a typed rejection, the fail-closed half
155// of the in-place reshape), a bounded key, a bounded non-empty holder.
156fn validate_lease_shape(
157    v: u32,
158    namespace: &str,
159    key: &[u8],
160    holder_id: &str,
161) -> Result<(), InvalidError> {
162    if v != KV_LEASE_OP_VERSION {
163        return Err(InvalidError::new(format!(
164            "unsupported fenced-lease op version (expected {KV_LEASE_OP_VERSION}, got {v})"
165        )));
166    }
167    crate::kv::validate_namespace(namespace)?;
168    validate_kv_key("lease key", key)?;
169    if holder_id.is_empty() {
170        return Err(InvalidError::new("lease holder id must not be empty"));
171    }
172    if holder_id.len() > MAX_HOLDER_ID_BYTES {
173        return Err(InvalidError::new(format!(
174            "lease holder id is {}B, exceeds cap {MAX_HOLDER_ID_BYTES}B",
175            holder_id.len()
176        )));
177    }
178    if holder_id.chars().any(char::is_control) {
179        return Err(InvalidError::new(
180            "lease holder id contains a control character",
181        ));
182    }
183    Ok(())
184}
185
186// A requested lease lifetime is bounded on both sides, and the bound is the
187// contract's, not each backend's: a floor so a grant survives one round trip
188// plus the holder's renewal cadence, a ceiling so a crashed holder's ownership
189// always expires. The store may still grant less than the request
190// (`granted_ttl_micros`), so this rejects only what no deployment would serve,
191// and it rejects it locally instead of after a round trip.
192fn validate_lease_ttl(ttl_micros: u64) -> Result<(), InvalidError> {
193    if !(MIN_LEASE_TTL_MICROS..=MAX_LEASE_TTL_MICROS).contains(&ttl_micros) {
194        return Err(InvalidError::new(format!(
195            "lease ttl is {ttl_micros}\u{b5}s, outside the supported range \
196             {MIN_LEASE_TTL_MICROS}..={MAX_LEASE_TTL_MICROS}\u{b5}s"
197        )));
198    }
199    Ok(())
200}
201
202fn validate_kv_key(label: &str, key: &[u8]) -> Result<(), InvalidError> {
203    if key.is_empty() {
204        return Err(InvalidError::new(format!("{label} is empty")));
205    }
206    if key.len() > MAX_KEY_BYTES {
207        return Err(InvalidError::new(format!(
208            "{label} is {}B, exceeds cap {MAX_KEY_BYTES}B",
209            key.len()
210        )));
211    }
212    Ok(())
213}
214
215impl Validate for KvLease {
216    fn validate(&self) -> Result<(), InvalidError> {
217        validate_lease_shape(self.v, &self.namespace, &self.key, &self.holder_id)?;
218        validate_lease_ttl(self.lease_ttl_micros)?;
219        Ok(())
220    }
221}
222
223impl Validate for KvLeaseRenew {
224    fn validate(&self) -> Result<(), InvalidError> {
225        validate_lease_shape(self.v, &self.namespace, &self.key, &self.holder_id)?;
226        validate_lease_ttl(self.lease_ttl_micros)?;
227        if self.lease_token == 0 {
228            return Err(InvalidError::new("lease token must not be zero"));
229        }
230        Ok(())
231    }
232}
233
234impl Validate for KvRelease {
235    fn validate(&self) -> Result<(), InvalidError> {
236        validate_lease_shape(self.v, &self.namespace, &self.key, &self.holder_id)?;
237        if self.lease_token == 0 {
238            return Err(InvalidError::new("lease token must not be zero"));
239        }
240        Ok(())
241    }
242}
243
244impl Validate for KvCasFenced {
245    fn validate(&self) -> Result<(), InvalidError> {
246        if self.v != KV_LEASE_OP_VERSION {
247            return Err(InvalidError::new(format!(
248                "unsupported fenced-lease op version (expected {KV_LEASE_OP_VERSION}, got {})",
249                self.v
250            )));
251        }
252        crate::kv::validate_namespace(&self.namespace)?;
253        crate::kv::validate_namespace(&self.fence_namespace)?;
254        validate_kv_key("key-value key", &self.key)?;
255        validate_kv_key("fence key", &self.fence_key)?;
256        if self.value.len() > MAX_VALUE_BYTES {
257            return Err(InvalidError::new(format!(
258                "value is {}B, exceeds cap {MAX_VALUE_BYTES}B",
259                self.value.len()
260            )));
261        }
262        // Zero is the reserved "unversioned" sentinel, never a granted token.
263        if self.fence_token == 0 {
264            return Err(InvalidError::new("fence token must not be zero"));
265        }
266        Ok(())
267    }
268}
269
270impl Validate for GraphQuery {
271    fn validate(&self) -> Result<(), InvalidError> {
272        crate::graph::validate_graph_name(&self.graph)?;
273        if self.traverse.len() > MAX_GRAPH_TRAVERSE_DEPTH as usize {
274            return Err(InvalidError::new(format!(
275                "graph traversal depth {} exceeds cap {MAX_GRAPH_TRAVERSE_DEPTH}",
276                self.traverse.len()
277            )));
278        }
279        if self.limit > MAX_GRAPH_RESULT_ELEMENTS {
280            return Err(InvalidError::new(format!(
281                "graph result limit {} exceeds cap {MAX_GRAPH_RESULT_ELEMENTS}",
282                self.limit
283            )));
284        }
285        Ok(())
286    }
287}
288
289impl Validate for SourceRef {
290    fn validate(&self) -> Result<(), InvalidError> {
291        let size = match self {
292            // A log pointer is fixed-width numerics plus an optional
293            // conversation id, so only the id needs a bound.
294            SourceRef::Message { conversation, .. } => conversation.as_ref().map_or(0, String::len),
295            SourceRef::Kv { namespace, key } => namespace.len() + key.len(),
296            SourceRef::Memory { id } => id.len(),
297        };
298        if size > MAX_SOURCE_REF_BYTES {
299            return Err(InvalidError::new(format!(
300                "source reference is {size}B, exceeds cap {MAX_SOURCE_REF_BYTES}B"
301            )));
302        }
303        Ok(())
304    }
305}
306
307// Attribute lists ride on both nodes and edges, so their bound lives once.
308fn validate_attrs(label: &str, attrs: &[(String, Value)]) -> Result<(), InvalidError> {
309    if attrs.len() > MAX_METADATA_ENTRIES {
310        return Err(InvalidError::new(format!(
311            "{label} carries {} attributes, exceeds cap {MAX_METADATA_ENTRIES}",
312            attrs.len()
313        )));
314    }
315    for (key, _) in attrs {
316        if key.len() > MAX_METADATA_KEY_BYTES {
317            return Err(InvalidError::new(format!(
318                "{label} attribute name is {}B, exceeds cap {MAX_METADATA_KEY_BYTES}B",
319                key.len()
320            )));
321        }
322    }
323    Ok(())
324}
325
326impl Validate for GraphNode {
327    fn validate(&self) -> Result<(), InvalidError> {
328        if self.labels.len() > MAX_GRAPH_NODE_LABELS {
329            return Err(InvalidError::new(format!(
330                "graph node carries {} labels, exceeds cap {MAX_GRAPH_NODE_LABELS}",
331                self.labels.len()
332            )));
333        }
334        for label in &self.labels {
335            validate_bounded_text("graph node label", label, MAX_GRAPH_LABEL_BYTES)?;
336        }
337        validate_attrs("graph node", &self.attrs)?;
338        if let Some(source) = &self.source {
339            source.validate()?;
340        }
341        Ok(())
342    }
343}
344
345impl Validate for GraphEdge {
346    fn validate(&self) -> Result<(), InvalidError> {
347        validate_bounded_text("graph edge type", &self.edge_type, MAX_GRAPH_LABEL_BYTES)?;
348        validate_attrs("graph edge", &self.attrs)?;
349        if let Some(source) = &self.source {
350            source.validate()?;
351        }
352        Ok(())
353    }
354}
355
356impl Validate for GraphUpsert {
357    fn validate(&self) -> Result<(), InvalidError> {
358        crate::graph::validate_graph_name(&self.graph)?;
359        let elements = self.nodes.len() + self.edges.len();
360        if elements > MAX_GRAPH_RESULT_ELEMENTS {
361            return Err(InvalidError::new(format!(
362                "graph upsert carries {elements} elements, exceeds cap {MAX_GRAPH_RESULT_ELEMENTS}"
363            )));
364        }
365        for node in &self.nodes {
366            node.validate()?;
367        }
368        for edge in &self.edges {
369            edge.validate()?;
370        }
371        Ok(())
372    }
373}
374
375impl Validate for TextQuery {
376    fn validate(&self) -> Result<(), InvalidError> {
377        if self.query.trim().is_empty() || self.query.len() > MAX_TEXT_QUERY_BYTES {
378            return Err(InvalidError::new(format!(
379                "text query is {}B, expected 1..={MAX_TEXT_QUERY_BYTES}B",
380                self.query.len()
381            )));
382        }
383        if self.query.chars().any(char::is_control) {
384            return Err(InvalidError::new("text query contains a control character"));
385        }
386        if let Some(field) = &self.field
387            && (field.is_empty()
388                || field.len() > MAX_QUERY_NAME_BYTES
389                || field.chars().any(char::is_control))
390        {
391            return Err(InvalidError::new("text query field is invalid"));
392        }
393        Ok(())
394    }
395}
396
397impl Validate for VectorQuery {
398    fn validate(&self) -> Result<(), InvalidError> {
399        if self.field.is_empty()
400            || self.field.len() > MAX_QUERY_NAME_BYTES
401            || self.field.chars().any(char::is_control)
402        {
403            return Err(InvalidError::new("vector field is invalid"));
404        }
405        if self.embedding.is_empty() || self.embedding.len() > MAX_VECTOR_DIMENSIONS {
406            return Err(InvalidError::new(format!(
407                "vector dimensions must be in 1..={MAX_VECTOR_DIMENSIONS}"
408            )));
409        }
410        if self
411            .embedding
412            .iter()
413            .any(|value| !value.is_finite() || (*value == 0.0 && value.is_sign_negative()))
414        {
415            return Err(InvalidError::new(
416                "vector values must be finite canonical floats",
417            ));
418        }
419        if self.top_k == 0 || self.top_k > MAX_PAGE_SIZE as u32 {
420            return Err(InvalidError::new(format!(
421                "vector top_k {} is outside 1..={MAX_PAGE_SIZE}",
422                self.top_k
423            )));
424        }
425        Ok(())
426    }
427}
428
429impl Validate for MemoryRecord {
430    fn validate(&self) -> Result<(), InvalidError> {
431        match self {
432            MemoryRecord::Item { id, kind, .. } => {
433                validate_bounded_text("memory item id", id, MAX_MEMORY_ID_BYTES)?;
434                validate_bounded_text("memory item kind", kind, MAX_MEMORY_ID_BYTES)?;
435            }
436            MemoryRecord::Forget { target, .. } | MemoryRecord::Feedback { target, .. } => {
437                validate_bounded_text("memory target", target, MAX_MEMORY_ID_BYTES)?;
438            }
439        }
440        if let MemoryRecord::Item { body, .. } = self
441            && body.len() > MAX_MEMORY_BODY_BYTES
442        {
443            return Err(InvalidError::new(format!(
444                "memory body is {}B, exceeds cap {MAX_MEMORY_BODY_BYTES}B",
445                body.len()
446            )));
447        }
448        if let MemoryRecord::Forget {
449            conversation: Some(conversation),
450            ..
451        }
452        | MemoryRecord::Feedback {
453            conversation: Some(conversation),
454            ..
455        } = self
456        {
457            conversation
458                .parse::<crate::agent::ConversationId>()
459                .map_err(|_| InvalidError::new("memory record conversation is not a valid id"))?;
460        }
461        Ok(())
462    }
463}
464
465impl Validate for RegisterSchema {
466    fn validate(&self) -> Result<(), InvalidError> {
467        if let Some(name) = &self.name {
468            validate_bounded_text("schema name", name, MAX_SCHEMA_NAME_BYTES)?;
469        }
470        Ok(())
471    }
472}
473
474impl Validate for ClientMetadata {
475    fn validate(&self) -> Result<(), InvalidError> {
476        let size = self.metadata.as_ref().map_or(0, Vec::len);
477        if size > MAX_CLIENT_METADATA {
478            return Err(InvalidError::new(format!(
479                "client metadata is {size}B, exceeds cap {MAX_CLIENT_METADATA}B"
480            )));
481        }
482        Ok(())
483    }
484}
485
486#[cfg(test)]
487mod tests {
488    use super::*;
489    use crate::batch::BatchItem;
490    use crate::codes::{
491        AGDX_KV_GET_CODE, BATCH_OP_VERSION, KV_LEASE_OP_VERSION, KV_OP_VERSION, QUERY_OP_VERSION,
492    };
493    use crate::control::SchemaSource;
494    use crate::limits::{MAX_GRAPH_LABEL_BYTES, MAX_MEMORY_ID_BYTES, MAX_SCHEMA_NAME_BYTES};
495
496    #[test]
497    fn given_an_invalid_graph_label_or_edge_type_when_upserted_then_should_refuse_it() {
498        let at_cap = "l".repeat(MAX_GRAPH_LABEL_BYTES);
499        let alice = node_with(vec![at_cap.clone()], Vec::new());
500        let acme = node_with(vec!["Company".to_owned()], Vec::new());
501        alice.validate().expect("a node label at the cap is valid");
502        GraphEdge::relate(&alice, at_cap, &acme)
503            .validate()
504            .expect("an edge type at the cap is valid");
505        let long = "l".repeat(MAX_GRAPH_LABEL_BYTES + 1);
506        for invalid in [String::new(), long, "Per\nson".to_owned()] {
507            let node = node_with(vec![invalid.clone()], Vec::new());
508            assert!(node.validate().is_err(), "{invalid:?}");
509            assert!(
510                GraphEdge::relate(&alice, invalid.clone(), &acme)
511                    .validate()
512                    .is_err()
513            );
514            let upsert = GraphUpsert {
515                v: 1,
516                graph: "knowledge".to_owned(),
517                session: None,
518                nodes: vec![node],
519                edges: Vec::new(),
520            };
521            assert!(upsert.validate().is_err());
522        }
523    }
524
525    #[test]
526    fn given_an_invalid_memory_id_kind_or_target_when_validated_then_should_refuse_it() {
527        let item = |id: &str, kind: &str| MemoryRecord::Item {
528            origin: None,
529            producer: None,
530            id: id.to_owned(),
531            kind: kind.to_owned(),
532            body: b"auth is slow".to_vec(),
533        };
534        let at_cap = "m".repeat(MAX_MEMORY_ID_BYTES);
535        item(&at_cap, &at_cap)
536            .validate()
537            .expect("an id and a kind at the cap are valid");
538        for invalid in [
539            String::new(),
540            "m".repeat(MAX_MEMORY_ID_BYTES + 1),
541            "fa\u{0}ct".to_owned(),
542        ] {
543            assert!(item(&invalid, "fact").validate().is_err(), "{invalid:?}");
544            assert!(
545                item("01KWM3K3XEP3NP5TN850J17YBP", &invalid)
546                    .validate()
547                    .is_err()
548            );
549            assert!(
550                MemoryRecord::Forget {
551                    target: invalid.clone(),
552                    conversation: None,
553                }
554                .validate()
555                .is_err()
556            );
557            assert!(
558                MemoryRecord::Feedback {
559                    target: invalid,
560                    weight: 1.0,
561                    conversation: None,
562                }
563                .validate()
564                .is_err()
565            );
566        }
567    }
568
569    #[test]
570    fn given_a_lease_holder_with_a_control_character_when_validated_then_should_refuse_it() {
571        let holder = "node\n1".to_owned();
572        let lease = KvLease {
573            v: KV_LEASE_OP_VERSION,
574            namespace: "connectors.coordination".to_owned(),
575            key: b"source-owner".to_vec(),
576            lease_ttl_micros: 30_000_000,
577            holder_id: holder.clone(),
578            subject_user_id: None,
579        };
580        assert!(lease.validate().is_err());
581        let renew = KvLeaseRenew {
582            v: KV_LEASE_OP_VERSION,
583            namespace: "connectors.coordination".to_owned(),
584            key: b"source-owner".to_vec(),
585            holder_id: holder.clone(),
586            subject_user_id: None,
587            lease_token: 7,
588            lease_ttl_micros: 30_000_000,
589        };
590        assert!(renew.validate().is_err());
591        let release = KvRelease {
592            v: KV_LEASE_OP_VERSION,
593            namespace: "connectors.coordination".to_owned(),
594            key: b"source-owner".to_vec(),
595            lease_token: 7,
596            holder_id: holder,
597        };
598        assert!(release.validate().is_err());
599    }
600
601    #[test]
602    fn given_an_invalid_schema_name_when_a_registration_is_validated_then_should_refuse_it() {
603        let register = |name: Option<String>| RegisterSchema {
604            v: QUERY_OP_VERSION,
605            source: SchemaSource::JsonSchema {
606                schema: "{}".to_owned(),
607            },
608            name,
609            version: None,
610            stream: Some("fleet".to_owned()),
611        };
612        register(None)
613            .validate()
614            .expect("a registration without a name is valid");
615        register(Some("reading".to_owned()))
616            .validate()
617            .expect("a short schema name is valid");
618        assert!(register(Some(String::new())).validate().is_err());
619        assert!(
620            register(Some("s".repeat(MAX_SCHEMA_NAME_BYTES + 1)))
621                .validate()
622                .is_err()
623        );
624        assert!(register(Some("read\ting".to_owned())).validate().is_err());
625    }
626
627    #[test]
628    fn given_a_batch_over_the_op_cap_when_validated_then_should_reject() {
629        let request = BatchRequest {
630            v: BATCH_OP_VERSION,
631            ops: (0..MAX_BATCH_OPS + 1)
632                .map(|_| BatchItem {
633                    code: AGDX_KV_GET_CODE,
634                    payload: Vec::new(),
635                })
636                .collect(),
637        };
638        assert!(request.validate().is_err());
639    }
640
641    #[test]
642    fn given_a_batch_nesting_a_batch_when_validated_then_should_reject() {
643        let request = BatchRequest {
644            v: BATCH_OP_VERSION,
645            ops: vec![BatchItem {
646                code: AGDX_BATCH_CODE,
647                payload: Vec::new(),
648            }],
649        };
650        assert!(request.validate().is_err());
651    }
652
653    #[test]
654    fn given_namespaces_when_validated_then_should_enforce_bounds() {
655        use crate::kv::validate_namespace;
656        use crate::limits::MAX_NAMESPACE_BYTES;
657        assert!(validate_namespace("default").is_ok());
658        assert!(validate_namespace("agent-abc/session").is_ok(), "hierarchy");
659        assert!(validate_namespace("").is_err(), "empty");
660        assert!(validate_namespace("bad\nns").is_err(), "control byte");
661        assert!(validate_namespace(&"n".repeat(MAX_NAMESPACE_BYTES)).is_ok());
662        assert!(validate_namespace(&"n".repeat(MAX_NAMESPACE_BYTES + 1)).is_err());
663    }
664
665    #[test]
666    fn given_graph_names_when_validated_then_should_enforce_bounds() {
667        use crate::graph::validate_graph_name;
668        use crate::limits::MAX_GRAPH_NAME_BYTES;
669        assert!(validate_graph_name("knowledge").is_ok());
670        assert!(validate_graph_name("").is_err(), "empty");
671        assert!(validate_graph_name("bad\tname").is_err(), "control byte");
672        assert!(validate_graph_name(&"g".repeat(MAX_GRAPH_NAME_BYTES + 1)).is_err());
673    }
674
675    #[test]
676    fn given_a_lease_request_when_validated_then_should_enforce_version_holder_and_ttl() {
677        use crate::codes::KV_LEASE_OP_VERSION;
678        use crate::kv::KvLease;
679        use crate::limits::MAX_HOLDER_ID_BYTES;
680        let valid = KvLease {
681            v: KV_LEASE_OP_VERSION,
682            namespace: "connectors.coordination".to_owned(),
683            key: b"source-owner".to_vec(),
684            lease_ttl_micros: 30_000_000,
685            holder_id: "node-1".to_owned(),
686            subject_user_id: Some(42),
687        };
688        assert!(valid.validate().is_ok());
689        let wrong_version = KvLease {
690            v: KV_LEASE_OP_VERSION + 1,
691            ..valid.clone()
692        };
693        assert!(
694            wrong_version.validate().is_err(),
695            "an unsupported version is rejected typed"
696        );
697        let empty_holder = KvLease {
698            holder_id: String::new(),
699            ..valid.clone()
700        };
701        assert!(empty_holder.validate().is_err());
702        let long_holder = KvLease {
703            holder_id: "h".repeat(MAX_HOLDER_ID_BYTES + 1),
704            ..valid.clone()
705        };
706        assert!(long_holder.validate().is_err());
707        let zero_ttl = KvLease {
708            lease_ttl_micros: 0,
709            ..valid
710        };
711        assert!(zero_ttl.validate().is_err());
712    }
713
714    // The TTL range is the contract's, so the same rejection happens on the
715    // acquire and the renewal, before any round trip, on every tier.
716    #[test]
717    fn given_out_of_range_lease_ttls_when_validated_then_should_reject_both_ends() {
718        use crate::codes::KV_LEASE_OP_VERSION;
719        use crate::kv::{KvLease, KvLeaseRenew};
720        use crate::limits::{MAX_LEASE_TTL_MICROS, MIN_LEASE_TTL_MICROS};
721        let acquire = KvLease {
722            v: KV_LEASE_OP_VERSION,
723            namespace: "connectors.coordination".to_owned(),
724            key: b"source-owner".to_vec(),
725            lease_ttl_micros: MIN_LEASE_TTL_MICROS,
726            holder_id: "node-1".to_owned(),
727            subject_user_id: None,
728        };
729        assert!(acquire.validate().is_ok(), "the floor itself is grantable");
730        assert!(
731            KvLease {
732                lease_ttl_micros: MAX_LEASE_TTL_MICROS,
733                ..acquire.clone()
734            }
735            .validate()
736            .is_ok(),
737            "the ceiling itself is grantable"
738        );
739        assert!(
740            KvLease {
741                lease_ttl_micros: MIN_LEASE_TTL_MICROS - 1,
742                ..acquire.clone()
743            }
744            .validate()
745            .is_err(),
746            "a lease too short to renew is rejected locally"
747        );
748        assert!(
749            KvLease {
750                lease_ttl_micros: MAX_LEASE_TTL_MICROS + 1,
751                ..acquire
752            }
753            .validate()
754            .is_err(),
755            "an unbounded ownership window is rejected locally"
756        );
757        let renew = KvLeaseRenew {
758            v: KV_LEASE_OP_VERSION,
759            namespace: "connectors.coordination".to_owned(),
760            key: b"source-owner".to_vec(),
761            holder_id: "node-1".to_owned(),
762            subject_user_id: None,
763            lease_token: 7,
764            lease_ttl_micros: MAX_LEASE_TTL_MICROS + 1,
765        };
766        assert!(renew.validate().is_err(), "renewal shares the same range");
767        assert!(
768            KvLeaseRenew {
769                lease_ttl_micros: MIN_LEASE_TTL_MICROS - 1,
770                ..renew
771            }
772            .validate()
773            .is_err()
774        );
775    }
776
777    #[test]
778    fn given_renew_and_release_requests_when_validated_then_should_require_a_nonzero_token() {
779        use crate::codes::KV_LEASE_OP_VERSION;
780        use crate::kv::{KvLeaseRenew, KvRelease};
781        let renew = KvLeaseRenew {
782            v: KV_LEASE_OP_VERSION,
783            namespace: "connectors.coordination".to_owned(),
784            key: b"source-owner".to_vec(),
785            holder_id: "node-1".to_owned(),
786            subject_user_id: None,
787            lease_token: 7,
788            lease_ttl_micros: 30_000_000,
789        };
790        assert!(renew.validate().is_ok());
791        let zero = KvLeaseRenew {
792            lease_token: 0,
793            ..renew
794        };
795        assert!(zero.validate().is_err());
796        let release = KvRelease {
797            v: KV_LEASE_OP_VERSION,
798            namespace: "connectors.coordination".to_owned(),
799            key: b"source-owner".to_vec(),
800            lease_token: 7,
801            holder_id: "node-1".to_owned(),
802        };
803        assert!(release.validate().is_ok());
804        let zero = KvRelease {
805            lease_token: 0,
806            ..release
807        };
808        assert!(zero.validate().is_err());
809    }
810
811    #[test]
812    fn given_a_fenced_cas_when_validated_then_should_enforce_both_namespaces_and_the_token() {
813        use crate::codes::KV_LEASE_OP_VERSION;
814        use crate::kv::{CasExpect, KvCasFenced};
815        let valid = KvCasFenced {
816            v: KV_LEASE_OP_VERSION,
817            namespace: "connectors.state".to_owned(),
818            key: b"source_state/pg".to_vec(),
819            value: b"cursor".to_vec(),
820            expires_at_micros: None,
821            expect: CasExpect::Absent,
822            fence_namespace: "connectors.coordination".to_owned(),
823            fence_key: b"source-owner".to_vec(),
824            fence_token: 8,
825        };
826        assert!(valid.validate().is_ok());
827        let wrong_version = KvCasFenced {
828            v: KV_LEASE_OP_VERSION + 1,
829            ..valid.clone()
830        };
831        assert!(
832            wrong_version.validate().is_err(),
833            "an unsupported version is rejected typed"
834        );
835        let bad_fence_namespace = KvCasFenced {
836            fence_namespace: "bad\nns".to_owned(),
837            ..valid.clone()
838        };
839        assert!(bad_fence_namespace.validate().is_err());
840        let zero_token = KvCasFenced {
841            fence_token: 0,
842            ..valid
843        };
844        assert!(zero_token.validate().is_err());
845    }
846
847    #[test]
848    fn given_an_oversized_key_when_validated_then_should_reject_and_a_valid_one_passes() {
849        let over = KvSet {
850            v: KV_OP_VERSION,
851            namespace: "ns".to_owned(),
852            session: None,
853            key: vec![b'x'; MAX_KEY_BYTES + 1],
854            value: vec![1, 2, 3],
855            expires_at_micros: None,
856        };
857        assert!(over.validate().is_err());
858        let ok = KvSet {
859            v: KV_OP_VERSION,
860            namespace: "ns".to_owned(),
861            session: None,
862            key: vec![b'x'; 8],
863            value: vec![1, 2, 3],
864            expires_at_micros: None,
865        };
866        assert!(ok.validate().is_ok());
867    }
868
869    fn node_with(labels: Vec<String>, attrs: Vec<(String, Value)>) -> GraphNode {
870        GraphNode {
871            producer: None,
872            id: crate::graph::NodeId::from_u128(1),
873            labels,
874            attrs,
875            embedding: None,
876            source: None,
877        }
878    }
879
880    #[test]
881    fn given_a_node_over_the_label_cap_when_validated_then_should_reject() {
882        let over = node_with(
883            (0..=MAX_GRAPH_NODE_LABELS)
884                .map(|index| format!("label-{index}"))
885                .collect(),
886            Vec::new(),
887        );
888        assert!(over.validate().is_err());
889        assert!(
890            node_with(vec!["Person".to_owned()], Vec::new())
891                .validate()
892                .is_ok()
893        );
894    }
895
896    #[test]
897    fn given_a_node_over_the_attribute_cap_when_validated_then_should_reject() {
898        let over = node_with(
899            Vec::new(),
900            (0..=MAX_METADATA_ENTRIES)
901                .map(|index| (format!("attr-{index}"), Value::from("x")))
902                .collect(),
903        );
904        assert!(over.validate().is_err());
905    }
906
907    #[test]
908    fn given_an_oversized_source_reference_when_validated_then_should_reject() {
909        let over = SourceRef::Kv {
910            namespace: "n".repeat(MAX_SOURCE_REF_BYTES),
911            key: "k".repeat(MAX_SOURCE_REF_BYTES),
912        };
913        assert!(over.validate().is_err());
914        assert!(
915            SourceRef::Memory {
916                id: "01KWM3K3XEP3NP5TN850J17YBP".to_owned()
917            }
918            .validate()
919            .is_ok()
920        );
921    }
922
923    #[test]
924    fn given_an_upsert_over_the_element_cap_when_validated_then_should_reject() {
925        let over = GraphUpsert {
926            v: 1,
927            graph: "knowledge".to_owned(),
928            session: None,
929            nodes: (0..=MAX_GRAPH_RESULT_ELEMENTS)
930                .map(|_| node_with(Vec::new(), Vec::new()))
931                .collect(),
932            edges: Vec::new(),
933        };
934        assert!(over.validate().is_err());
935        let ok = GraphUpsert {
936            v: 1,
937            graph: "knowledge".to_owned(),
938            session: None,
939            nodes: vec![node_with(vec!["Person".to_owned()], Vec::new())],
940            edges: Vec::new(),
941        };
942        assert!(ok.validate().is_ok());
943    }
944
945    #[test]
946    fn given_an_upsert_carrying_an_over_cap_node_when_validated_then_should_reject() {
947        let upsert = GraphUpsert {
948            v: 1,
949            graph: "knowledge".to_owned(),
950            session: None,
951            nodes: vec![node_with(
952                (0..=MAX_GRAPH_NODE_LABELS)
953                    .map(|index| format!("label-{index}"))
954                    .collect(),
955                Vec::new(),
956            )],
957            edges: Vec::new(),
958        };
959        assert!(
960            upsert.validate().is_err(),
961            "an upsert must enforce its elements' caps, not just its own"
962        );
963    }
964
965    #[test]
966    fn given_an_oversized_text_query_when_validated_then_should_reject() {
967        let over = TextQuery {
968            field: None,
969            query: "q".repeat(MAX_TEXT_QUERY_BYTES + 1),
970        };
971        assert!(over.validate().is_err());
972        let ok = TextQuery {
973            field: None,
974            query: "auth is slow".to_owned(),
975        };
976        assert!(ok.validate().is_ok());
977    }
978
979    #[test]
980    fn given_an_over_cap_vector_top_k_when_validated_then_should_reject() {
981        let over = VectorQuery {
982            field: "embedding".to_owned(),
983            embedding: vec![0.0; 4],
984            top_k: MAX_PAGE_SIZE as u32 + 1,
985        };
986        assert!(over.validate().is_err());
987        let ok = VectorQuery {
988            field: "embedding".to_owned(),
989            embedding: vec![0.0; 4],
990            top_k: 10,
991        };
992        assert!(ok.validate().is_ok());
993    }
994
995    #[test]
996    fn given_an_oversized_memory_body_when_validated_then_should_reject() {
997        let over = MemoryRecord::Item {
998            origin: None,
999            producer: None,
1000            id: "01KWM3K3XEP3NP5TN850J17YBP".to_owned(),
1001            kind: "fact".to_owned(),
1002            body: vec![0u8; MAX_MEMORY_BODY_BYTES + 1],
1003        };
1004        assert!(over.validate().is_err());
1005        let ok = MemoryRecord::Item {
1006            origin: None,
1007            producer: None,
1008            id: "01KWM3K3XEP3NP5TN850J17YBP".to_owned(),
1009            kind: "fact".to_owned(),
1010            body: b"auth is slow".to_vec(),
1011        };
1012        assert!(ok.validate().is_ok());
1013        assert!(
1014            MemoryRecord::Forget {
1015                target: "01KWM3K3XEP3NP5TN850J17YBP".to_owned(),
1016                conversation: None,
1017            }
1018            .validate()
1019            .is_ok()
1020        );
1021        assert!(
1022            MemoryRecord::Forget {
1023                target: "01KWM3K3XEP3NP5TN850J17YBP".to_owned(),
1024                conversation: Some("not an id".to_owned()),
1025            }
1026            .validate()
1027            .is_err()
1028        );
1029    }
1030
1031    #[test]
1032    fn given_oversized_client_metadata_when_validated_then_should_reject() {
1033        let over = ClientMetadata {
1034            client_id: 1,
1035            user_id: None,
1036            transport: 1,
1037            address: "127.0.0.1:8090".to_owned(),
1038            consumer_groups_count: 0,
1039            metadata: Some(vec![0u8; MAX_CLIENT_METADATA + 1]),
1040        };
1041        assert!(over.validate().is_err());
1042        let ok = ClientMetadata {
1043            client_id: 1,
1044            user_id: None,
1045            transport: 1,
1046            address: "127.0.0.1:8090".to_owned(),
1047            consumer_groups_count: 0,
1048            metadata: Some(b"card".to_vec()),
1049        };
1050        assert!(ok.validate().is_ok());
1051    }
1052}
laser_wire/validate.rs

laser_wire/
validate.rs
