laser_wire/
kv.rs

1use crate::error::InvalidError;
2use crate::limits::MAX_NAMESPACE_BYTES;
3use serde::{Deserialize, Serialize};
4
5/// The canonical namespace rule, shared by the SDK client edge and every
6/// serving tier: non-empty, at most [`MAX_NAMESPACE_BYTES`] bytes, no ASCII
7/// control characters. The charset is otherwise open, so `/`-style hierarchy
8/// stays legal; the bound is a size and sanity cap, not a safelist.
9pub fn validate_namespace(namespace: &str) -> Result<(), InvalidError> {
10    if namespace.is_empty() {
11        return Err(InvalidError::new("namespace must not be empty"));
12    }
13    if namespace.len() > MAX_NAMESPACE_BYTES {
14        return Err(InvalidError::new(format!(
15            "namespace is {}B, exceeds cap {MAX_NAMESPACE_BYTES}B",
16            namespace.len()
17        )));
18    }
19    if namespace.bytes().any(|byte| byte.is_ascii_control()) {
20        return Err(InvalidError::new(
21            "namespace must not contain ASCII control characters",
22        ));
23    }
24    Ok(())
25}
26
27/// The memory scope a read-view row carries, stamped by the fold from the
28/// record's headers so recall reconstructs a memory item and narrows by scope.
29/// Absent on a generic key-value entry, which carries no scope.
30#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
31pub struct MemoryRowScope {
32    /// The item kind word, from the record body.
33    #[serde(default, skip_serializing_if = "Option::is_none")]
34    pub kind: Option<String>,
35    /// The producing agent, from `gen_ai.agent.id`.
36    #[serde(default, skip_serializing_if = "Option::is_none")]
37    pub agent: Option<String>,
38    /// The user scope layer, from `agdx.mem.user`.
39    #[serde(default, skip_serializing_if = "Option::is_none")]
40    pub user: Option<String>,
41    /// The app scope layer, from `agdx.mem.app`.
42    #[serde(default, skip_serializing_if = "Option::is_none")]
43    pub app: Option<String>,
44    /// The conversation that wrote the record, from `gen_ai.conversation.id`.
45    #[serde(default, skip_serializing_if = "Option::is_none")]
46    pub conversation: Option<String>,
47    /// The origin log record this memory item was folded from (stream, topic,
48    /// partition, offset), so a reader can navigate back to the source message
49    /// while it is still on the log. Absent when unknown.
50    #[serde(default, skip_serializing_if = "Option::is_none")]
51    pub source: Option<crate::graph::SourceRef>,
52}
53
54/// One stored entry: an arbitrary-bytes key and value plus optional expiry. The
55/// key and value are owned `Vec<u8>`, so the public API never leaks the `bytes`
56/// crate, and they ride the wire as CBOR byte strings, byte-exact.
57#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
58pub struct KvEntry {
59    #[serde(with = "crate::encoding::bin_bytes")]
60    pub key: Vec<u8>,
61    #[serde(with = "crate::encoding::bin_bytes")]
62    pub value: Vec<u8>,
63    /// Absolute expiry in epoch microseconds, or `None` for no expiry. Expired
64    /// entries are hidden on read.
65    #[serde(default, skip_serializing_if = "Option::is_none")]
66    pub expires_at_micros: Option<u64>,
67    /// Optimistic-concurrency version, assigned by the store and bumped on every
68    /// successful mutation of this key. The token a [`KvCas`] precondition
69    /// matches against. A versioned managed backend reports `>= 1` for a live
70    /// entry and reserves `0` solely for "unversioned": a store that does not
71    /// track versions, or an entry written before versioning. A caller must
72    /// therefore never treat `0` as a valid compare token. The field is skipped
73    /// on the wire when `0` so those entries stay byte-identical.
74    #[serde(default, skip_serializing_if = "is_zero_u64")]
75    pub version: u64,
76    /// The memory scope, present only on a memory read-view row. Recall reads it
77    /// to rebuild the item and filter by scope. A generic entry leaves it `None`.
78    /// Boxed so a generic entry stays small (the scope is five optional strings).
79    #[serde(default, skip_serializing_if = "Option::is_none")]
80    pub scope: Option<Box<MemoryRowScope>>,
81}
82
83fn is_zero_u64(value: &u64) -> bool {
84    *value == 0
85}
86
87impl KvEntry {
88    /// The key decoded as UTF-8, or `None` when it is not valid UTF-8. Keys are
89    /// arbitrary bytes, so a binary key has no string form.
90    pub fn key_str(&self) -> Option<&str> {
91        std::str::from_utf8(&self.key).ok()
92    }
93}
94
95/// A page of scanned entries plus the cursor to resume after the last one.
96/// `cursor` is `None` when the scan reached the end.
97#[derive(Clone, Debug, Default, Serialize, Deserialize)]
98pub struct KvPage {
99    pub entries: Vec<KvEntry>,
100    #[serde(
101        default,
102        skip_serializing_if = "Option::is_none",
103        with = "crate::encoding::opt_bin_bytes"
104    )]
105    pub cursor: Option<Vec<u8>>,
106}
107
108/// Request to read the value at `key` in `namespace`. With `if_none_match` set to
109/// a version, the read returns [`KvOutcome::NotModified`] instead of the value
110/// when the live version matches (a conditional GET), so an up-to-date cache
111/// skips the body transfer.
112#[derive(Clone, Debug, Serialize, Deserialize)]
113pub struct KvGet {
114    pub v: u32,
115    pub namespace: String,
116    #[serde(with = "crate::encoding::bin_bytes")]
117    pub key: Vec<u8>,
118    #[serde(default, skip_serializing_if = "Option::is_none")]
119    pub if_none_match: Option<u64>,
120}
121
122/// Request to write `value` at `key` in `namespace`, with an optional expiry.
123#[derive(Clone, Debug, Serialize, Deserialize)]
124pub struct KvSet {
125    pub v: u32,
126    pub namespace: String,
127    #[serde(with = "crate::encoding::bin_bytes")]
128    pub key: Vec<u8>,
129    #[serde(with = "crate::encoding::bin_bytes")]
130    pub value: Vec<u8>,
131    #[serde(default, skip_serializing_if = "Option::is_none")]
132    pub expires_at_micros: Option<u64>,
133}
134
135/// The precondition a [`KvCas`] write must satisfy to apply. The compare half
136/// of compare-and-swap: lock-free optimistic concurrency for callers contending
137/// on one key.
138#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
139pub enum CasExpect {
140    /// Apply only if the key currently holds this exact [`KvEntry::version`].
141    Match(u64),
142    /// Apply only if the key does not currently exist (a create-if-absent).
143    Absent,
144}
145
146/// Compare-and-swap write at `key` in `namespace`: apply `value` (with optional
147/// expiry) only if `expect` holds, else fail with
148/// [`KvError::VersionConflict`]. The swap half of optimistic concurrency, paired
149/// with [`KvEntry::version`] as the compare token.
150#[derive(Clone, Debug, Serialize, Deserialize)]
151pub struct KvCas {
152    pub v: u32,
153    pub namespace: String,
154    #[serde(with = "crate::encoding::bin_bytes")]
155    pub key: Vec<u8>,
156    #[serde(with = "crate::encoding::bin_bytes")]
157    pub value: Vec<u8>,
158    #[serde(default, skip_serializing_if = "Option::is_none")]
159    pub expires_at_micros: Option<u64>,
160    pub expect: CasExpect,
161}
162
163/// Fenced compare-and-swap: the [`KvCas`] write and precondition, applied only
164/// while the task's fence sequence still equals `fence_token` (the at-most-one
165/// effective-writer gate). A failed fence maps to [`KvError::LeaseLost`], a failed
166/// precondition to [`KvError::VersionConflict`]. Additive over
167/// [`crate::codes::KV_OP_VERSION`] 1.
168#[derive(Clone, Debug, Serialize, Deserialize)]
169pub struct KvCasFenced {
170    pub v: u32,
171    pub namespace: String,
172    #[serde(with = "crate::encoding::bin_bytes")]
173    pub key: Vec<u8>,
174    #[serde(with = "crate::encoding::bin_bytes")]
175    pub value: Vec<u8>,
176    #[serde(default, skip_serializing_if = "Option::is_none")]
177    pub expires_at_micros: Option<u64>,
178    pub expect: CasExpect,
179    /// The task's fence-sequence key, in the reserved fence namespace the plane
180    /// owns.
181    #[serde(with = "crate::encoding::bin_bytes")]
182    pub fence_key: Vec<u8>,
183    /// The fencing token: the fence-sequence value the holder was granted.
184    /// Strictly monotonic per task, never the lease version.
185    pub fence_token: u64,
186}
187
188/// Request to remove `key` from `namespace`. With `if_match` set to a version,
189/// the delete applies only when the live version matches (a conditional delete),
190/// returning [`KvError::VersionConflict`] otherwise.
191#[derive(Clone, Debug, Serialize, Deserialize)]
192pub struct KvDelete {
193    pub v: u32,
194    pub namespace: String,
195    #[serde(with = "crate::encoding::bin_bytes")]
196    pub key: Vec<u8>,
197    #[serde(default, skip_serializing_if = "Option::is_none")]
198    pub if_match: Option<u64>,
199}
200
201/// Request to test presence and read metadata without the value. The cheap way
202/// to check a precondition before transferring a large value (the formal
203/// `EXISTS` primitive).
204#[derive(Clone, Debug, Serialize, Deserialize)]
205pub struct KvExists {
206    pub v: u32,
207    pub namespace: String,
208    #[serde(with = "crate::encoding::bin_bytes")]
209    pub key: Vec<u8>,
210}
211
212/// Request to set, refresh, or clear a key's expiry in place without rewriting
213/// its value. `expires_at_micros` of `None` clears the expiry (the formal
214/// `EXPIRE` primitive).
215#[derive(Clone, Debug, Serialize, Deserialize)]
216pub struct KvExpire {
217    pub v: u32,
218    pub namespace: String,
219    #[serde(with = "crate::encoding::bin_bytes")]
220    pub key: Vec<u8>,
221    #[serde(default, skip_serializing_if = "Option::is_none")]
222    pub expires_at_micros: Option<u64>,
223}
224
225/// Request to apply a merge patch to a structured value without transferring the
226/// whole object. `patch` is a codec-specific patch document (the `content_type`
227/// names the format). With `if_match` set, the patch applies only at that version
228/// (the formal `PATCH` primitive).
229#[derive(Clone, Debug, Serialize, Deserialize)]
230pub struct KvPatch {
231    pub v: u32,
232    pub namespace: String,
233    #[serde(with = "crate::encoding::bin_bytes")]
234    pub key: Vec<u8>,
235    #[serde(with = "crate::encoding::bin_bytes")]
236    pub patch: Vec<u8>,
237    #[serde(default, skip_serializing_if = "Option::is_none")]
238    pub if_match: Option<u64>,
239}
240
241/// Request to copy the value at `key` to `to_key` (optionally into another
242/// namespace via `to_namespace`) in a single backend transaction. `Committed`
243/// on success, `NotFound` when the source is absent. The destination is
244/// overwritten (a guarded copy composes `exists` + `cas` instead).
245#[derive(Clone, Debug, Serialize, Deserialize)]
246pub struct KvCopy {
247    pub v: u32,
248    pub namespace: String,
249    #[serde(with = "crate::encoding::bin_bytes")]
250    pub key: Vec<u8>,
251    /// The destination namespace. Absent means the source `namespace`.
252    #[serde(default, skip_serializing_if = "Option::is_none")]
253    pub to_namespace: Option<String>,
254    #[serde(with = "crate::encoding::bin_bytes")]
255    pub to_key: Vec<u8>,
256}
257
258/// Request to move the value at `key` to `to_key`: copy plus delete of the
259/// source, one backend transaction, the same outcomes as [`KvCopy`].
260#[derive(Clone, Debug, Serialize, Deserialize)]
261pub struct KvMove {
262    pub v: u32,
263    pub namespace: String,
264    #[serde(with = "crate::encoding::bin_bytes")]
265    pub key: Vec<u8>,
266    /// The destination namespace. Absent means the source `namespace`.
267    #[serde(default, skip_serializing_if = "Option::is_none")]
268    pub to_namespace: Option<String>,
269    #[serde(with = "crate::encoding::bin_bytes")]
270    pub to_key: Vec<u8>,
271}
272
273/// Request to acquire an advisory lease (a bounded-TTL distributed lock) on
274/// `key`. On success the holder gets a `lease_token` to present on protected
275/// mutations (the formal `LEASE` primitive). Built on compare-and-swap.
276#[derive(Clone, Debug, Serialize, Deserialize)]
277pub struct KvLease {
278    pub v: u32,
279    pub namespace: String,
280    #[serde(with = "crate::encoding::bin_bytes")]
281    pub key: Vec<u8>,
282    pub lease_ttl_micros: u64,
283}
284
285/// Request to release an advisory lease early, presenting the `lease_token` the
286/// grant returned (the formal `RELEASE` primitive).
287#[derive(Clone, Debug, Serialize, Deserialize)]
288pub struct KvRelease {
289    pub v: u32,
290    pub namespace: String,
291    #[serde(with = "crate::encoding::bin_bytes")]
292    pub key: Vec<u8>,
293    pub lease_token: u64,
294}
295
296/// One key's metadata without its value: version, expiry, and value size. The
297/// reply to [`KvExists`].
298#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
299pub struct KvMetadata {
300    pub version: u64,
301    #[serde(default, skip_serializing_if = "Option::is_none")]
302    pub expires_at_micros: Option<u64>,
303    pub size_bytes: usize,
304}
305
306/// Request to list every namespace that holds at least one entry for the caller.
307#[derive(Clone, Debug, Serialize, Deserialize)]
308pub struct KvNamespaces {
309    pub v: u32,
310}
311
312/// One namespace summary in a `Namespaces` reply.
313#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
314pub struct KvNamespaceInfo {
315    pub namespace: String,
316    pub entries: usize,
317}
318
319/// Request to list entries in `namespace`. With no bounds it lists the whole
320/// namespace. `prefix` matches keys that start with it in byte order. `start`
321/// and `end` bound an inclusive-start, exclusive-end key range. `key_contains`
322/// keeps only keys that are valid UTF-8 and contain the substring (binary keys
323/// are skipped). All bounds compose.
324#[derive(Clone, Debug, Serialize, Deserialize)]
325pub struct KvScan {
326    pub v: u32,
327    pub namespace: String,
328    #[serde(
329        default,
330        skip_serializing_if = "Option::is_none",
331        with = "crate::encoding::opt_bin_bytes"
332    )]
333    pub prefix: Option<Vec<u8>>,
334    #[serde(
335        default,
336        skip_serializing_if = "Option::is_none",
337        with = "crate::encoding::opt_bin_bytes"
338    )]
339    pub start: Option<Vec<u8>>,
340    #[serde(
341        default,
342        skip_serializing_if = "Option::is_none",
343        with = "crate::encoding::opt_bin_bytes"
344    )]
345    pub end: Option<Vec<u8>>,
346    #[serde(default, skip_serializing_if = "Option::is_none")]
347    pub key_contains: Option<String>,
348    /// The conversation lens: keep only rows the given conversation wrote (the
349    /// text form of its `gen_ai.conversation.id`). The memory read view stamps
350    /// this on each record it materializes, generic key-value rows leave it
351    /// unset, so a scan narrows to one conversation's memory. Absent on the wire
352    /// when unfiltered, so a pre-lens scan stays byte-identical.
353    #[serde(default, skip_serializing_if = "Option::is_none")]
354    pub conversation: Option<String>,
355    pub limit: usize,
356    #[serde(
357        default,
358        skip_serializing_if = "Option::is_none",
359        with = "crate::encoding::opt_bin_bytes"
360    )]
361    pub cursor: Option<Vec<u8>>,
362}
363
364/// Request to bulk-delete entries in `namespace` matching the same composed
365/// bounds as a scan (`prefix`/`start`/`end`/`key_contains`). With no bounds it
366/// clears the whole namespace. No `limit`/`cursor`, and expiry is ignored (a
367/// matching expired entry is removed too).
368#[derive(Clone, Debug, Serialize, Deserialize)]
369pub struct KvDeleteMany {
370    pub v: u32,
371    pub namespace: String,
372    #[serde(
373        default,
374        skip_serializing_if = "Option::is_none",
375        with = "crate::encoding::opt_bin_bytes"
376    )]
377    pub prefix: Option<Vec<u8>>,
378    #[serde(
379        default,
380        skip_serializing_if = "Option::is_none",
381        with = "crate::encoding::opt_bin_bytes"
382    )]
383    pub start: Option<Vec<u8>>,
384    #[serde(
385        default,
386        skip_serializing_if = "Option::is_none",
387        with = "crate::encoding::opt_bin_bytes"
388    )]
389    pub end: Option<Vec<u8>>,
390    #[serde(default, skip_serializing_if = "Option::is_none")]
391    pub key_contains: Option<String>,
392    /// The conversation lens over a bulk delete: clear only the rows a given
393    /// conversation wrote. See [`KvScan::conversation`]. Absent when unfiltered.
394    #[serde(default, skip_serializing_if = "Option::is_none")]
395    pub conversation: Option<String>,
396}
397
398/// The result of a key-value operation: `Ok` with the operation's outcome, or
399/// `Err` with a structured failure.
400#[derive(Clone, Debug, Serialize, Deserialize)]
401#[non_exhaustive]
402pub enum KvReply {
403    Ok(KvOutcome),
404    Err(KvError),
405}
406
407/// The successful outcome of a KV command, shaped per op.
408#[derive(Clone, Debug, Serialize, Deserialize)]
409#[non_exhaustive]
410pub enum KvOutcome {
411    /// `get`: the live entry, or `None` when absent or expired.
412    Value(Option<KvEntry>),
413    /// `set`: the write was applied.
414    Written,
415    /// `cas`: the compare-and-swap applied. Carries the entry's new version, so
416    /// the caller can chain a further conditional write without a re-read.
417    Committed { version: u64 },
418    /// `delete`: `true` when a live entry was removed, `false` when none existed.
419    Deleted(bool),
420    /// `delete_many`: the number of entries removed by a filtered bulk delete.
421    DeletedMany(usize),
422    /// `scan`: one page of entries.
423    Page(KvPage),
424    /// `namespaces`: every namespace holding at least one entry for the
425    /// caller with its entry count, sorted by name.
426    Namespaces(Vec<KvNamespaceInfo>),
427    /// `get` with `if_none_match`: the live version matched, so the body is
428    /// unchanged and not transferred (a conditional-GET hit).
429    NotModified,
430    /// `exists`: the key's metadata, or `None` when absent or expired.
431    Metadata(Option<KvMetadata>),
432    /// `expire` / `patch`: applied, carrying the entry's version. `expire` leaves
433    /// the version unchanged, `patch` bumps it.
434    Versioned { version: u64 },
435    /// `lease`: the lease was granted, carrying the fencing token and the granted
436    /// TTL (which the store may shorten from the request).
437    Leased {
438        lease_token: u64,
439        granted_ttl_micros: u64,
440    },
441    /// `release`: `true` when a held lease was released, `false` when none was
442    /// held (idempotent release).
443    Released(bool),
444}
445
446/// Why a key-value operation failed.
447#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
448#[non_exhaustive]
449pub enum KvError {
450    #[error("kv not supported: {0}")]
451    Unsupported(String),
452    #[error("invalid key: {0}")]
453    InvalidKey(String),
454    /// The request named a namespace that fails [`validate_namespace`].
455    #[error("invalid namespace: {0}")]
456    InvalidNamespace(String),
457    #[error("{what} is {size}B, exceeds cap {cap}B")]
458    TooLarge {
459        what: String,
460        size: usize,
461        cap: usize,
462    },
463    #[error("kv backend error: {0}")]
464    Backend(String),
465    #[error("unsupported kv op version (expected {expected}, got {got})")]
466    Version { expected: u32, got: u32 },
467    /// A [`KvCas`] precondition was not met. `current` is the key's present
468    /// version (`Some`) or `None` when the key does not exist, so the caller can
469    /// re-read and retry, or learn that an `Absent` precondition lost a race.
470    #[error("kv version conflict (current: {current:?})")]
471    VersionConflict { current: Option<u64> },
472    /// A held [`KvLease`] expired or was released, so its token no longer
473    /// protects mutations. The caller must re-acquire the lease.
474    #[error("kv lease lost")]
475    LeaseLost,
476    /// An in-place op (`expire`, `patch`) targeted a key that is absent or
477    /// expired (the formal `NOT_FOUND`).
478    #[error("kv key not found")]
479    NotFound,
480}
481
482#[cfg(all(test, feature = "cbor"))]
483mod tests {
484    use super::*;
485    use crate::codes::KV_OP_VERSION;
486    use crate::framing::{decode_named, encode_named};
487
488    #[test]
489    fn given_kv_delete_many_when_round_tripped_then_should_preserve_bounds() {
490        let request = KvDeleteMany {
491            v: KV_OP_VERSION,
492            namespace: "sessions".to_owned(),
493            prefix: Some(b"user:".to_vec()),
494            start: None,
495            end: None,
496            key_contains: Some("stale".to_owned()),
497            conversation: None,
498        };
499        let bytes = encode_named(&request).expect("serializes");
500        let back: KvDeleteMany = decode_named(&bytes).expect("deserializes");
501        assert_eq!(back.prefix, Some(b"user:".to_vec()));
502        assert_eq!(back.key_contains.as_deref(), Some("stale"));
503    }
504
505    #[test]
506    fn given_kv_deleted_many_reply_when_round_tripped_then_should_preserve_count() {
507        let reply = KvReply::Ok(KvOutcome::DeletedMany(7));
508        let bytes = encode_named(&reply).expect("serializes");
509        let back: KvReply = decode_named(&bytes).expect("deserializes");
510        match back {
511            KvReply::Ok(KvOutcome::DeletedMany(n)) => assert_eq!(n, 7),
512            other => panic!("expected DeletedMany, got {other:?}"),
513        }
514    }
515
516    #[test]
517    fn given_a_set_request_when_round_tripped_then_should_preserve_value_and_expiry() {
518        let request = KvSet {
519            v: KV_OP_VERSION,
520            namespace: "sessions".to_owned(),
521            key: b"user:42".to_vec(),
522            value: b"online".to_vec(),
523            expires_at_micros: Some(1_700_000_000_000_000),
524        };
525        let bytes = encode_named(&request).expect("the request serializes");
526        let back: KvSet = decode_named(&bytes).expect("the request deserializes");
527        assert_eq!(back.key, b"user:42");
528        assert_eq!(back.value, b"online");
529        assert_eq!(back.expires_at_micros, Some(1_700_000_000_000_000));
530    }
531
532    #[test]
533    fn given_a_binary_key_entry_when_round_tripped_then_should_preserve_raw_bytes() {
534        let reply = KvReply::Ok(KvOutcome::Value(Some(KvEntry {
535            key: vec![0xff, 0x00, 0xfe],
536            value: vec![0x00, 0x01, 0x02],
537            expires_at_micros: None,
538            version: 0,
539            scope: None,
540        })));
541        let bytes = encode_named(&reply).expect("the reply serializes");
542        let back: KvReply = decode_named(&bytes).expect("the reply deserializes");
543        let KvReply::Ok(KvOutcome::Value(Some(entry))) = back else {
544            panic!("expected an Ok(Value(Some)) reply");
545        };
546        assert_eq!(entry.key, vec![0xff, 0x00, 0xfe]);
547        assert_eq!(entry.key_str(), None, "non-UTF-8 key has no string form");
548        assert_eq!(entry.value, vec![0x00, 0x01, 0x02]);
549    }
550
551    #[test]
552    fn given_a_scan_page_when_round_tripped_then_should_preserve_cursor() {
553        let reply = KvReply::Ok(KvOutcome::Page(KvPage {
554            entries: vec![KvEntry {
555                key: b"a".to_vec(),
556                value: b"1".to_vec(),
557                expires_at_micros: None,
558                version: 0,
559                scope: None,
560            }],
561            cursor: Some(b"a".to_vec()),
562        }));
563        let bytes = encode_named(&reply).expect("serializes");
564        let back: KvReply = decode_named(&bytes).expect("deserializes");
565        let KvReply::Ok(KvOutcome::Page(page)) = back else {
566            panic!("expected an Ok(Page) reply");
567        };
568        assert_eq!(page.entries.len(), 1);
569        assert_eq!(page.entries[0].key_str(), Some("a"));
570        assert_eq!(page.cursor.as_deref(), Some(b"a".as_ref()));
571    }
572
573    #[test]
574    fn given_a_cas_request_when_round_tripped_then_should_preserve_the_precondition() {
575        for expect in [CasExpect::Match(7), CasExpect::Absent] {
576            let request = KvCas {
577                v: KV_OP_VERSION,
578                namespace: "counters".to_owned(),
579                key: b"hits".to_vec(),
580                value: b"42".to_vec(),
581                expires_at_micros: None,
582                expect,
583            };
584            let bytes = encode_named(&request).expect("serializes");
585            let back: KvCas = decode_named(&bytes).expect("deserializes");
586            assert_eq!(back.expect, expect);
587            assert_eq!(back.key, b"hits");
588        }
589    }
590
591    #[test]
592    fn given_a_committed_reply_when_round_tripped_then_should_preserve_the_version() {
593        let reply = KvReply::Ok(KvOutcome::Committed { version: 9 });
594        let bytes = encode_named(&reply).expect("serializes");
595        let back: KvReply = decode_named(&bytes).expect("deserializes");
596        match back {
597            KvReply::Ok(KvOutcome::Committed { version }) => assert_eq!(version, 9),
598            other => panic!("expected Committed, got {other:?}"),
599        }
600    }
601
602    #[test]
603    fn given_an_exists_metadata_reply_when_round_tripped_then_should_preserve_metadata() {
604        let reply = KvReply::Ok(KvOutcome::Metadata(Some(KvMetadata {
605            version: 4,
606            expires_at_micros: Some(1_700_000_000_000_000),
607            size_bytes: 128,
608        })));
609        let bytes = encode_named(&reply).expect("serializes");
610        let back: KvReply = decode_named(&bytes).expect("deserializes");
611        let KvReply::Ok(KvOutcome::Metadata(Some(meta))) = back else {
612            panic!("expected Ok(Metadata(Some))");
613        };
614        assert_eq!(meta.version, 4);
615        assert_eq!(meta.size_bytes, 128);
616    }
617
618    #[test]
619    fn given_a_patch_request_when_round_tripped_then_should_preserve_patch_and_precondition() {
620        let request = KvPatch {
621            v: KV_OP_VERSION,
622            namespace: "docs".to_owned(),
623            key: b"doc:1".to_vec(),
624            patch: br#"{"status":"closed"}"#.to_vec(),
625            if_match: Some(3),
626        };
627        let bytes = encode_named(&request).expect("serializes");
628        let back: KvPatch = decode_named(&bytes).expect("deserializes");
629        assert_eq!(back.patch, br#"{"status":"closed"}"#);
630        assert_eq!(back.if_match, Some(3));
631    }
632
633    #[test]
634    fn given_a_lease_reply_when_round_tripped_then_should_preserve_token_and_ttl() {
635        let reply = KvReply::Ok(KvOutcome::Leased {
636            lease_token: 77,
637            granted_ttl_micros: 30_000_000,
638        });
639        let bytes = encode_named(&reply).expect("serializes");
640        let back: KvReply = decode_named(&bytes).expect("deserializes");
641        match back {
642            KvReply::Ok(KvOutcome::Leased {
643                lease_token,
644                granted_ttl_micros,
645            }) => {
646                assert_eq!(lease_token, 77);
647                assert_eq!(granted_ttl_micros, 30_000_000);
648            }
649            other => panic!("expected Leased, got {other:?}"),
650        }
651    }
652
653    #[test]
654    fn given_a_conditional_get_when_round_tripped_then_should_preserve_if_none_match() {
655        let request = KvGet {
656            v: KV_OP_VERSION,
657            namespace: "sessions".to_owned(),
658            key: b"user:1".to_vec(),
659            if_none_match: Some(5),
660        };
661        let bytes = encode_named(&request).expect("serializes");
662        let back: KvGet = decode_named(&bytes).expect("deserializes");
663        assert_eq!(back.if_none_match, Some(5));
664        // A plain get omits the precondition on the wire, so the pre-conditional
665        // contract stays byte-identical.
666        let plain = KvGet {
667            if_none_match: None,
668            ..request
669        };
670        let json = serde_json::to_string(&plain).expect("json");
671        assert!(
672            !json.contains("if_none_match"),
673            "absent precondition omitted"
674        );
675    }
676
677    #[test]
678    fn given_a_version_conflict_when_round_tripped_then_should_preserve_the_current_version() {
679        for current in [Some(3u64), None] {
680            let reply = KvReply::Err(KvError::VersionConflict { current });
681            let bytes = encode_named(&reply).expect("serializes");
682            let back: KvReply = decode_named(&bytes).expect("deserializes");
683            match back {
684                KvReply::Err(KvError::VersionConflict { current: got }) => assert_eq!(got, current),
685                other => panic!("expected VersionConflict, got {other:?}"),
686            }
687        }
688    }
689
690    #[test]
691    fn given_a_versioned_entry_when_round_tripped_then_should_preserve_version_and_skip_zero() {
692        let entry = KvEntry {
693            key: b"k".to_vec(),
694            value: b"v".to_vec(),
695            expires_at_micros: None,
696            version: 5,
697            scope: None,
698        };
699        let bytes = encode_named(&entry).expect("serializes");
700        let back: KvEntry = decode_named(&bytes).expect("deserializes");
701        assert_eq!(back.version, 5);
702        // An unversioned entry (version 0) omits the field on the wire, so a
703        // pre-versioning store stays byte-identical.
704        let unversioned = KvEntry {
705            version: 0,
706            ..entry
707        };
708        let json = serde_json::to_string(&unversioned).expect("json");
709        assert!(
710            !json.contains("version"),
711            "version 0 must be omitted: {json}"
712        );
713    }
714
715    #[test]
716    fn given_a_scan_with_bounds_when_round_tripped_then_should_preserve_filters() {
717        let scan = KvScan {
718            v: KV_OP_VERSION,
719            namespace: "sessions".to_owned(),
720            prefix: Some(b"user:".to_vec()),
721            start: None,
722            end: None,
723            key_contains: Some("admin".to_owned()),
724            conversation: None,
725            limit: 50,
726            cursor: Some(b"user:9".to_vec()),
727        };
728        let bytes = encode_named(&scan).expect("serializes");
729        let back: KvScan = decode_named(&bytes).expect("deserializes");
730        assert_eq!(back.prefix.as_deref(), Some(b"user:".as_ref()));
731        assert_eq!(back.key_contains.as_deref(), Some("admin"));
732        assert_eq!(back.cursor.as_deref(), Some(b"user:9".as_ref()));
733    }
734}
laser_wire/kv.rs

laser_wire/
kv.rs
