dialcache/protocol/
frame.rs

1//! Version-1 frames and semantic read results.
2//!
3//! A frame is `[0x01] ++ big-endian u64 createdAtMs ++ [0x00 text | 0x01 binary]
4//! ++ payload`. [`decode_frame`] applies the protocol's classification
5//! precedence to one raw read; [`normalize_read_result`] is the trust boundary
6//! above wire decoding that every semantic adapter result passes through.
7
8use crate::codec::Payload;
9use crate::limits::{MAX_SAFE_INTEGER, MAX_SUPPORTED_DURATION_MS};
10use crate::remote::{Frame, MissReason, ReadResult};
11
12use super::text::replacement_utf8;
13
14/// Wire-level failures that are errors rather than misses.
15#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
16pub enum ProtocolError {
17    /// A frame's encoding tag is neither text (`0x00`) nor binary (`0x01`).
18    #[error("Invalid DialCache Redis payload encoding")]
19    PayloadEncoding,
20    /// The server accepted a command but replied outside the protocol; the
21    /// text says how, for example a watermark that is not a digit string.
22    #[error("Invalid DialCache Redis reply: {0}")]
23    InvalidReply(String),
24    /// A stamp is not a nonnegative safe integer, or an invalidation would
25    /// push its watermark past one.
26    #[error("DialCache timestamp must be a nonnegative safe integer")]
27    InvalidTimestamp,
28    /// A TTL is not positive, or a TTL or future buffer exceeds 365 days.
29    #[error("DialCache cache TTL must be positive and no greater than 365 days")]
30    InvalidDuration,
31    /// zstd or its configuration failed on the write side; the text is
32    /// zstd's error name or the rejected option.
33    #[error("DialCache compression failed: {0}")]
34    Compression(String),
35}
36
37const FRAME_VERSION: u8 = 1;
38const FRAME_HEADER_LEN: usize = 10;
39const ENCODING_TEXT: u8 = 0;
40const ENCODING_BINARY: u8 = 1;
41
42/// Encode a frame: version byte, big-endian uint64 stamp, encoding tag, payload.
43/// Text payloads receive replacement UTF-8 conversion first.
44pub fn encode_frame(frame: &Frame) -> Result<Vec<u8>, ProtocolError> {
45    if frame.created_at_ms > MAX_SAFE_INTEGER {
46        return Err(ProtocolError::InvalidTimestamp);
47    }
48    let mut out = Vec::with_capacity(FRAME_HEADER_LEN + frame.payload.len());
49    out.push(FRAME_VERSION);
50    out.extend_from_slice(&frame.created_at_ms.to_be_bytes());
51    if frame.payload.binary {
52        out.push(ENCODING_BINARY);
53        out.extend_from_slice(&frame.payload.bytes);
54    } else {
55        out.push(ENCODING_TEXT);
56        out.extend_from_slice(&replacement_utf8(&frame.payload.bytes));
57    }
58    Ok(out)
59}
60
61/// Parse a watermark bulk string. `None` input is absent (a valid zero
62/// baseline); `Some(Err)` is malformed; `Some(Ok(value))` is a valid stamp.
63///
64/// Only ASCII digits are accepted. Leading zeros are ignored even when the
65/// literal would overflow a machine integer; the remaining digits must fit
66/// the safe-integer domain.
67pub fn parse_watermark(raw: Option<&[u8]>) -> Option<Result<u64, ProtocolError>> {
68    let raw = raw?;
69    Some(parse_watermark_digits(raw))
70}
71
72fn parse_watermark_digits(raw: &[u8]) -> Result<u64, ProtocolError> {
73    let malformed =
74        || ProtocolError::InvalidReply("watermark is not a nonnegative safe integer".to_string());
75    if raw.is_empty() || !raw.iter().all(u8::is_ascii_digit) {
76        return Err(malformed());
77    }
78    let mut value: u64 = 0;
79    for digit in raw.iter().skip_while(|byte| **byte == b'0') {
80        value = value
81            .checked_mul(10)
82            .and_then(|scaled| scaled.checked_add(u64::from(digit - b'0')))
83            .ok_or_else(malformed)?;
84    }
85    if value > MAX_SAFE_INTEGER {
86        return Err(malformed());
87    }
88    Ok(value)
89}
90
91/// Classify one raw read with the protocol's precedence: absent value,
92/// unsupported frame, tracked marker validity and zero stamp, fence,
93/// payload encoding, then a hit.
94///
95/// `watermark` is consulted only when `tracked`. A miss carries the parsed
96/// fence whenever the marker was valid (including the absent baseline, which
97/// carries none); a malformed marker yields an unclassified miss without one.
98/// The stamp is returned with full `u64` precision; the safe-integer check
99/// belongs to [`normalize_read_result`].
100pub fn decode_frame(
101    raw: Option<&[u8]>,
102    tracked: bool,
103    watermark: Option<&[u8]>,
104) -> Result<ReadResult, ProtocolError> {
105    let (fence, valid_marker) = if tracked {
106        match parse_watermark(watermark) {
107            None => (None, true),
108            Some(Ok(value)) => (Some(value), true),
109            Some(Err(_)) => (None, false),
110        }
111    } else {
112        (None, true)
113    };
114    let miss = |reason: MissReason| {
115        Ok(ReadResult::Miss {
116            reason,
117            observed_watermark_ms: fence,
118        })
119    };
120
121    let Some(raw) = raw else {
122        return miss(MissReason::ValueAbsent);
123    };
124    if raw.len() < FRAME_HEADER_LEN || raw[0] != FRAME_VERSION {
125        return miss(MissReason::Unclassified);
126    }
127    let stamp = u64::from_be_bytes(
128        raw[1..9]
129            .try_into()
130            .expect("frame header holds eight stamp bytes"),
131    );
132    if tracked && (!valid_marker || stamp == 0) {
133        return miss(MissReason::Unclassified);
134    }
135    if let (true, Some(fence)) = (tracked, fence) {
136        if stamp <= fence {
137            return miss(MissReason::WatermarkFenced);
138        }
139    }
140    let payload = match raw[9] {
141        ENCODING_TEXT => Payload {
142            bytes: replacement_utf8(&raw[FRAME_HEADER_LEN..]),
143            binary: false,
144        },
145        ENCODING_BINARY => Payload::binary(&raw[FRAME_HEADER_LEN..]),
146        _ => return Err(ProtocolError::PayloadEncoding),
147    };
148    Ok(ReadResult::Hit(Frame {
149        created_at_ms: stamp,
150        payload,
151    }))
152}
153
154/// Interpret an untrusted JSON-shaped semantic reply from a custom adapter.
155/// Unknown shapes become unclassified misses; frame-shaped objects ignore
156/// stray miss metadata.
157///
158/// Only `"kind": "miss"` selects the miss branch. Its reason is parsed from
159/// `"reason"` (unknown strings become unclassified) and its
160/// `observedWatermarkMs` is kept only when it is a JSON number in the
161/// timestamp domain. Any other object is frame-shaped: `createdAtMs` must be a
162/// number in the timestamp domain, and a JSON string `payload` is text while
163/// anything else is an empty text payload. The result still needs
164/// [`normalize_read_result`] for the tracked-key fence rules.
165pub fn read_result_from_untrusted_json(value: &serde_json::Value) -> ReadResult {
166    let Some(object) = value.as_object() else {
167        return ReadResult::miss(MissReason::Unclassified);
168    };
169    if object.get("kind").and_then(serde_json::Value::as_str) == Some("miss") {
170        let reason = object
171            .get("reason")
172            .and_then(serde_json::Value::as_str)
173            .and_then(MissReason::parse)
174            .unwrap_or(MissReason::Unclassified);
175        let observed_watermark_ms = object
176            .get("observedWatermarkMs")
177            .and_then(serde_json::Value::as_f64)
178            .and_then(|number| validate_timestamp_ms(number).ok());
179        return ReadResult::Miss {
180            reason,
181            observed_watermark_ms,
182        };
183    }
184    let Some(created_at_ms) = object
185        .get("createdAtMs")
186        .and_then(serde_json::Value::as_f64)
187        .and_then(|number| validate_timestamp_ms(number).ok())
188    else {
189        return ReadResult::miss(MissReason::Unclassified);
190    };
191    let payload = match object.get("payload").and_then(serde_json::Value::as_str) {
192        Some(text) => Payload::text(text),
193        None => Payload::default(),
194    };
195    ReadResult::Hit(Frame {
196        created_at_ms,
197        payload,
198    })
199}
200
201/// The core trust boundary above wire decoding: drop fences on untracked
202/// keys or outside the safe domain, and demote a fenced claim without a fence.
203pub fn normalize_read_result(result: ReadResult, tracked: bool) -> ReadResult {
204    match result {
205        ReadResult::Miss {
206            reason,
207            observed_watermark_ms,
208        } => {
209            let observed_watermark_ms =
210                observed_watermark_ms.filter(|fence| tracked && *fence <= MAX_SAFE_INTEGER);
211            let reason = match reason {
212                MissReason::WatermarkFenced if observed_watermark_ms.is_none() => {
213                    MissReason::Unclassified
214                }
215                other => other,
216            };
217            ReadResult::Miss {
218                reason,
219                observed_watermark_ms,
220            }
221        }
222        ReadResult::Hit(frame) if frame.created_at_ms > MAX_SAFE_INTEGER => {
223            ReadResult::miss(MissReason::Unclassified)
224        }
225        hit @ ReadResult::Hit(_) => hit,
226    }
227}
228
229/// A nonnegative safe-integer timestamp.
230pub fn validate_timestamp_ms(value: f64) -> Result<u64, ProtocolError> {
231    if !value.is_finite()
232        || value < 0.0
233        || value.trunc() != value
234        || value > MAX_SAFE_INTEGER as f64
235    {
236        return Err(ProtocolError::InvalidTimestamp);
237    }
238    Ok(value as u64)
239}
240
241/// Round a write TTL up to whole milliseconds and check the 365-day domain.
242pub fn ceil_supported_cache_ttl_ms(value: f64) -> Result<u64, ProtocolError> {
243    let ceiled = value.ceil();
244    if !ceiled.is_finite() || ceiled <= 0.0 || ceiled > MAX_SUPPORTED_DURATION_MS as f64 {
245        return Err(ProtocolError::InvalidDuration);
246    }
247    Ok(ceiled as u64)
248}
249
250#[cfg(test)]
251mod tests {
252    use serde_json::json;
253
254    use super::*;
255
256    fn text_frame(stamp: u64, text: &str) -> Vec<u8> {
257        encode_frame(&Frame {
258            created_at_ms: stamp,
259            payload: Payload::text(text),
260        })
261        .expect("encodes")
262    }
263
264    #[test]
265    fn encode_layout_and_domain() {
266        assert_eq!(text_frame(1, "A"), [1, 0, 0, 0, 0, 0, 0, 0, 1, 0, b'A']);
267        let binary = Frame {
268            created_at_ms: 0x0102,
269            payload: Payload::binary(vec![0xFF]),
270        };
271        assert_eq!(
272            encode_frame(&binary).unwrap(),
273            [1, 0, 0, 0, 0, 0, 0, 1, 2, 1, 0xFF]
274        );
275        let unsafe_stamp = Frame {
276            created_at_ms: MAX_SAFE_INTEGER + 1,
277            payload: Payload::text(""),
278        };
279        assert_eq!(
280            encode_frame(&unsafe_stamp),
281            Err(ProtocolError::InvalidTimestamp)
282        );
283        // Ill-formed text is replaced at encode time; binary bytes stay exact.
284        let ill_formed = Frame {
285            created_at_ms: 1,
286            payload: Payload {
287                bytes: vec![0xE2, 0x82],
288                binary: false,
289            },
290        };
291        assert_eq!(
292            &encode_frame(&ill_formed).unwrap()[10..],
293            [0xEF, 0xBF, 0xBD]
294        );
295    }
296
297    #[test]
298    fn watermark_parsing() {
299        assert_eq!(parse_watermark(None), None);
300        assert_eq!(parse_watermark(Some(b"0")), Some(Ok(0)));
301        assert_eq!(
302            parse_watermark(Some(b"000000000000000000000000000042")),
303            Some(Ok(42))
304        );
305        assert_eq!(
306            parse_watermark(Some(b"9007199254740991")),
307            Some(Ok(MAX_SAFE_INTEGER))
308        );
309        for malformed in [
310            &b""[..],
311            b"9007199254740992",
312            b"99999999999999999999999",
313            b"-1",
314            b"1.0",
315            b" 1",
316            b"1a",
317            "١".as_bytes(),
318        ] {
319            assert!(
320                matches!(
321                    parse_watermark(Some(malformed)),
322                    Some(Err(ProtocolError::InvalidReply(_)))
323                ),
324                "{malformed:?}"
325            );
326        }
327    }
328
329    #[test]
330    fn decode_precedence() {
331        let frame = text_frame(1000, "cached");
332        let fenced = decode_frame(Some(&frame), true, Some(b"1000")).unwrap();
333        assert_eq!(
334            fenced,
335            ReadResult::Miss {
336                reason: MissReason::WatermarkFenced,
337                observed_watermark_ms: Some(1000)
338            }
339        );
340        let hit = decode_frame(Some(&frame), true, Some(b"999")).unwrap();
341        assert_eq!(
342            hit,
343            ReadResult::Hit(Frame {
344                created_at_ms: 1000,
345                payload: Payload::text("cached")
346            })
347        );
348        // Untracked reads never consult the marker.
349        assert_eq!(
350            decode_frame(Some(&frame), false, Some(b"garbage")).unwrap(),
351            hit
352        );
353        // Absent value carries a valid fence; a malformed marker drops it.
354        assert_eq!(
355            decode_frame(None, true, Some(b"7")).unwrap(),
356            ReadResult::Miss {
357                reason: MissReason::ValueAbsent,
358                observed_watermark_ms: Some(7)
359            }
360        );
361        assert_eq!(
362            decode_frame(None, true, Some(b"x")).unwrap(),
363            ReadResult::miss(MissReason::ValueAbsent)
364        );
365        // Short or unversioned frames are unclassified before the marker is examined.
366        assert_eq!(
367            decode_frame(Some(&frame[..9]), true, Some(b"5")).unwrap(),
368            ReadResult::Miss {
369                reason: MissReason::Unclassified,
370                observed_watermark_ms: Some(5)
371            }
372        );
373        assert_eq!(
374            decode_frame(Some(b""), false, None).unwrap(),
375            ReadResult::miss(MissReason::Unclassified)
376        );
377        // Zero stamps and malformed markers are unclassified only on tracked keys.
378        let zero = text_frame(0, "z");
379        assert_eq!(
380            decode_frame(Some(&zero), true, None).unwrap(),
381            ReadResult::miss(MissReason::Unclassified)
382        );
383        assert!(matches!(
384            decode_frame(Some(&zero), false, None).unwrap(),
385            ReadResult::Hit(_)
386        ));
387        assert_eq!(
388            decode_frame(Some(&frame), true, Some(b"")).unwrap(),
389            ReadResult::miss(MissReason::Unclassified)
390        );
391        // The fence check precedes the encoding tag; a fenced bad tag is a miss.
392        let mut bad_tag = frame.clone();
393        bad_tag[9] = 2;
394        assert_eq!(
395            decode_frame(Some(&bad_tag), false, None),
396            Err(ProtocolError::PayloadEncoding)
397        );
398        assert_eq!(
399            decode_frame(Some(&bad_tag), true, Some(b"1000")).unwrap(),
400            ReadResult::Miss {
401                reason: MissReason::WatermarkFenced,
402                observed_watermark_ms: Some(1000)
403            }
404        );
405        // Full u64 stamps survive decoding; normalization demotes them.
406        let mut huge = text_frame(1, "h");
407        huge[1..9].copy_from_slice(&u64::MAX.to_be_bytes());
408        let decoded = decode_frame(Some(&huge), false, None).unwrap();
409        assert!(matches!(
410            decoded,
411            ReadResult::Hit(Frame {
412                created_at_ms: u64::MAX,
413                ..
414            })
415        ));
416        assert_eq!(
417            normalize_read_result(decoded, false),
418            ReadResult::miss(MissReason::Unclassified)
419        );
420    }
421
422    #[test]
423    fn untrusted_json_shapes() {
424        assert_eq!(
425            read_result_from_untrusted_json(&json!(null)),
426            ReadResult::miss(MissReason::Unclassified)
427        );
428        assert_eq!(
429            read_result_from_untrusted_json(&json!([1])),
430            ReadResult::miss(MissReason::Unclassified)
431        );
432        assert_eq!(
433            read_result_from_untrusted_json(
434                &json!({"kind": "miss", "reason": "watermark_fenced", "observedWatermarkMs": 5})
435            ),
436            ReadResult::Miss {
437                reason: MissReason::WatermarkFenced,
438                observed_watermark_ms: Some(5)
439            }
440        );
441        assert_eq!(
442            read_result_from_untrusted_json(
443                &json!({"kind": "miss", "reason": "bogus", "observedWatermarkMs": "5"})
444            ),
445            ReadResult::miss(MissReason::Unclassified)
446        );
447        assert_eq!(
448            read_result_from_untrusted_json(
449                &json!({"kind": "miss", "reason": "expired", "observedWatermarkMs": 1.5})
450            ),
451            ReadResult::miss(MissReason::Expired)
452        );
453        // Frame-shaped objects ignore stray miss metadata and non-string payloads.
454        assert_eq!(
455            read_result_from_untrusted_json(
456                &json!({"kind": "hit", "createdAtMs": 3, "payload": "p", "reason": "expired"})
457            ),
458            ReadResult::Hit(Frame {
459                created_at_ms: 3,
460                payload: Payload::text("p")
461            })
462        );
463        assert_eq!(
464            read_result_from_untrusted_json(&json!({"createdAtMs": 3, "payload": [1, 2]})),
465            ReadResult::Hit(Frame {
466                created_at_ms: 3,
467                payload: Payload::default()
468            })
469        );
470        for bad in [
471            json!({"createdAtMs": -1}),
472            json!({"createdAtMs": "3"}),
473            json!({"payload": "p"}),
474            json!({"createdAtMs": 1.5}),
475        ] {
476            assert_eq!(
477                read_result_from_untrusted_json(&bad),
478                ReadResult::miss(MissReason::Unclassified),
479                "{bad}"
480            );
481        }
482    }
483
484    #[test]
485    fn normalization_rules() {
486        let fenced = ReadResult::Miss {
487            reason: MissReason::WatermarkFenced,
488            observed_watermark_ms: Some(9),
489        };
490        assert_eq!(normalize_read_result(fenced.clone(), true), fenced);
491        assert_eq!(
492            normalize_read_result(fenced, false),
493            ReadResult::miss(MissReason::Unclassified)
494        );
495        let unsafe_fence = ReadResult::Miss {
496            reason: MissReason::Expired,
497            observed_watermark_ms: Some(MAX_SAFE_INTEGER + 1),
498        };
499        assert_eq!(
500            normalize_read_result(unsafe_fence, true),
501            ReadResult::miss(MissReason::Expired)
502        );
503        let hit = ReadResult::Hit(Frame {
504            created_at_ms: MAX_SAFE_INTEGER,
505            payload: Payload::text("ok"),
506        });
507        assert_eq!(normalize_read_result(hit.clone(), true), hit);
508    }
509
510    #[test]
511    fn timestamp_and_duration_domains() {
512        assert_eq!(validate_timestamp_ms(0.0), Ok(0));
513        assert_eq!(
514            validate_timestamp_ms(MAX_SAFE_INTEGER as f64),
515            Ok(MAX_SAFE_INTEGER)
516        );
517        for bad in [
518            -1.0,
519            1.5,
520            9007199254740992.0,
521            f64::NAN,
522            f64::INFINITY,
523            f64::NEG_INFINITY,
524            -0.5,
525        ] {
526            assert_eq!(
527                validate_timestamp_ms(bad),
528                Err(ProtocolError::InvalidTimestamp),
529                "{bad}"
530            );
531        }
532        assert_eq!(validate_timestamp_ms(-0.0), Ok(0));
533        assert_eq!(ceil_supported_cache_ttl_ms(0.1), Ok(1));
534        assert_eq!(ceil_supported_cache_ttl_ms(1.1), Ok(2));
535        assert_eq!(
536            ceil_supported_cache_ttl_ms(31535999999.999),
537            Ok(MAX_SUPPORTED_DURATION_MS)
538        );
539        assert_eq!(
540            ceil_supported_cache_ttl_ms(MAX_SUPPORTED_DURATION_MS as f64),
541            Ok(MAX_SUPPORTED_DURATION_MS)
542        );
543        for bad in [
544            0.0,
545            -1.0,
546            31536000000.001,
547            f64::NAN,
548            f64::INFINITY,
549            f64::NEG_INFINITY,
550        ] {
551            assert_eq!(
552                ceil_supported_cache_ttl_ms(bad),
553                Err(ProtocolError::InvalidDuration),
554                "{bad}"
555            );
556        }
557    }
558}
dialcache/protocol/frame.rs

dialcache/protocol/
frame.rs
