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

laser_wire/
hello.rs
