laser_wire/
hello.rs

1use crate::destination::{BackendResourceId, FileFormat, TableFormat};
2use crate::error::InvalidError;
3use crate::query::{Consistency, SqlDialect};
4use crate::schema::LogicalTypeKind;
5use crate::topology::WireTopology;
6use crate::validate::Validate;
7use serde::{Deserialize, Serialize};
8use std::collections::BTreeSet;
9
10/// Capability feature bits advertised in [`OpVersions::features`]. Each constant
11/// names one managed sub-feature a server serves beyond the base surface, so a
12/// binary client feature-detects it (before attempting the op) the way the HTTP
13/// surface reads the boolean flags on `Capabilities`. Additive and pinned
14/// cross-repo: a new bit is set by a newer server and ignored by an older
15/// client (which simply does not light up that capability).
16pub mod feature {
17    /// The key-value store serves compare-and-swap (`AGDX_KV_CAS`).
18    pub const KV_CAS: u64 = 1 << 0;
19    /// The query surface honors `Consistency::ReadYourWrites`.
20    pub const READ_YOUR_WRITES: u64 = 1 << 1;
21    /// The query surface honors `Consistency::Strong`.
22    pub const STRONG_CONSISTENCY: u64 = 1 << 2;
23    /// The key-value store serves fenced compare-and-swap (`AGDX_KV_CAS_FENCED`).
24    pub const KV_CAS_FENCED: u64 = 1 << 3;
25    /// The plane serves the agent and workflow control band (`AGDX_AGENT_*`).
26    pub const AGENT_WORKFLOW: u64 = 1 << 4;
27    /// The query surface serves lexical relevance search (`Query.text`).
28    pub const KEYWORD_SEARCH: u64 = 1 << 5;
29    /// The deployment publishes the change feed (`ChangeRecord`s on the
30    /// changes topic) for bindings that opt into `notify`.
31    pub const WATCH: u64 = 1 << 6;
32    /// The streaming server serves the authorization control band (`AGDX_AUTHZ_*`).
33    pub const AUTHZ: u64 = 1 << 7;
34    /// The deployment serves destination and checkpoint lifecycle operations.
35    pub const DESTINATIONS: u64 = 1 << 8;
36    /// The key-value store serves the revocable fenced-lease contract at
37    /// `KV_LEASE_OP_VERSION`: holder-scoped acquire, renewal
38    /// (`AGDX_KV_LEASE_RENEW`), holder-and-fence-validated release, fenced
39    /// compare-and-swap requiring a live lease, and the barriered read
40    /// (`KvGet::min_position`). Subsumes [`KV_CAS_FENCED`]: a client must not
41    /// send the reshaped lease ops (or a barriered read) to a server without
42    /// this bit, which would silently decode them under the old contract.
43    pub const KV_FENCED_LEASES: u64 = 1 << 9;
44}
45
46/// The wire op versions a server accepts, one per surface, plus the capability
47/// feature bits it advertises. A pinned wire shape, mirrored by the HTTP
48/// capabilities `versions` block.
49#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
50#[non_exhaustive]
51pub struct OpVersions {
52    pub query: u32,
53    pub control: u32,
54    pub kv: u32,
55    pub fork: u32,
56    /// The agent envelope (AGDX) version LaserData Cloud consumes for its
57    /// conversation projections. `0` means "not advertised" and is skipped on
58    /// encode, so pre-AGDX hello frames stay byte-identical and decode unchanged.
59    #[serde(default, skip_serializing_if = "is_zero")]
60    pub agent: u32,
61    /// The knowledge-graph op version served. `0` means not served, skipped on
62    /// encode so a pre-graph hello frame stays byte-identical. Mirrors the
63    /// `managed_graph` HTTP capability flag. (Agentic memory rides this plus the
64    /// query surface, so it has no op version of its own.)
65    #[serde(default, skip_serializing_if = "is_zero")]
66    pub graph: u32,
67    /// Destination and checkpoint operation version. Zero means not served.
68    #[serde(default, skip_serializing_if = "is_zero")]
69    pub checkpoint: u32,
70    /// Capability feature bits (see [`feature`]): managed sub-features served
71    /// beyond the base surface (compare-and-swap, read-your-writes, strong
72    /// consistency). `0` (the default) is skipped on encode, so a pre-feature
73    /// hello reply stays byte-identical and an old client just sees no extra
74    /// capabilities.
75    #[serde(default, skip_serializing_if = "is_zero_u64")]
76    pub features: u64,
77}
78
79fn is_zero(value: &u32) -> bool {
80    *value == 0
81}
82
83fn is_zero_u64(value: &u64) -> bool {
84    *value == 0
85}
86
87impl OpVersions {
88    /// Versions per surface. The struct is `#[non_exhaustive]` (new surfaces
89    /// land without a breaking change), so this is the constructor.
90    pub fn new(query: u32, control: u32, kv: u32, fork: u32) -> Self {
91        Self {
92            query,
93            control,
94            kv,
95            fork,
96            agent: 0,
97            graph: 0,
98            checkpoint: 0,
99            features: 0,
100        }
101    }
102
103    /// Returns a copy advertising this agent-envelope (AGDX) version.
104    #[must_use]
105    pub fn with_agent(mut self, agent: u32) -> Self {
106        self.agent = agent;
107        self
108    }
109
110    /// Returns a copy advertising the knowledge-graph op version served.
111    #[must_use]
112    pub fn with_graph(mut self, graph: u32) -> Self {
113        self.graph = graph;
114        self
115    }
116
117    /// Returns a copy advertising the destination and checkpoint version.
118    #[must_use]
119    pub fn with_checkpoint(mut self, checkpoint: u32) -> Self {
120        self.checkpoint = checkpoint;
121        self
122    }
123
124    /// Returns a copy advertising the capability feature bits in `features`
125    /// (an OR of [`feature`] constants).
126    #[must_use]
127    pub fn with_features(mut self, features: u64) -> Self {
128        self.features = features;
129        self
130    }
131
132    /// Whether a [`feature`] bit (or set of bits) is advertised.
133    pub const fn has_feature(&self, bit: u64) -> bool {
134        self.features & bit == bit
135    }
136}
137
138/// Body of the `AGDX_HELLO` probe reply: the wire op versions the server (and
139/// its managed backend) accepts, mirroring the HTTP capabilities `versions`
140/// block. A pinned wire shape. Pre-versioned
141/// servers answer the probe with an empty body, which a client treats as "no
142/// versions advertised", never an error.
143#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
144#[non_exhaustive]
145pub struct HelloReply {
146    pub versions: OpVersions,
147}
148
149impl HelloReply {
150    /// Constructor for the non-exhaustive wire struct.
151    pub fn new(versions: OpVersions) -> Self {
152        Self { versions }
153    }
154}
155
156pub const BACKEND_DESCRIPTOR_VERSION: u32 = 1;
157
158/// One observed backend resource and its structured, secret-free capabilities.
159#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
160#[non_exhaustive]
161pub struct BackendDescriptor {
162    pub descriptor_version: u32,
163    pub resource_id: BackendResourceId,
164    pub mode: BackendMode,
165    pub label: String,
166    pub implementation: BackendImplementation,
167    pub observed_backend_generation: u64,
168    pub runtime_configuration_revision: u64,
169    pub desired_state: BackendDesiredState,
170    pub observed_state: BackendObservedState,
171    pub readiness: BackendReadiness,
172    pub materialization: Vec<MaterializationCapability>,
173    #[serde(default, skip_serializing_if = "Option::is_none")]
174    pub query: Option<QueryCapabilities>,
175    pub schema: SchemaCapabilities,
176    pub maintenance: MaintenanceCapabilities,
177    pub limits: BackendLimits,
178}
179
180impl BackendDescriptor {
181    pub fn new(
182        resource_id: BackendResourceId,
183        mode: BackendMode,
184        label: impl Into<String>,
185        implementation: BackendImplementation,
186        observed_backend_generation: u64,
187        runtime_configuration_revision: u64,
188    ) -> Self {
189        Self {
190            descriptor_version: BACKEND_DESCRIPTOR_VERSION,
191            resource_id,
192            mode,
193            label: label.into(),
194            implementation,
195            observed_backend_generation,
196            runtime_configuration_revision,
197            desired_state: BackendDesiredState::Disabled,
198            observed_state: BackendObservedState::Disabled,
199            readiness: BackendReadiness::not_ready(BackendReadinessCode::Disabled),
200            materialization: Vec::new(),
201            query: None,
202            schema: SchemaCapabilities::default(),
203            maintenance: MaintenanceCapabilities::default(),
204            limits: BackendLimits::default(),
205        }
206    }
207
208    #[must_use]
209    pub fn with_state(
210        mut self,
211        desired_state: BackendDesiredState,
212        observed_state: BackendObservedState,
213        readiness: BackendReadiness,
214    ) -> Self {
215        self.desired_state = desired_state;
216        self.observed_state = observed_state;
217        self.readiness = readiness;
218        self
219    }
220
221    #[must_use]
222    pub fn with_materialization(mut self, capabilities: Vec<MaterializationCapability>) -> Self {
223        self.materialization = capabilities;
224        self
225    }
226
227    #[must_use]
228    pub fn with_query(mut self, capabilities: QueryCapabilities) -> Self {
229        self.query = Some(capabilities);
230        self
231    }
232
233    #[must_use]
234    pub fn with_schema(mut self, capabilities: SchemaCapabilities) -> Self {
235        self.schema = capabilities;
236        self
237    }
238
239    #[must_use]
240    pub fn with_maintenance(mut self, capabilities: MaintenanceCapabilities) -> Self {
241        self.maintenance = capabilities;
242        self
243    }
244
245    #[must_use]
246    pub fn with_limits(mut self, limits: BackendLimits) -> Self {
247        self.limits = limits;
248        self
249    }
250}
251
252impl Validate for BackendDescriptor {
253    fn validate(&self) -> Result<(), InvalidError> {
254        if self.descriptor_version != BACKEND_DESCRIPTOR_VERSION {
255            return Err(InvalidError::new(format!(
256                "backend descriptor version must be {BACKEND_DESCRIPTOR_VERSION}"
257            )));
258        }
259        if self.resource_id.as_u128() == 0
260            || self.observed_backend_generation == 0
261            || self.runtime_configuration_revision == 0
262        {
263            return Err(InvalidError::new(
264                "backend identity and observed revisions must be nonzero",
265            ));
266        }
267        validate_descriptor_text("backend label", &self.label)?;
268        validate_descriptor_text("backend implementation kind", &self.implementation.kind)?;
269        validate_descriptor_text(
270            "backend implementation version",
271            &self.implementation.version,
272        )?;
273        self.readiness.validate()?;
274        if self.readiness.ready && self.observed_state != BackendObservedState::Ready {
275            return Err(InvalidError::new(
276                "ready backend must report the ready observed state",
277            ));
278        }
279        let mut materialization = BTreeSet::new();
280        for capability in &self.materialization {
281            let key = (capability.file_format as u8, capability.table_format as u8);
282            if !materialization.insert(key) {
283                return Err(InvalidError::new(
284                    "backend repeats a materialization capability",
285                ));
286            }
287        }
288        if let Some(query) = &self.query {
289            query.validate()?;
290        }
291        if self.readiness.ready {
292            if self.schema.logical_schema && self.limits.max_schema_fields == 0 {
293                return Err(InvalidError::new(
294                    "ready logical-schema backend must advertise a schema-field limit",
295                ));
296            }
297            if !self.materialization.is_empty() && self.limits.max_materialization_file_bytes == 0 {
298                return Err(InvalidError::new(
299                    "ready materialization backend must advertise a file-size limit",
300                ));
301            }
302            if self.query.is_some()
303                && (self.limits.max_query_rows == 0
304                    || self.limits.max_query_bytes == 0
305                    || self.limits.max_scan_bytes == 0
306                    || self.limits.max_query_micros == 0
307                    || self.limits.max_concurrent_queries == 0)
308            {
309                return Err(InvalidError::new(
310                    "ready query backend must advertise nonzero query limits",
311                ));
312            }
313        }
314        Ok(())
315    }
316}
317
318#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
319#[serde(rename_all = "snake_case")]
320#[non_exhaustive]
321pub enum BackendMode {
322    Operational,
323    Lakehouse,
324}
325
326#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
327pub struct BackendImplementation {
328    pub kind: String,
329    pub version: String,
330}
331
332#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
333#[serde(rename_all = "snake_case")]
334#[non_exhaustive]
335pub enum BackendDesiredState {
336    Disabled,
337    Enabled,
338}
339
340#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
341#[serde(rename_all = "snake_case")]
342#[non_exhaustive]
343pub enum BackendObservedState {
344    Disabled,
345    Starting,
346    Ready,
347    Degraded,
348    Unavailable,
349}
350
351#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
352pub struct BackendReadiness {
353    pub ready: bool,
354    pub reasons: Vec<BackendReadinessReason>,
355    pub observed_at_micros: u64,
356}
357
358impl BackendReadiness {
359    pub fn not_ready(code: BackendReadinessCode) -> Self {
360        Self {
361            ready: false,
362            reasons: vec![BackendReadinessReason { code, detail: None }],
363            observed_at_micros: 0,
364        }
365    }
366
367    pub fn ready(observed_at_micros: u64) -> Self {
368        Self {
369            ready: true,
370            reasons: Vec::new(),
371            observed_at_micros,
372        }
373    }
374}
375
376impl Validate for BackendReadiness {
377    fn validate(&self) -> Result<(), InvalidError> {
378        if self.ready {
379            if !self.reasons.is_empty() || self.observed_at_micros == 0 {
380                return Err(InvalidError::new(
381                    "ready backend requires an observation time and no failure reasons",
382                ));
383            }
384            return Ok(());
385        }
386        if self.reasons.is_empty() {
387            return Err(InvalidError::new(
388                "not-ready backend requires at least one stable reason",
389            ));
390        }
391        if self.reasons.len() > 32 {
392            return Err(InvalidError::new(
393                "backend readiness reason count exceeds cap 32",
394            ));
395        }
396        for reason in &self.reasons {
397            if let Some(detail) = &reason.detail {
398                validate_descriptor_text("backend readiness detail", detail)?;
399            }
400        }
401        Ok(())
402    }
403}
404
405#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
406pub struct BackendReadinessReason {
407    pub code: BackendReadinessCode,
408    #[serde(default, skip_serializing_if = "Option::is_none")]
409    pub detail: Option<String>,
410}
411
412#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
413#[serde(rename_all = "snake_case")]
414#[non_exhaustive]
415pub enum BackendReadinessCode {
416    Disabled,
417    ConfigurationPending,
418    ConfigurationRejected,
419    CredentialUnavailable,
420    ObjectStoreUnavailable,
421    CatalogUnavailable,
422    QueryRuntimeUnavailable,
423    GenerationMismatch,
424    ProbeFailed,
425}
426
427#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
428pub struct MaterializationCapability {
429    pub file_format: FileFormat,
430    pub table_format: TableFormat,
431    pub create_table: bool,
432    pub append: bool,
433}
434
435#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
436pub struct QueryCapabilities {
437    pub dialects: Vec<SqlDialect>,
438    pub time_travel: Vec<TimeTravelCapability>,
439    pub consistency: Vec<Consistency>,
440    pub logical_types: Vec<LogicalTypeKind>,
441    pub paging: Vec<QueryPagingCapability>,
442    pub cancellation: bool,
443    pub execution_status: bool,
444    pub raw_sql: bool,
445}
446
447impl Validate for QueryCapabilities {
448    fn validate(&self) -> Result<(), InvalidError> {
449        if self.dialects.is_empty()
450            || self.consistency.is_empty()
451            || self.logical_types.is_empty()
452            || self.paging.is_empty()
453        {
454            return Err(InvalidError::new(
455                "query capabilities require dialect, consistency, and logical-type coverage",
456            ));
457        }
458        ensure_unique(
459            "query dialect",
460            self.dialects.iter().map(|value| *value as u8),
461        )?;
462        ensure_unique(
463            "query time-travel capability",
464            self.time_travel.iter().map(|value| *value as u8),
465        )?;
466        ensure_unique(
467            "query consistency",
468            self.consistency.iter().map(|value| *value as u8),
469        )?;
470        ensure_unique(
471            "query logical type",
472            self.logical_types.iter().map(|value| *value as u8),
473        )?;
474        ensure_unique(
475            "query paging capability",
476            self.paging.iter().map(|value| *value as u8),
477        )?;
478        if self.raw_sql && self.dialects.is_empty() {
479            return Err(InvalidError::new(
480                "raw SQL capability requires an explicit dialect",
481            ));
482        }
483        Ok(())
484    }
485}
486
487fn ensure_unique(label: &str, values: impl IntoIterator<Item = u8>) -> Result<(), InvalidError> {
488    let mut seen = BTreeSet::new();
489    if values.into_iter().all(|value| seen.insert(value)) {
490        Ok(())
491    } else {
492        Err(InvalidError::new(format!("backend repeats a {label}")))
493    }
494}
495
496fn validate_descriptor_text(label: &str, value: &str) -> Result<(), InvalidError> {
497    if value.is_empty() || value.len() > 4096 || value.chars().any(char::is_control) {
498        return Err(InvalidError::new(format!(
499            "{label} must contain 1..=4096 bytes without control characters"
500        )));
501    }
502    Ok(())
503}
504
505#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
506#[serde(rename_all = "snake_case")]
507#[non_exhaustive]
508pub enum TimeTravelCapability {
509    SnapshotId,
510    TimestampMicros,
511}
512
513#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
514#[serde(rename_all = "snake_case")]
515#[non_exhaustive]
516pub enum QueryPagingCapability {
517    Offset,
518    Cursor,
519}
520
521#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
522pub struct SchemaCapabilities {
523    pub logical_schema: bool,
524    pub arrow_ipc_stream: bool,
525    pub schema_evolution: bool,
526}
527
528#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
529pub struct MaintenanceCapabilities {
530    pub expire_snapshots: bool,
531    pub remove_orphan_files: bool,
532    pub compact_data_files: bool,
533}
534
535#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
536pub struct BackendLimits {
537    pub max_query_rows: u64,
538    pub max_query_bytes: u64,
539    pub max_scan_bytes: u64,
540    pub max_query_micros: u64,
541    pub max_concurrent_queries: u32,
542    pub max_schema_fields: u32,
543    pub max_materialization_file_bytes: u64,
544}
545
546/// The managed backend's capability announcement to the streaming server, sent over their
547/// private socket on connect (`AGDX_BACKEND_HELLO_CODE`). The streaming server caches the
548/// `versions` and the advertised `backends`, and relays them verbatim when it answers a
549/// client `AGDX_HELLO` / capabilities probe, so the streaming server never hardcodes feature
550/// bits or backend identities the backend may or may not serve.
551/// This makes the backend the single source of its own capability truth and
552/// keeps the binary `features` bitset and the HTTP capability flags in agreement
553/// with what is actually served. A separate type from [`HelloReply`] because the
554/// direction and sender differ (backend to streaming server, not server to client).
555const fn backend_ready_by_default() -> bool {
556    true
557}
558
559const fn backend_is_ready(ready: &bool) -> bool {
560    *ready
561}
562
563#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
564#[non_exhaustive]
565pub struct BackendAnnounce {
566    pub versions: OpVersions,
567    #[serde(
568        default = "backend_ready_by_default",
569        skip_serializing_if = "backend_is_ready"
570    )]
571    pub ready: bool,
572    /// Materialization backends the server currently exposes (the ones it has
573    /// open). A client routes only to an advertised id. Empty (the default) is
574    /// skipped on encode, so a pre-backends announce stays byte-identical and an
575    /// older reader simply sees no advertised backends.
576    #[serde(default, skip_serializing_if = "Vec::is_empty")]
577    pub backends: Vec<BackendDescriptor>,
578    /// The stream/topic names this deployment uses. Absent (the default) is
579    /// skipped on encode, so a pre-topology announce stays byte-identical and
580    /// an older reader sees no advertised topology (falls back to its
581    /// own [`WireTopology::default`]).
582    #[serde(default, skip_serializing_if = "Option::is_none")]
583    pub topology: Option<WireTopology>,
584}
585
586impl BackendAnnounce {
587    /// Constructor for the non-exhaustive wire struct.
588    pub fn new(versions: OpVersions) -> Self {
589        Self {
590            versions,
591            ready: true,
592            backends: Vec::new(),
593            topology: None,
594        }
595    }
596
597    #[must_use]
598    pub const fn unavailable(mut self) -> Self {
599        self.ready = false;
600        self
601    }
602
603    /// Returns a copy advertising `backends`.
604    #[must_use]
605    pub fn with_backends(mut self, backends: Vec<BackendDescriptor>) -> Self {
606        self.backends = backends;
607        self
608    }
609
610    /// Returns a copy advertising the deployment's `topology`.
611    #[must_use]
612    pub fn with_topology(mut self, topology: WireTopology) -> Self {
613        self.topology = Some(topology);
614        self
615    }
616}
617
618impl Validate for BackendAnnounce {
619    fn validate(&self) -> Result<(), InvalidError> {
620        let mut resources = BTreeSet::new();
621        for backend in &self.backends {
622            backend.validate()?;
623            if !resources.insert(backend.resource_id) {
624                return Err(InvalidError::new(
625                    "backend announcement repeats a resource id",
626                ));
627            }
628        }
629        if self.ready
630            && !self.backends.is_empty()
631            && !self.backends.iter().any(|backend| backend.readiness.ready)
632        {
633            return Err(InvalidError::new(
634                "ready backend announcement must contain a ready backend",
635            ));
636        }
637        Ok(())
638    }
639}
640
641#[cfg(all(test, feature = "cbor"))]
642mod tests {
643    use super::*;
644    use crate::codes::{CONTROL_OP_VERSION, FORK_OP_VERSION, KV_OP_VERSION, QUERY_OP_VERSION};
645    use crate::framing::{decode_named, encode_named};
646
647    fn backend(resource_id: u128, mode: BackendMode, label: &str, kind: &str) -> BackendDescriptor {
648        BackendDescriptor::new(
649            BackendResourceId::from_u128(resource_id),
650            mode,
651            label,
652            BackendImplementation {
653                kind: kind.to_owned(),
654                version: "1.0.0".to_owned(),
655            },
656            1,
657            1,
658        )
659    }
660
661    #[test]
662    fn given_a_hello_reply_when_round_tripped_then_should_preserve_versions() {
663        // The pinned `HelloReply` shape (CBOR named fields). The connect-time
664        // probe decodes exactly this shape.
665        let reply = HelloReply::new(OpVersions::new(
666            QUERY_OP_VERSION,
667            CONTROL_OP_VERSION,
668            KV_OP_VERSION,
669            FORK_OP_VERSION,
670        ));
671        let bytes = encode_named(&reply).expect("hello reply serializes");
672        let back: HelloReply = decode_named(&bytes).expect("hello reply deserializes");
673        assert_eq!(back, reply);
674    }
675
676    #[test]
677    fn given_a_backend_announce_when_round_tripped_then_should_preserve_features() {
678        let announce = BackendAnnounce::new(
679            OpVersions::new(
680                QUERY_OP_VERSION,
681                CONTROL_OP_VERSION,
682                KV_OP_VERSION,
683                FORK_OP_VERSION,
684            )
685            .with_features(feature::KV_CAS | feature::READ_YOUR_WRITES),
686        );
687        let bytes = encode_named(&announce).expect("serializes");
688        let back: BackendAnnounce = decode_named(&bytes).expect("deserializes");
689        assert_eq!(back, announce);
690        assert!(back.versions.has_feature(feature::KV_CAS));
691    }
692
693    #[test]
694    fn given_an_unavailable_backend_announce_when_round_tripped_then_should_stay_unavailable() {
695        let announce = BackendAnnounce::new(OpVersions::new(
696            QUERY_OP_VERSION,
697            CONTROL_OP_VERSION,
698            KV_OP_VERSION,
699            FORK_OP_VERSION,
700        ))
701        .unavailable();
702        let bytes = encode_named(&announce).expect("serializes");
703        let back: BackendAnnounce = decode_named(&bytes).expect("deserializes");
704        assert!(!back.ready);
705    }
706
707    #[test]
708    fn given_an_empty_hello_body_when_decoded_then_should_yield_no_versions() {
709        // Pre-versioned servers answer the probe with an empty body. The probe
710        // treats a failed decode as "no versions advertised", never an error.
711        assert!(decode_named::<HelloReply>(&[]).is_err());
712    }
713
714    #[test]
715    fn given_advertised_backends_when_round_tripped_then_should_preserve_them_and_skip_empty() {
716        let announce = BackendAnnounce::new(OpVersions::new(
717            QUERY_OP_VERSION,
718            CONTROL_OP_VERSION,
719            KV_OP_VERSION,
720            FORK_OP_VERSION,
721        ))
722        .with_backends(vec![
723            backend(1, BackendMode::Operational, "Embedded", "embedded"),
724            backend(2, BackendMode::Lakehouse, "Analytics warehouse", "columnar"),
725        ]);
726        let bytes = encode_named(&announce).expect("encodes");
727        let back: BackendAnnounce = decode_named(&bytes).expect("decodes");
728        assert_eq!(back, announce);
729        assert_eq!(back.backends.len(), 2);
730        assert_eq!(
731            back.backends[1].resource_id,
732            BackendResourceId::from_u128(2)
733        );
734        assert_eq!(back.backends[1].implementation.kind, "columnar");
735        assert_eq!(back.backends[1].label, "Analytics warehouse");
736
737        // No advertised backends (the default) is omitted on the wire, so a
738        // pre-backends announce stays byte-identical.
739        let plain = BackendAnnounce::new(OpVersions::new(1, 1, 1, 1));
740        let json = serde_json::to_string(&plain).expect("json");
741        assert!(!json.contains("backends"), "empty backends omitted: {json}");
742    }
743
744    #[test]
745    fn given_announce_without_topology_when_decoded_then_should_default_none() {
746        let announce = BackendAnnounce::new(OpVersions::new(
747            QUERY_OP_VERSION,
748            CONTROL_OP_VERSION,
749            KV_OP_VERSION,
750            FORK_OP_VERSION,
751        ));
752        let bytes = encode_named(&announce).expect("encodes");
753        let back: BackendAnnounce = decode_named(&bytes).expect("decodes");
754        assert_eq!(back.topology, None);
755        let json = serde_json::to_string(&announce).expect("json");
756        assert!(
757            !json.contains("topology"),
758            "absent topology omitted: {json}"
759        );
760    }
761
762    #[test]
763    fn given_announce_with_topology_when_round_tripped_then_should_preserve_it() {
764        let custom = WireTopology {
765            ops_stream: "custom-ops".to_owned(),
766            ..WireTopology::default()
767        };
768        let announce = BackendAnnounce::new(OpVersions::new(
769            QUERY_OP_VERSION,
770            CONTROL_OP_VERSION,
771            KV_OP_VERSION,
772            FORK_OP_VERSION,
773        ))
774        .with_topology(custom.clone());
775        let bytes = encode_named(&announce).expect("encodes");
776        let back: BackendAnnounce = decode_named(&bytes).expect("decodes");
777        assert_eq!(back.topology, Some(custom));
778    }
779
780    #[test]
781    fn given_advertised_features_when_round_tripped_then_should_preserve_bits_and_skip_zero() {
782        let versions = OpVersions::new(
783            QUERY_OP_VERSION,
784            CONTROL_OP_VERSION,
785            KV_OP_VERSION,
786            FORK_OP_VERSION,
787        )
788        .with_features(feature::KV_CAS | feature::READ_YOUR_WRITES);
789        assert!(versions.has_feature(feature::KV_CAS));
790        assert!(versions.has_feature(feature::READ_YOUR_WRITES));
791        assert!(!versions.has_feature(feature::STRONG_CONSISTENCY));
792        // has_feature on a combined mask requires every bit present.
793        assert!(versions.has_feature(feature::KV_CAS | feature::READ_YOUR_WRITES));
794        assert!(!versions.has_feature(feature::KV_CAS | feature::STRONG_CONSISTENCY));
795        let reply = HelloReply::new(versions);
796        let bytes = encode_named(&reply).expect("encodes");
797        let back: HelloReply = decode_named(&bytes).expect("decodes");
798        assert_eq!(back, reply);
799        assert!(back.versions.has_feature(feature::READ_YOUR_WRITES));
800        // No advertised feature (0) is omitted on the wire, so a pre-feature
801        // hello reply stays byte-identical.
802        let plain = HelloReply::new(OpVersions::new(1, 1, 1, 1));
803        let json = serde_json::to_string(&plain).expect("json");
804        assert!(!json.contains("features"), "zero features omitted: {json}");
805    }
806}
laser_wire/hello.rs

laser_wire/
hello.rs
