hydracache_client_protocol/
lib.rs

1//! Stable external client protocol primitives.
2//!
3//! Release 0.49 starts the external-consumer surface by reserving a small,
4//! deterministic frame contract and golden fixtures. W1 expands the payload
5//! schema; W0 keeps the compatibility substrate intentionally narrow.
6
7use bytes::Bytes;
8use serde::{Deserialize, Serialize};
9use thiserror::Error;
10
11pub mod hibernate;
12pub mod java_migration;
13
14/// First supported external client protocol version.
15pub const PROTOCOL_VERSION: u16 = 1;
16
17/// Bytes used by the unsigned length prefix.
18pub const LENGTH_PREFIX_BYTES: usize = 4;
19
20/// Bytes used by the protocol-version field inside the frame body.
21pub const VERSION_BYTES: usize = 2;
22
23/// Smallest complete frame: length prefix plus version.
24pub const MIN_FRAME_BYTES: usize = LENGTH_PREFIX_BYTES + VERSION_BYTES;
25
26/// A length-prefixed external client frame.
27///
28/// The wire shape is:
29///
30/// ```text
31/// u32 body_len_be | u16 protocol_version_be | payload bytes
32/// ```
33///
34/// `body_len` includes the version field and the payload. Unknown future
35/// protocol versions are rejected loud, matching RULES R-3/R-4.
36#[derive(Debug, Clone, PartialEq, Eq)]
37pub struct ClientFrame {
38    protocol_version: u16,
39    payload: Bytes,
40}
41
42impl ClientFrame {
43    /// Build a v1 frame.
44    pub fn new(payload: impl Into<Bytes>) -> Self {
45        Self {
46            protocol_version: PROTOCOL_VERSION,
47            payload: payload.into(),
48        }
49    }
50
51    /// Encode a typed wire message as this frame payload.
52    pub fn from_message(message: &ClientWireMessage) -> Result<Self, ClientProtocolError> {
53        let payload = postcard::to_allocvec(message)
54            .map_err(|error| ClientProtocolError::Codec(error.to_string()))?;
55        Ok(Self::new(payload))
56    }
57
58    /// Build a frame with an explicit protocol version for compatibility tests.
59    pub fn with_version(protocol_version: u16, payload: impl Into<Bytes>) -> Self {
60        Self {
61            protocol_version,
62            payload: payload.into(),
63        }
64    }
65
66    /// Return the frame protocol version.
67    pub fn protocol_version(&self) -> u16 {
68        self.protocol_version
69    }
70
71    /// Return the opaque payload bytes.
72    pub fn payload(&self) -> &Bytes {
73        &self.payload
74    }
75
76    /// Encode the frame with a big-endian length prefix.
77    pub fn encode(&self) -> Result<Bytes, ClientProtocolError> {
78        let body_len = VERSION_BYTES.checked_add(self.payload.len()).ok_or(
79            ClientProtocolError::FrameTooLarge {
80                actual: usize::MAX,
81                max: u32::MAX as usize,
82            },
83        )?;
84        if body_len > u32::MAX as usize {
85            return Err(ClientProtocolError::FrameTooLarge {
86                actual: body_len,
87                max: u32::MAX as usize,
88            });
89        }
90
91        let mut out = Vec::with_capacity(LENGTH_PREFIX_BYTES + body_len);
92        out.extend_from_slice(&(body_len as u32).to_be_bytes());
93        out.extend_from_slice(&self.protocol_version.to_be_bytes());
94        out.extend_from_slice(&self.payload);
95        Ok(Bytes::from(out))
96    }
97
98    /// Decode the frame payload as a typed wire message.
99    pub fn decode_message(&self) -> Result<ClientWireMessage, ClientProtocolError> {
100        postcard::from_bytes(self.payload.as_ref())
101            .map_err(|error| ClientProtocolError::Codec(error.to_string()))
102    }
103
104    /// Decode and validate a frame.
105    pub fn decode(bytes: &[u8], max_frame_bytes: usize) -> Result<Self, ClientProtocolError> {
106        if bytes.len() > max_frame_bytes {
107            return Err(ClientProtocolError::FrameTooLarge {
108                actual: bytes.len(),
109                max: max_frame_bytes,
110            });
111        }
112        if bytes.len() < MIN_FRAME_BYTES {
113            return Err(ClientProtocolError::TruncatedFrame {
114                actual: bytes.len(),
115                needed: MIN_FRAME_BYTES,
116            });
117        }
118
119        let body_len = u32::from_be_bytes(
120            bytes[0..LENGTH_PREFIX_BYTES]
121                .try_into()
122                .expect("slice length is checked"),
123        ) as usize;
124        if body_len < VERSION_BYTES {
125            return Err(ClientProtocolError::TruncatedFrame {
126                actual: body_len,
127                needed: VERSION_BYTES,
128            });
129        }
130
131        let expected = LENGTH_PREFIX_BYTES + body_len;
132        if expected != bytes.len() {
133            return Err(ClientProtocolError::LengthMismatch {
134                declared: body_len,
135                actual: bytes.len().saturating_sub(LENGTH_PREFIX_BYTES),
136            });
137        }
138
139        let version_start = LENGTH_PREFIX_BYTES;
140        let version_end = version_start + VERSION_BYTES;
141        let protocol_version = u16::from_be_bytes(
142            bytes[version_start..version_end]
143                .try_into()
144                .expect("slice length is checked"),
145        );
146        if protocol_version > PROTOCOL_VERSION {
147            return Err(ClientProtocolError::UnsupportedVersion {
148                version: protocol_version,
149                supported_max: PROTOCOL_VERSION,
150            });
151        }
152
153        Ok(Self {
154            protocol_version,
155            payload: Bytes::copy_from_slice(&bytes[version_end..]),
156        })
157    }
158}
159
160/// Negotiated protocol support window.
161#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
162pub struct VersionHandshake {
163    /// Lowest protocol version supported by the caller.
164    pub min: u16,
165    /// Highest protocol version supported by the caller.
166    pub max: u16,
167}
168
169impl VersionHandshake {
170    /// Create a handshake range.
171    pub fn new(min: u16, max: u16) -> Self {
172        Self { min, max }
173    }
174
175    /// Negotiate the highest common version.
176    pub fn negotiate(self, server: VersionHandshake) -> Result<u16, ClientErrorEnvelope> {
177        let min = self.min.max(server.min);
178        let max = self.max.min(server.max);
179        if min <= max {
180            Ok(max)
181        } else {
182            Err(ClientErrorEnvelope::new(
183                ClientErrorCode::IncompatibleVersion,
184                false,
185                "no common HydraCache client protocol version",
186            ))
187        }
188    }
189}
190
191impl Default for VersionHandshake {
192    fn default() -> Self {
193        Self {
194            min: PROTOCOL_VERSION,
195            max: PROTOCOL_VERSION,
196        }
197    }
198}
199
200/// Namespace carried on the wire.
201#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
202pub struct Namespace(String);
203
204impl Namespace {
205    /// Create a namespace.
206    pub fn new(value: impl Into<String>) -> Result<Self, ClientProtocolError> {
207        let value = value.into();
208        if value.trim().is_empty() {
209            return Err(ClientProtocolError::InvalidField("namespace"));
210        }
211        Ok(Self(value))
212    }
213
214    /// Return the namespace string.
215    pub fn as_str(&self) -> &str {
216        &self.0
217    }
218}
219
220/// Region id carried on the wire.
221#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
222pub struct RegionId(String);
223
224impl RegionId {
225    /// Create a region id.
226    pub fn new(value: impl Into<String>) -> Result<Self, ClientProtocolError> {
227        let value = value.into();
228        if value.trim().is_empty() {
229            return Err(ClientProtocolError::InvalidField("region"));
230        }
231        Ok(Self(value))
232    }
233
234    /// Return the region id string.
235    pub fn as_str(&self) -> &str {
236        &self.0
237    }
238}
239
240/// Structured cache key made of reviewable segments.
241#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
242pub struct StructuredKey {
243    segments: Vec<String>,
244}
245
246impl StructuredKey {
247    /// Create a structured key from segments.
248    pub fn new(segments: Vec<String>) -> Result<Self, ClientProtocolError> {
249        if segments.is_empty() || segments.iter().any(|segment| segment.trim().is_empty()) {
250            return Err(ClientProtocolError::InvalidField("key_segments"));
251        }
252        Ok(Self { segments })
253    }
254
255    /// Return the key segments.
256    pub fn segments(&self) -> &[String] {
257        &self.segments
258    }
259
260    /// Deterministic display form for local maps and diagnostics.
261    pub fn stable_key(&self) -> String {
262        self.segments.join(":")
263    }
264}
265
266/// Remote read consistency labels.
267#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
268#[serde(rename_all = "snake_case")]
269pub enum ReadConsistency {
270    /// Eventual read.
271    Eventual,
272    /// Strong read within the region.
273    Strong,
274    /// Session-aware read.
275    Session,
276}
277
278/// Remote write consistency labels.
279#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
280#[serde(rename_all = "snake_case")]
281pub enum WriteConsistency {
282    /// Local acknowledged write.
283    Local,
284    /// Quorum write.
285    Quorum,
286}
287
288/// Optional context carried by every request.
289#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
290pub struct ClientContext {
291    /// Opaque session token from 0.47 causal+.
292    pub session_token: Option<String>,
293    /// Requested read consistency.
294    pub read: Option<ReadConsistency>,
295    /// Requested write consistency.
296    pub write: Option<WriteConsistency>,
297    /// Preferred region for routing.
298    pub preferred_region: Option<RegionId>,
299}
300
301/// Watermark used by remote near-cache repair.
302#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
303pub struct Watermark {
304    /// B1 `last_uuid` / source generation.
305    pub source_generation: u64,
306    /// B1 `last_seq` / message id.
307    pub message_id: u64,
308}
309
310impl Watermark {
311    /// Create a watermark.
312    pub const fn new(source_generation: u64, message_id: u64) -> Self {
313        Self {
314            source_generation,
315            message_id,
316        }
317    }
318}
319
320/// Repair action selected for remote near-cache streams.
321#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
322#[serde(rename_all = "snake_case")]
323pub enum RepairAction {
324    /// Apply normally.
325    Apply,
326    /// Owner/source generation changed; clear the partition.
327    ClearPartition,
328    /// A sequence gap was observed; repair conservatively.
329    InvalidateConservatively,
330}
331
332/// Region-scoped subscription state.
333#[derive(Debug, Clone, Default, PartialEq, Eq)]
334pub struct SubscriptionWatermarkTracker {
335    last: Option<Watermark>,
336}
337
338impl SubscriptionWatermarkTracker {
339    /// Apply one event watermark and return the repair action.
340    pub fn on_event(&mut self, event: &InvalidationEvent) -> RepairAction {
341        let next = event.watermark();
342        let Some(last) = self.last else {
343            self.last = Some(next);
344            return RepairAction::ClearPartition;
345        };
346
347        if next.source_generation != last.source_generation {
348            self.last = Some(next);
349            return RepairAction::ClearPartition;
350        }
351        if next.message_id > last.message_id.saturating_add(1) {
352            self.last = Some(next);
353            return RepairAction::InvalidateConservatively;
354        }
355        self.last = Some(Watermark::new(
356            last.source_generation,
357            last.message_id.max(next.message_id),
358        ));
359        RepairAction::Apply
360    }
361}
362
363/// Client request envelope.
364#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
365pub struct ClientRequestEnvelope {
366    /// Stable request id.
367    pub request_id: String,
368    /// Negotiated protocol version.
369    pub protocol_version: u16,
370    /// Optional context.
371    pub context: ClientContext,
372    /// Deadline expressed as a logical millisecond timestamp for deterministic tests.
373    pub deadline_ms: Option<u64>,
374    /// Idempotency key for retry-safe writes.
375    pub idempotency_key: Option<String>,
376    /// Operation.
377    pub request: ClientRequest,
378}
379
380impl ClientRequestEnvelope {
381    /// Create an envelope for v1.
382    pub fn new(request_id: impl Into<String>, request: ClientRequest) -> Self {
383        Self {
384            request_id: request_id.into(),
385            protocol_version: PROTOCOL_VERSION,
386            context: ClientContext::default(),
387            deadline_ms: None,
388            idempotency_key: None,
389            request,
390        }
391    }
392
393    /// Attach a context.
394    pub fn with_context(mut self, context: ClientContext) -> Self {
395        self.context = context;
396        self
397    }
398
399    /// Attach a deadline.
400    pub fn with_deadline_ms(mut self, deadline_ms: u64) -> Self {
401        self.deadline_ms = Some(deadline_ms);
402        self
403    }
404
405    /// Attach an idempotency key.
406    pub fn with_idempotency_key(mut self, idempotency_key: impl Into<String>) -> Self {
407        self.idempotency_key = Some(idempotency_key.into());
408        self
409    }
410
411    /// Return whether the deadline is expired at a logical timestamp.
412    pub fn deadline_expired(&self, now_ms: u64) -> bool {
413        self.deadline_ms.is_some_and(|deadline| deadline <= now_ms)
414    }
415}
416
417/// Client operations.
418#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
419#[serde(rename_all = "snake_case")]
420pub enum ClientRequest {
421    /// Read one key.
422    Get { ns: Namespace, key: StructuredKey },
423    /// Store one value.
424    Put {
425        ns: Namespace,
426        key: StructuredKey,
427        value: Vec<u8>,
428        ttl_ms: Option<u64>,
429        dimensions: Vec<String>,
430    },
431    /// Invalidate one key.
432    Invalidate { ns: Namespace, key: StructuredKey },
433    /// Read many keys.
434    BatchGet {
435        ns: Namespace,
436        keys: Vec<StructuredKey>,
437    },
438    /// Store many key/value pairs.
439    BatchPut {
440        ns: Namespace,
441        entries: Vec<BatchPutEntry>,
442    },
443    /// Evict a whole namespace/region mapping.
444    EvictRegion { ns: Namespace },
445    /// Subscribe to invalidations.
446    SubscribeInvalidations {
447        ns: Namespace,
448        region: Option<RegionId>,
449        from: Option<Watermark>,
450        include_value: bool,
451    },
452}
453
454/// One batch put entry.
455#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
456pub struct BatchPutEntry {
457    /// Key to store.
458    pub key: StructuredKey,
459    /// Value bytes.
460    pub value: Vec<u8>,
461}
462
463/// Client response envelope.
464#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
465pub struct ClientResponseEnvelope {
466    /// Request id copied from the request.
467    pub request_id: String,
468    /// Protocol version used by the response.
469    pub protocol_version: u16,
470    /// Operation result.
471    pub result: Result<ClientResponse, ClientErrorEnvelope>,
472}
473
474impl ClientResponseEnvelope {
475    /// Build a successful response.
476    pub fn ok(request_id: impl Into<String>, response: ClientResponse) -> Self {
477        Self {
478            request_id: request_id.into(),
479            protocol_version: PROTOCOL_VERSION,
480            result: Ok(response),
481        }
482    }
483
484    /// Build an error response.
485    pub fn error(request_id: impl Into<String>, error: ClientErrorEnvelope) -> Self {
486        Self {
487            request_id: request_id.into(),
488            protocol_version: PROTOCOL_VERSION,
489            result: Err(error),
490        }
491    }
492}
493
494/// Client responses.
495#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
496#[serde(rename_all = "snake_case")]
497pub enum ClientResponse {
498    /// Optional value.
499    Value { value: Option<Vec<u8>> },
500    /// Put accepted.
501    Stored,
502    /// Invalidation accepted.
503    Invalidated,
504    /// Batch result in request order.
505    Batch { items: Vec<BatchItemStatus> },
506    /// Region/namespace eviction accepted.
507    Evicted,
508    /// Subscription accepted.
509    Subscribed { from: Option<Watermark> },
510}
511
512/// Per-item batch status.
513#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
514pub struct BatchItemStatus {
515    /// Original item index.
516    pub index: usize,
517    /// Per-item result.
518    pub result: Result<Option<Vec<u8>>, ClientErrorEnvelope>,
519}
520
521/// Stable error envelope.
522#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
523pub struct ClientErrorEnvelope {
524    /// Stable machine-readable error code.
525    pub code: ClientErrorCode,
526    /// Whether the SDK may retry.
527    pub retryable: bool,
528    /// Optional retry-after in milliseconds.
529    pub retry_after_ms: Option<u64>,
530    /// Redacted message for humans.
531    pub message: String,
532}
533
534impl ClientErrorEnvelope {
535    /// Create a redacted error envelope.
536    pub fn new(code: ClientErrorCode, retryable: bool, message: impl Into<String>) -> Self {
537        Self {
538            code,
539            retryable,
540            retry_after_ms: None,
541            message: redact_message(message.into()),
542        }
543    }
544
545    /// Attach retry-after.
546    pub fn with_retry_after_ms(mut self, retry_after_ms: u64) -> Self {
547        self.retry_after_ms = Some(retry_after_ms);
548        self
549    }
550}
551
552/// Stable client error codes.
553#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
554#[serde(rename_all = "snake_case")]
555pub enum ClientErrorCode {
556    /// No common supported protocol version.
557    IncompatibleVersion,
558    /// Identity is missing.
559    Unauthenticated,
560    /// Identity is not allowed.
561    Unauthorized,
562    /// Tenant quota exceeded.
563    TenantQuota,
564    /// Rate limited.
565    RateLimited,
566    /// Residency policy denied value movement.
567    ResidencyDenied,
568    /// Request or value too large.
569    TooLarge,
570    /// Deadline expired.
571    DeadlineExceeded,
572    /// Optimistic conflict.
573    Conflict,
574    /// Backend unavailable.
575    BackendUnavailable,
576    /// Frame or payload is malformed.
577    MalformedFrame,
578}
579
580/// Invalidation event streamed to remote near-caches.
581#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
582pub struct InvalidationEvent {
583    /// Namespace.
584    pub ns: Namespace,
585    /// Structured key.
586    pub key: StructuredKey,
587    /// B1 source generation.
588    pub generation: u64,
589    /// B1 message id.
590    pub message_id: u64,
591    /// Region where the event was applied.
592    pub applied_region: Option<RegionId>,
593    /// Optional value, gated by residency.
594    pub value: Option<Vec<u8>>,
595    /// Whether value was stripped by residency.
596    pub residency_degraded: bool,
597    /// Whether this event affects a subscriber's tracked cross-region view.
598    pub affects_subscriber_view: bool,
599}
600
601impl InvalidationEvent {
602    /// Create an invalidation event.
603    pub fn new(ns: Namespace, key: StructuredKey, generation: u64, message_id: u64) -> Self {
604        Self {
605            ns,
606            key,
607            generation,
608            message_id,
609            applied_region: None,
610            value: None,
611            residency_degraded: false,
612            affects_subscriber_view: false,
613        }
614    }
615
616    /// Attach applied region.
617    pub fn applied_in(mut self, region: RegionId) -> Self {
618        self.applied_region = Some(region);
619        self
620    }
621
622    /// Attach an optional value.
623    pub fn with_value(mut self, value: Vec<u8>) -> Self {
624        self.value = Some(value);
625        self
626    }
627
628    /// Mark that a cross-region invalidation affects the subscriber's tracked view.
629    pub fn affects_subscriber_view(mut self) -> Self {
630        self.affects_subscriber_view = true;
631        self
632    }
633
634    /// Return event watermark.
635    pub fn watermark(&self) -> Watermark {
636        Watermark::new(self.generation, self.message_id)
637    }
638
639    /// Return whether this event should be delivered for a region filter.
640    pub fn should_deliver_to(&self, region: Option<&RegionId>) -> bool {
641        match region {
642            None => true,
643            Some(region) => {
644                self.applied_region.as_ref() == Some(region) || self.affects_subscriber_view
645            }
646        }
647    }
648
649    /// Enforce residency for include-value streams.
650    pub fn residency_gated(mut self, value_allowed: bool) -> Self {
651        if !value_allowed && self.value.is_some() {
652            self.value = None;
653            self.residency_degraded = true;
654        }
655        self
656    }
657}
658
659/// Wire messages carried inside [`ClientFrame`] payloads.
660#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
661#[serde(rename_all = "snake_case")]
662pub enum ClientWireMessage {
663    /// Version negotiation.
664    Handshake(VersionHandshake),
665    /// Client request.
666    Request(ClientRequestEnvelope),
667    /// Server response.
668    Response(ClientResponseEnvelope),
669    /// Server-pushed invalidation.
670    Invalidation(InvalidationEvent),
671    /// Stream heartbeat.
672    Heartbeat(Watermark),
673}
674
675fn redact_message(message: String) -> String {
676    let mut redacted = message;
677    for marker in ["value=", "secret=", "token="] {
678        if let Some(index) = redacted.find(marker) {
679            redacted.truncate(index + marker.len());
680            redacted.push_str("<redacted>");
681        }
682    }
683    redacted
684}
685
686/// External client protocol decode/encode errors.
687#[derive(Debug, Clone, PartialEq, Eq, Error)]
688pub enum ClientProtocolError {
689    /// Frame exceeds the configured limit.
690    #[error("client frame is {actual} bytes, exceeding max_frame_bytes={max}")]
691    FrameTooLarge {
692        /// Observed frame length.
693        actual: usize,
694        /// Configured limit.
695        max: usize,
696    },
697    /// Not enough bytes were supplied to parse a complete frame.
698    #[error("truncated client frame: {actual} bytes available, {needed} needed")]
699    TruncatedFrame {
700        /// Observed frame length.
701        actual: usize,
702        /// Required frame length.
703        needed: usize,
704    },
705    /// The length prefix and supplied bytes disagree.
706    #[error(
707        "client frame length mismatch: declared body {declared} bytes, actual body {actual} bytes"
708    )]
709    LengthMismatch {
710        /// Body length from the prefix.
711        declared: usize,
712        /// Body length present after the prefix.
713        actual: usize,
714    },
715    /// The frame is from a future protocol version.
716    #[error("unsupported client protocol version {version}; supported max is {supported_max}")]
717    UnsupportedVersion {
718        /// Version from the frame.
719        version: u16,
720        /// Highest version this reader supports.
721        supported_max: u16,
722    },
723    /// Payload codec failed.
724    #[error("client protocol codec error: {0}")]
725    Codec(String),
726    /// Required field is invalid.
727    #[error("invalid client protocol field: {0}")]
728    InvalidField(&'static str),
729}
hydracache_client_protocol/lib.rs

hydracache_client_protocol/
lib.rs
