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 MIN_PROTOCOL_VERSION: u16 = 1;
16
17/// Highest supported external client protocol version.
18pub const PROTOCOL_VERSION: u16 = 2;
19
20/// First protocol version that carries the IMap/Fenced Lock operation family.
21pub const LOCK_PROTOCOL_VERSION: u16 = 2;
22
23/// Bytes used by the unsigned length prefix.
24pub const LENGTH_PREFIX_BYTES: usize = 4;
25
26/// Bytes used by the protocol-version field inside the frame body.
27pub const VERSION_BYTES: usize = 2;
28
29/// Smallest complete frame: length prefix plus version.
30pub const MIN_FRAME_BYTES: usize = LENGTH_PREFIX_BYTES + VERSION_BYTES;
31
32/// A length-prefixed external client frame.
33///
34/// The wire shape is:
35///
36/// ```text
37/// u32 body_len_be | u16 protocol_version_be | payload bytes
38/// ```
39///
40/// `body_len` includes the version field and the payload. Unknown future
41/// protocol versions are rejected loud, matching RULES R-3/R-4.
42#[derive(Debug, Clone, PartialEq, Eq)]
43pub struct ClientFrame {
44    protocol_version: u16,
45    payload: Bytes,
46}
47
48impl ClientFrame {
49    /// Build a frame at the highest supported protocol version.
50    pub fn new(payload: impl Into<Bytes>) -> Self {
51        Self {
52            protocol_version: PROTOCOL_VERSION,
53            payload: payload.into(),
54        }
55    }
56
57    /// Encode a typed wire message as this frame payload.
58    pub fn from_message(message: &ClientWireMessage) -> Result<Self, ClientProtocolError> {
59        let payload = postcard::to_allocvec(message)
60            .map_err(|error| ClientProtocolError::Codec(error.to_string()))?;
61        Ok(Self::new(payload))
62    }
63
64    /// Encode a typed wire message as this frame payload at an explicit version.
65    pub fn from_message_with_version(
66        protocol_version: u16,
67        message: &ClientWireMessage,
68    ) -> Result<Self, ClientProtocolError> {
69        let payload = postcard::to_allocvec(message)
70            .map_err(|error| ClientProtocolError::Codec(error.to_string()))?;
71        Ok(Self::with_version(protocol_version, payload))
72    }
73
74    /// Build a frame with an explicit protocol version for compatibility tests.
75    pub fn with_version(protocol_version: u16, payload: impl Into<Bytes>) -> Self {
76        Self {
77            protocol_version,
78            payload: payload.into(),
79        }
80    }
81
82    /// Return the frame protocol version.
83    pub fn protocol_version(&self) -> u16 {
84        self.protocol_version
85    }
86
87    /// Return the opaque payload bytes.
88    pub fn payload(&self) -> &Bytes {
89        &self.payload
90    }
91
92    /// Encode the frame with a big-endian length prefix.
93    pub fn encode(&self) -> Result<Bytes, ClientProtocolError> {
94        let body_len = VERSION_BYTES.checked_add(self.payload.len()).ok_or(
95            ClientProtocolError::FrameTooLarge {
96                actual: usize::MAX,
97                max: u32::MAX as usize,
98            },
99        )?;
100        if body_len > u32::MAX as usize {
101            return Err(ClientProtocolError::FrameTooLarge {
102                actual: body_len,
103                max: u32::MAX as usize,
104            });
105        }
106
107        let mut out = Vec::with_capacity(LENGTH_PREFIX_BYTES + body_len);
108        out.extend_from_slice(&(body_len as u32).to_be_bytes());
109        out.extend_from_slice(&self.protocol_version.to_be_bytes());
110        out.extend_from_slice(&self.payload);
111        Ok(Bytes::from(out))
112    }
113
114    /// Decode the frame payload as a typed wire message.
115    pub fn decode_message(&self) -> Result<ClientWireMessage, ClientProtocolError> {
116        postcard::from_bytes(self.payload.as_ref())
117            .map_err(|error| ClientProtocolError::Codec(error.to_string()))
118    }
119
120    /// Decode and validate a frame.
121    pub fn decode(bytes: &[u8], max_frame_bytes: usize) -> Result<Self, ClientProtocolError> {
122        if bytes.len() > max_frame_bytes {
123            return Err(ClientProtocolError::FrameTooLarge {
124                actual: bytes.len(),
125                max: max_frame_bytes,
126            });
127        }
128        if bytes.len() < MIN_FRAME_BYTES {
129            return Err(ClientProtocolError::TruncatedFrame {
130                actual: bytes.len(),
131                needed: MIN_FRAME_BYTES,
132            });
133        }
134
135        let body_len = u32::from_be_bytes(
136            bytes[0..LENGTH_PREFIX_BYTES]
137                .try_into()
138                .expect("slice length is checked"),
139        ) as usize;
140        if body_len < VERSION_BYTES {
141            return Err(ClientProtocolError::TruncatedFrame {
142                actual: body_len,
143                needed: VERSION_BYTES,
144            });
145        }
146
147        let expected = LENGTH_PREFIX_BYTES + body_len;
148        if expected != bytes.len() {
149            return Err(ClientProtocolError::LengthMismatch {
150                declared: body_len,
151                actual: bytes.len().saturating_sub(LENGTH_PREFIX_BYTES),
152            });
153        }
154
155        let version_start = LENGTH_PREFIX_BYTES;
156        let version_end = version_start + VERSION_BYTES;
157        let protocol_version = u16::from_be_bytes(
158            bytes[version_start..version_end]
159                .try_into()
160                .expect("slice length is checked"),
161        );
162        if !(MIN_PROTOCOL_VERSION..=PROTOCOL_VERSION).contains(&protocol_version) {
163            return Err(ClientProtocolError::UnsupportedVersion {
164                version: protocol_version,
165                supported_max: PROTOCOL_VERSION,
166            });
167        }
168
169        Ok(Self {
170            protocol_version,
171            payload: Bytes::copy_from_slice(&bytes[version_end..]),
172        })
173    }
174}
175
176/// Negotiated protocol support window.
177#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
178pub struct VersionHandshake {
179    /// Lowest protocol version supported by the caller.
180    pub min: u16,
181    /// Highest protocol version supported by the caller.
182    pub max: u16,
183}
184
185impl VersionHandshake {
186    /// Create a handshake range.
187    pub fn new(min: u16, max: u16) -> Self {
188        Self { min, max }
189    }
190
191    /// Negotiate the highest common version.
192    pub fn negotiate(self, server: VersionHandshake) -> Result<u16, ClientErrorEnvelope> {
193        let min = self.min.max(server.min);
194        let max = self.max.min(server.max);
195        if min <= max {
196            Ok(max)
197        } else {
198            Err(ClientErrorEnvelope::new(
199                ClientErrorCode::IncompatibleVersion,
200                false,
201                "no common HydraCache client protocol version",
202            ))
203        }
204    }
205}
206
207impl Default for VersionHandshake {
208    fn default() -> Self {
209        Self {
210            min: MIN_PROTOCOL_VERSION,
211            max: PROTOCOL_VERSION,
212        }
213    }
214}
215
216/// Return whether a request/response envelope version is in the supported window.
217pub fn protocol_version_supported(protocol_version: u16) -> bool {
218    (MIN_PROTOCOL_VERSION..=PROTOCOL_VERSION).contains(&protocol_version)
219}
220
221/// Reject an unsupported protocol version with the stable wire error.
222pub fn ensure_supported_protocol_version(protocol_version: u16) -> Result<(), ClientErrorEnvelope> {
223    if protocol_version_supported(protocol_version) {
224        Ok(())
225    } else {
226        Err(ClientErrorEnvelope::new(
227            ClientErrorCode::IncompatibleVersion,
228            false,
229            format!(
230                "unsupported HydraCache client protocol version {protocol_version}; supported range is {MIN_PROTOCOL_VERSION}..={PROTOCOL_VERSION}"
231            ),
232        ))
233    }
234}
235
236/// Reject an operation whose minimum version is newer than the negotiated version.
237pub fn require_protocol_version(
238    protocol_version: u16,
239    required_min: u16,
240    operation: &'static str,
241) -> Result<(), ClientErrorEnvelope> {
242    ensure_supported_protocol_version(protocol_version)?;
243    if protocol_version >= required_min {
244        Ok(())
245    } else {
246        Err(ClientErrorEnvelope::new(
247            ClientErrorCode::IncompatibleVersion,
248            false,
249            format!(
250                "{operation} requires HydraCache client protocol version {required_min} or newer"
251            ),
252        ))
253    }
254}
255
256/// Namespace carried on the wire.
257#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
258pub struct Namespace(String);
259
260impl Namespace {
261    /// Create a namespace.
262    pub fn new(value: impl Into<String>) -> Result<Self, ClientProtocolError> {
263        let value = value.into();
264        if value.trim().is_empty() {
265            return Err(ClientProtocolError::InvalidField("namespace"));
266        }
267        Ok(Self(value))
268    }
269
270    /// Return the namespace string.
271    pub fn as_str(&self) -> &str {
272        &self.0
273    }
274}
275
276/// Region id carried on the wire.
277#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
278pub struct RegionId(String);
279
280impl RegionId {
281    /// Create a region id.
282    pub fn new(value: impl Into<String>) -> Result<Self, ClientProtocolError> {
283        let value = value.into();
284        if value.trim().is_empty() {
285            return Err(ClientProtocolError::InvalidField("region"));
286        }
287        Ok(Self(value))
288    }
289
290    /// Return the region id string.
291    pub fn as_str(&self) -> &str {
292        &self.0
293    }
294}
295
296/// Structured cache key made of reviewable segments.
297#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
298pub struct StructuredKey {
299    segments: Vec<String>,
300}
301
302impl StructuredKey {
303    /// Create a structured key from segments.
304    pub fn new(segments: Vec<String>) -> Result<Self, ClientProtocolError> {
305        if segments.is_empty() || segments.iter().any(|segment| segment.trim().is_empty()) {
306            return Err(ClientProtocolError::InvalidField("key_segments"));
307        }
308        Ok(Self { segments })
309    }
310
311    /// Return the key segments.
312    pub fn segments(&self) -> &[String] {
313        &self.segments
314    }
315
316    /// Deterministic display form for local maps and diagnostics.
317    pub fn stable_key(&self) -> String {
318        self.segments.join(":")
319    }
320}
321
322/// Remote read consistency labels.
323#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
324#[serde(rename_all = "snake_case")]
325pub enum ReadConsistency {
326    /// Eventual read.
327    Eventual,
328    /// Strong read within the region.
329    Strong,
330    /// Session-aware read.
331    Session,
332}
333
334/// Remote write consistency labels.
335#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
336#[serde(rename_all = "snake_case")]
337pub enum WriteConsistency {
338    /// Local acknowledged write.
339    Local,
340    /// Quorum write.
341    Quorum,
342}
343
344/// Linearizable-capable consistency labels for lock/CAS operations.
345#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
346#[serde(rename_all = "snake_case")]
347pub enum LockConsistency {
348    /// A single replica; rejected for lock/CAS operations.
349    One,
350    /// Quorum-applied command.
351    #[default]
352    Quorum,
353    /// Each quorum-applied command.
354    EachQuorum,
355    /// All replicas applied the command.
356    All,
357}
358
359/// Expected value shape for single-key compare-and-set operations.
360#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
361#[serde(rename_all = "snake_case")]
362pub enum CasExpectation {
363    /// Match one exact current value.
364    Exact(Vec<u8>),
365    /// Match any live value, but fail when the key is absent/tombstoned.
366    Present,
367}
368
369/// Entry-event projection requested by Java/IMap-style listeners.
370#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
371#[serde(rename_all = "snake_case")]
372pub enum EntryEventProjection {
373    /// Plain near-cache invalidation signal.
374    Invalidation,
375    /// IMap entry-event shaped cache signal.
376    IMapEntryEvent,
377}
378
379/// Source signal that can be projected into an IMap entry event kind.
380#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
381#[serde(rename_all = "snake_case")]
382pub enum EntryEventSource {
383    /// A value was written, but the signal does not prove whether it was add or update.
384    Stored,
385    /// A value was explicitly removed or tombstoned.
386    Removed,
387    /// A key was invalidated without a stronger transition reason.
388    KeyInvalidated,
389    /// A tag invalidated one or more keys.
390    TagInvalidated,
391    /// A whole cache/namespace was flushed.
392    Flushed,
393    /// A value expired.
394    Expired,
395    /// A value was evicted.
396    Evicted,
397    /// A stale loader result was discarded.
398    StaleLoadDiscarded,
399    /// Unknown or transport-specific signal.
400    Unknown,
401}
402
403/// Entry-event kind exposed to Java/IMap-style listeners.
404#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
405#[serde(rename_all = "snake_case")]
406pub enum EntryEventKind {
407    /// A key changed, but the signal cannot distinguish add from update.
408    Upserted,
409    /// A key was removed/tombstoned.
410    Removed,
411    /// A key was evicted or expired.
412    Evicted,
413    /// A freshness invalidation without business-event semantics.
414    Invalidated,
415}
416
417impl EntryEventKind {
418    /// Conservatively project a cache/invalidation source into an entry-event kind.
419    pub const fn from_source(source: EntryEventSource) -> Self {
420        match source {
421            EntryEventSource::Stored => Self::Upserted,
422            EntryEventSource::Removed => Self::Removed,
423            EntryEventSource::Expired | EntryEventSource::Evicted => Self::Evicted,
424            EntryEventSource::KeyInvalidated
425            | EntryEventSource::TagInvalidated
426            | EntryEventSource::Flushed
427            | EntryEventSource::StaleLoadDiscarded
428            | EntryEventSource::Unknown => Self::Invalidated,
429        }
430    }
431}
432
433/// IMap entry-event shaped cache signal.
434#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
435pub struct EntryEvent {
436    /// Namespace.
437    pub ns: Namespace,
438    /// Key when the underlying signal is key-scoped.
439    pub key: Option<StructuredKey>,
440    /// Conservative event kind.
441    pub kind: EntryEventKind,
442    /// Optional value, gated by residency and transport support.
443    pub value: Option<Vec<u8>>,
444    /// Whether value inclusion was degraded by residency.
445    pub residency_degraded: bool,
446    /// Event watermark, if the source carries one.
447    pub watermark: Option<Watermark>,
448}
449
450/// Executable contract for W6 listener semantics.
451#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
452pub struct EntryListenerContract {
453    /// Signals may be coalesced and are not a complete event history.
454    pub coalesced: bool,
455    /// Delivery uses bounded buffers.
456    pub bounded_buffer: bool,
457    /// Slow listeners are dropped/reported through lag counters.
458    pub lag_drop_counter: bool,
459    /// This surface must not be used as a business event log.
460    pub business_event_log: bool,
461}
462
463impl EntryListenerContract {
464    /// Return the shipped W6 listener contract.
465    pub const fn cache_signal() -> Self {
466        Self {
467            coalesced: true,
468            bounded_buffer: true,
469            lag_drop_counter: true,
470            business_event_log: false,
471        }
472    }
473}
474
475/// Optional context carried by every request.
476#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
477pub struct ClientContext {
478    /// Opaque session token from 0.47 causal+.
479    pub session_token: Option<String>,
480    /// Requested read consistency.
481    pub read: Option<ReadConsistency>,
482    /// Requested write consistency.
483    pub write: Option<WriteConsistency>,
484    /// Preferred region for routing.
485    pub preferred_region: Option<RegionId>,
486}
487
488/// Watermark used by remote near-cache repair.
489#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
490pub struct Watermark {
491    /// B1 `last_uuid` / source generation.
492    pub source_generation: u64,
493    /// B1 `last_seq` / message id.
494    pub message_id: u64,
495}
496
497impl Watermark {
498    /// Create a watermark.
499    pub const fn new(source_generation: u64, message_id: u64) -> Self {
500        Self {
501            source_generation,
502            message_id,
503        }
504    }
505}
506
507/// Repair action selected for remote near-cache streams.
508#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
509#[serde(rename_all = "snake_case")]
510pub enum RepairAction {
511    /// Apply normally.
512    Apply,
513    /// Owner/source generation changed; clear the partition.
514    ClearPartition,
515    /// A sequence gap was observed; repair conservatively.
516    InvalidateConservatively,
517}
518
519/// Region-scoped subscription state.
520#[derive(Debug, Clone, Default, PartialEq, Eq)]
521pub struct SubscriptionWatermarkTracker {
522    last: Option<Watermark>,
523}
524
525impl SubscriptionWatermarkTracker {
526    /// Apply one event watermark and return the repair action.
527    pub fn on_event(&mut self, event: &InvalidationEvent) -> RepairAction {
528        let next = event.watermark();
529        let Some(last) = self.last else {
530            self.last = Some(next);
531            return RepairAction::ClearPartition;
532        };
533
534        if next.source_generation != last.source_generation {
535            self.last = Some(next);
536            return RepairAction::ClearPartition;
537        }
538        if next.message_id > last.message_id.saturating_add(1) {
539            self.last = Some(next);
540            return RepairAction::InvalidateConservatively;
541        }
542        self.last = Some(Watermark::new(
543            last.source_generation,
544            last.message_id.max(next.message_id),
545        ));
546        RepairAction::Apply
547    }
548}
549
550/// Client request envelope.
551#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
552pub struct ClientRequestEnvelope {
553    /// Stable request id.
554    pub request_id: String,
555    /// Negotiated protocol version.
556    pub protocol_version: u16,
557    /// Optional context.
558    pub context: ClientContext,
559    /// Deadline expressed as a logical millisecond timestamp for deterministic tests.
560    pub deadline_ms: Option<u64>,
561    /// Idempotency key for retry-safe writes.
562    pub idempotency_key: Option<String>,
563    /// Operation.
564    pub request: ClientRequest,
565}
566
567impl ClientRequestEnvelope {
568    /// Create an envelope for the highest supported protocol version.
569    pub fn new(request_id: impl Into<String>, request: ClientRequest) -> Self {
570        Self {
571            request_id: request_id.into(),
572            protocol_version: PROTOCOL_VERSION,
573            context: ClientContext::default(),
574            deadline_ms: None,
575            idempotency_key: None,
576            request,
577        }
578    }
579
580    /// Attach a context.
581    pub fn with_context(mut self, context: ClientContext) -> Self {
582        self.context = context;
583        self
584    }
585
586    /// Attach a deadline.
587    pub fn with_deadline_ms(mut self, deadline_ms: u64) -> Self {
588        self.deadline_ms = Some(deadline_ms);
589        self
590    }
591
592    /// Attach an idempotency key.
593    pub fn with_idempotency_key(mut self, idempotency_key: impl Into<String>) -> Self {
594        self.idempotency_key = Some(idempotency_key.into());
595        self
596    }
597
598    /// Return whether the deadline is expired at a logical timestamp.
599    pub fn deadline_expired(&self, now_ms: u64) -> bool {
600        self.deadline_ms.is_some_and(|deadline| deadline <= now_ms)
601    }
602
603    /// Validate the envelope version and operation minimum version.
604    pub fn validate_protocol(&self) -> Result<(), ClientErrorEnvelope> {
605        self.request.ensure_supported_by(self.protocol_version)
606    }
607}
608
609/// Client operations.
610#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
611#[serde(rename_all = "snake_case")]
612pub enum ClientRequest {
613    /// Read one key.
614    Get { ns: Namespace, key: StructuredKey },
615    /// Store one value.
616    Put {
617        ns: Namespace,
618        key: StructuredKey,
619        value: Vec<u8>,
620        ttl_ms: Option<u64>,
621        dimensions: Vec<String>,
622    },
623    /// Invalidate one key.
624    Invalidate { ns: Namespace, key: StructuredKey },
625    /// Read many keys.
626    BatchGet {
627        ns: Namespace,
628        keys: Vec<StructuredKey>,
629    },
630    /// Store many key/value pairs.
631    BatchPut {
632        ns: Namespace,
633        entries: Vec<BatchPutEntry>,
634    },
635    /// Evict a whole namespace/region mapping.
636    EvictRegion { ns: Namespace },
637    /// Subscribe to invalidations.
638    SubscribeInvalidations {
639        ns: Namespace,
640        region: Option<RegionId>,
641        from: Option<Watermark>,
642        include_value: bool,
643    },
644    /// Subscribe to IMap entry-event shaped cache signals.
645    SubscribeEntryEvents {
646        ns: Namespace,
647        region: Option<RegionId>,
648        from: Option<Watermark>,
649        include_value: bool,
650        projection: EntryEventProjection,
651    },
652    /// Try to acquire a session-bound fenced lock.
653    TryLock {
654        ns: Namespace,
655        key: StructuredKey,
656        lease_ms: u64,
657        wait_ms: u64,
658        level: LockConsistency,
659    },
660    /// Release a fenced lock with the current fence token.
661    Unlock {
662        ns: Namespace,
663        key: StructuredKey,
664        fence: u64,
665    },
666    /// Renew the lease for the current lock owner.
667    RenewLockLease {
668        ns: Namespace,
669        key: StructuredKey,
670        fence: u64,
671        lease_ms: u64,
672    },
673    /// Privileged fence-advancing release.
674    ForceUnlock { ns: Namespace, key: StructuredKey },
675    /// Read current lock ownership metadata.
676    GetLockOwnership { ns: Namespace, key: StructuredKey },
677    /// Single-key compare-and-set for IMap replace ergonomics.
678    CompareAndSet {
679        ns: Namespace,
680        key: StructuredKey,
681        expected: CasExpectation,
682        new_value: Vec<u8>,
683        level: LockConsistency,
684    },
685    /// Single-key conditional tombstone for IMap remove(key, value).
686    RemoveIfValue {
687        ns: Namespace,
688        key: StructuredKey,
689        expected: Vec<u8>,
690        level: LockConsistency,
691    },
692}
693
694impl ClientRequest {
695    /// Minimum protocol version required by this operation.
696    pub fn minimum_protocol_version(&self) -> u16 {
697        match self {
698            Self::Get { .. }
699            | Self::Put { .. }
700            | Self::Invalidate { .. }
701            | Self::BatchGet { .. }
702            | Self::BatchPut { .. }
703            | Self::EvictRegion { .. }
704            | Self::SubscribeInvalidations { .. } => MIN_PROTOCOL_VERSION,
705            Self::SubscribeEntryEvents { .. }
706            | Self::TryLock { .. }
707            | Self::Unlock { .. }
708            | Self::RenewLockLease { .. }
709            | Self::ForceUnlock { .. }
710            | Self::GetLockOwnership { .. }
711            | Self::CompareAndSet { .. }
712            | Self::RemoveIfValue { .. } => LOCK_PROTOCOL_VERSION,
713        }
714    }
715
716    /// Validate this operation against a negotiated protocol version.
717    pub fn ensure_supported_by(&self, protocol_version: u16) -> Result<(), ClientErrorEnvelope> {
718        require_protocol_version(
719            protocol_version,
720            self.minimum_protocol_version(),
721            self.operation_name(),
722        )
723    }
724
725    fn operation_name(&self) -> &'static str {
726        match self {
727            Self::Get { .. } => "get",
728            Self::Put { .. } => "put",
729            Self::Invalidate { .. } => "invalidate",
730            Self::BatchGet { .. } => "batch_get",
731            Self::BatchPut { .. } => "batch_put",
732            Self::EvictRegion { .. } => "evict_region",
733            Self::SubscribeInvalidations { .. } => "subscribe_invalidations",
734            Self::SubscribeEntryEvents { .. } => "subscribe_entry_events",
735            Self::TryLock { .. } => "try_lock",
736            Self::Unlock { .. } => "unlock",
737            Self::RenewLockLease { .. } => "renew_lock_lease",
738            Self::ForceUnlock { .. } => "force_unlock",
739            Self::GetLockOwnership { .. } => "get_lock_ownership",
740            Self::CompareAndSet { .. } => "compare_and_set",
741            Self::RemoveIfValue { .. } => "remove_if_value",
742        }
743    }
744}
745
746/// One batch put entry.
747#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
748pub struct BatchPutEntry {
749    /// Key to store.
750    pub key: StructuredKey,
751    /// Value bytes.
752    pub value: Vec<u8>,
753}
754
755/// Client response envelope.
756#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
757pub struct ClientResponseEnvelope {
758    /// Request id copied from the request.
759    pub request_id: String,
760    /// Protocol version used by the response.
761    pub protocol_version: u16,
762    /// Operation result.
763    pub result: Result<ClientResponse, ClientErrorEnvelope>,
764}
765
766impl ClientResponseEnvelope {
767    /// Build a successful response.
768    pub fn ok(request_id: impl Into<String>, response: ClientResponse) -> Self {
769        Self {
770            request_id: request_id.into(),
771            protocol_version: PROTOCOL_VERSION,
772            result: Ok(response),
773        }
774    }
775
776    /// Build an error response.
777    pub fn error(request_id: impl Into<String>, error: ClientErrorEnvelope) -> Self {
778        Self {
779            request_id: request_id.into(),
780            protocol_version: PROTOCOL_VERSION,
781            result: Err(error),
782        }
783    }
784
785    /// Return this response encoded for a negotiated protocol version.
786    pub fn with_protocol_version(mut self, protocol_version: u16) -> Self {
787        self.protocol_version = protocol_version;
788        self
789    }
790}
791
792/// Client responses.
793#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
794#[serde(rename_all = "snake_case")]
795pub enum ClientResponse {
796    /// Optional value.
797    Value { value: Option<Vec<u8>> },
798    /// Put accepted.
799    Stored,
800    /// Invalidation accepted.
801    Invalidated,
802    /// Batch result in request order.
803    Batch { items: Vec<BatchItemStatus> },
804    /// Region/namespace eviction accepted.
805    Evicted,
806    /// Subscription accepted.
807    Subscribed { from: Option<Watermark> },
808    /// Fenced lock acquired.
809    LockAcquired { fence: u64 },
810    /// Fenced lock is currently held by another owner.
811    LockBusy,
812    /// Fenced lock released.
813    LockReleased,
814    /// Fenced lock lease renewed.
815    LockLeaseRenewed,
816    /// Current lock ownership.
817    LockOwnership { fence: Option<u64>, locked: bool },
818    /// CAS applied and produced a new monotonic version.
819    CasApplied { new_version: u64 },
820    /// CAS did not apply; carries the current live value if present.
821    CasMismatch { current: Option<Vec<u8>> },
822}
823
824/// Per-item batch status.
825#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
826pub struct BatchItemStatus {
827    /// Original item index.
828    pub index: usize,
829    /// Per-item result.
830    pub result: Result<Option<Vec<u8>>, ClientErrorEnvelope>,
831}
832
833/// Stable error envelope.
834#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
835pub struct ClientErrorEnvelope {
836    /// Stable machine-readable error code.
837    pub code: ClientErrorCode,
838    /// Whether the SDK may retry.
839    pub retryable: bool,
840    /// Optional retry-after in milliseconds.
841    pub retry_after_ms: Option<u64>,
842    /// Redacted message for humans.
843    pub message: String,
844}
845
846impl ClientErrorEnvelope {
847    /// Create a redacted error envelope.
848    pub fn new(code: ClientErrorCode, retryable: bool, message: impl Into<String>) -> Self {
849        Self {
850            code,
851            retryable,
852            retry_after_ms: None,
853            message: redact_message(message.into()),
854        }
855    }
856
857    /// Attach retry-after.
858    pub fn with_retry_after_ms(mut self, retry_after_ms: u64) -> Self {
859        self.retry_after_ms = Some(retry_after_ms);
860        self
861    }
862}
863
864/// Stable client error codes.
865#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
866#[serde(rename_all = "snake_case")]
867pub enum ClientErrorCode {
868    /// No common supported protocol version.
869    IncompatibleVersion,
870    /// Identity is missing.
871    Unauthenticated,
872    /// Identity is not allowed.
873    Unauthorized,
874    /// Tenant quota exceeded.
875    TenantQuota,
876    /// Rate limited.
877    RateLimited,
878    /// Residency policy denied value movement.
879    ResidencyDenied,
880    /// Request or value too large.
881    TooLarge,
882    /// Deadline expired.
883    DeadlineExceeded,
884    /// Optimistic conflict.
885    Conflict,
886    /// Backend unavailable.
887    BackendUnavailable,
888    /// Frame or payload is malformed.
889    MalformedFrame,
890}
891
892/// Invalidation event streamed to remote near-caches.
893#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
894pub struct InvalidationEvent {
895    /// Namespace.
896    pub ns: Namespace,
897    /// Structured key.
898    pub key: StructuredKey,
899    /// B1 source generation.
900    pub generation: u64,
901    /// B1 message id.
902    pub message_id: u64,
903    /// Region where the event was applied.
904    pub applied_region: Option<RegionId>,
905    /// Optional value, gated by residency.
906    pub value: Option<Vec<u8>>,
907    /// Whether value was stripped by residency.
908    pub residency_degraded: bool,
909    /// Whether this event affects a subscriber's tracked cross-region view.
910    pub affects_subscriber_view: bool,
911}
912
913impl InvalidationEvent {
914    /// Create an invalidation event.
915    pub fn new(ns: Namespace, key: StructuredKey, generation: u64, message_id: u64) -> Self {
916        Self {
917            ns,
918            key,
919            generation,
920            message_id,
921            applied_region: None,
922            value: None,
923            residency_degraded: false,
924            affects_subscriber_view: false,
925        }
926    }
927
928    /// Attach applied region.
929    pub fn applied_in(mut self, region: RegionId) -> Self {
930        self.applied_region = Some(region);
931        self
932    }
933
934    /// Attach an optional value.
935    pub fn with_value(mut self, value: Vec<u8>) -> Self {
936        self.value = Some(value);
937        self
938    }
939
940    /// Mark that a cross-region invalidation affects the subscriber's tracked view.
941    pub fn affects_subscriber_view(mut self) -> Self {
942        self.affects_subscriber_view = true;
943        self
944    }
945
946    /// Return event watermark.
947    pub fn watermark(&self) -> Watermark {
948        Watermark::new(self.generation, self.message_id)
949    }
950
951    /// Return whether this event should be delivered for a region filter.
952    pub fn should_deliver_to(&self, region: Option<&RegionId>) -> bool {
953        match region {
954            None => true,
955            Some(region) => {
956                self.applied_region.as_ref() == Some(region) || self.affects_subscriber_view
957            }
958        }
959    }
960
961    /// Enforce residency for include-value streams.
962    pub fn residency_gated(mut self, value_allowed: bool) -> Self {
963        if !value_allowed && self.value.is_some() {
964            self.value = None;
965            self.residency_degraded = true;
966        }
967        self
968    }
969}
970
971impl EntryEvent {
972    /// Project a near-cache invalidation into an IMap-shaped entry-event signal.
973    pub fn from_invalidation(event: InvalidationEvent) -> Self {
974        let watermark = event.watermark();
975        Self {
976            ns: event.ns,
977            key: Some(event.key),
978            kind: EntryEventKind::Invalidated,
979            value: event.value,
980            residency_degraded: event.residency_degraded,
981            watermark: Some(watermark),
982        }
983    }
984
985    /// Build an entry event from a known cache signal source.
986    pub fn from_source(
987        ns: Namespace,
988        key: Option<StructuredKey>,
989        source: EntryEventSource,
990        value: Option<Vec<u8>>,
991        watermark: Option<Watermark>,
992    ) -> Self {
993        Self {
994            ns,
995            key,
996            kind: EntryEventKind::from_source(source),
997            value,
998            residency_degraded: false,
999            watermark,
1000        }
1001    }
1002}
1003
1004/// Wire messages carried inside [`ClientFrame`] payloads.
1005#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
1006#[serde(rename_all = "snake_case")]
1007pub enum ClientWireMessage {
1008    /// Version negotiation.
1009    Handshake(VersionHandshake),
1010    /// Client request.
1011    Request(ClientRequestEnvelope),
1012    /// Server response.
1013    Response(ClientResponseEnvelope),
1014    /// Server-pushed invalidation.
1015    Invalidation(InvalidationEvent),
1016    /// Stream heartbeat.
1017    Heartbeat(Watermark),
1018}
1019
1020impl ClientWireMessage {
1021    /// Return the protocol version that should be used for the outer frame.
1022    pub fn protocol_version(&self) -> u16 {
1023        match self {
1024            Self::Handshake(handshake) => handshake.max,
1025            Self::Request(envelope) => envelope.protocol_version,
1026            Self::Response(envelope) => envelope.protocol_version,
1027            Self::Invalidation(_) | Self::Heartbeat(_) => PROTOCOL_VERSION,
1028        }
1029    }
1030}
1031
1032fn redact_message(message: String) -> String {
1033    let mut redacted = message;
1034    for marker in ["value=", "secret=", "token="] {
1035        if let Some(index) = redacted.find(marker) {
1036            redacted.truncate(index + marker.len());
1037            redacted.push_str("<redacted>");
1038        }
1039    }
1040    redacted
1041}
1042
1043/// External client protocol decode/encode errors.
1044#[derive(Debug, Clone, PartialEq, Eq, Error)]
1045pub enum ClientProtocolError {
1046    /// Frame exceeds the configured limit.
1047    #[error("client frame is {actual} bytes, exceeding max_frame_bytes={max}")]
1048    FrameTooLarge {
1049        /// Observed frame length.
1050        actual: usize,
1051        /// Configured limit.
1052        max: usize,
1053    },
1054    /// Not enough bytes were supplied to parse a complete frame.
1055    #[error("truncated client frame: {actual} bytes available, {needed} needed")]
1056    TruncatedFrame {
1057        /// Observed frame length.
1058        actual: usize,
1059        /// Required frame length.
1060        needed: usize,
1061    },
1062    /// The length prefix and supplied bytes disagree.
1063    #[error(
1064        "client frame length mismatch: declared body {declared} bytes, actual body {actual} bytes"
1065    )]
1066    LengthMismatch {
1067        /// Body length from the prefix.
1068        declared: usize,
1069        /// Body length present after the prefix.
1070        actual: usize,
1071    },
1072    /// The frame is from a future protocol version.
1073    #[error("unsupported client protocol version {version}; supported max is {supported_max}")]
1074    UnsupportedVersion {
1075        /// Version from the frame.
1076        version: u16,
1077        /// Highest version this reader supports.
1078        supported_max: u16,
1079    },
1080    /// Payload codec failed.
1081    #[error("client protocol codec error: {0}")]
1082    Codec(String),
1083    /// Required field is invalid.
1084    #[error("invalid client protocol field: {0}")]
1085    InvalidField(&'static str),
1086}
hydracache_client_protocol/lib.rs

hydracache_client_protocol/
lib.rs
