laser_wire/
browse.rs

1use crate::control::{
2    GapAcceptance, OversizePolicy, Projection, ProjectionBinding, ProjectionHistory,
3    ProjectionKind, SchemaDef, SchemaSource, SourceSelector,
4};
5use crate::query::QueryError;
6use serde::{Deserialize, Serialize};
7
8/// A registered projection plus the bindings that route topics into it. The full
9/// picture of one materialized view: its extraction schema, expected content
10/// type, indexed fields, and where it applies.
11#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
12pub struct ProjectionInfo {
13    pub projection: Projection,
14    #[serde(default, skip_serializing_if = "Vec::is_empty")]
15    pub bindings: Vec<ProjectionBinding>,
16    /// How far each binding has materialized on the answering node, present
17    /// only when the request asked for it.
18    #[serde(default, skip_serializing_if = "Vec::is_empty")]
19    pub status: Vec<BindingStatus>,
20}
21
22/// One binding's materialization as the answering node sees it.
23#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
24pub struct BindingStatus {
25    #[serde(default, skip_serializing_if = "ProjectionHistory::is_append")]
26    pub history: ProjectionHistory,
27    pub source: SourceSelector,
28    pub index: String,
29    pub realization: RealizationState,
30    #[serde(default, skip_serializing_if = "Option::is_none")]
31    pub realization_reason: Option<String>,
32    /// The class of the last failed attempt to create the destination table,
33    /// present while realization is pending after a failure.
34    #[serde(default, skip_serializing_if = "Option::is_none")]
35    pub failure: Option<FailureClass>,
36    /// Why the binding makes no visible progress although no record holds it.
37    #[serde(default, skip_serializing_if = "Option::is_none")]
38    pub stall: Option<StallReason>,
39    /// Creation time of the source topic incarnation being materialized.
40    #[serde(default, skip_serializing_if = "Option::is_none")]
41    pub source_incarnation: Option<u64>,
42    #[serde(default, skip_serializing_if = "Vec::is_empty")]
43    pub partitions: Vec<PartitionProgress>,
44    /// Records or history gaps that stop a partition from advancing.
45    #[serde(default, skip_serializing_if = "Vec::is_empty")]
46    pub holds: Vec<Hold>,
47    /// Missing history an operator accepted for this source.
48    #[serde(default, skip_serializing_if = "Vec::is_empty")]
49    pub accepted_gaps: Vec<GapAcceptance>,
50    /// Records skipped from this binding's target and dead-lettered.
51    #[serde(default)]
52    pub skipped_records: u64,
53}
54
55/// Whether a binding's destination table exists.
56#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
57#[serde(rename_all = "snake_case")]
58#[non_exhaustive]
59pub enum RealizationState {
60    Pending,
61    /// Another node owns the shared destination and creates the table.
62    AwaitingOwner,
63    Realized,
64    #[serde(other)]
65    Unknown,
66}
67
68/// Why the destination table could not be created.
69#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
70#[serde(rename_all = "snake_case")]
71#[non_exhaustive]
72pub enum FailureClass {
73    /// The binding names a backend the server cannot resolve.
74    Configuration,
75    /// The destination refused the table definition.
76    Schema,
77    /// The destination refused the statement itself.
78    Poison,
79    /// The destination failed in a way no retry recovers.
80    Fatal,
81    /// The destination was briefly out of reach.
82    Transient,
83    /// The destination ran out of capacity.
84    Capacity,
85    /// The destination accepts no writes for now.
86    ReadOnly,
87    #[serde(other)]
88    Unknown,
89}
90
91/// What keeps a binding from visible progress when no record holds it.
92#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
93#[serde(rename_all = "snake_case")]
94#[non_exhaustive]
95pub enum StallReason {
96    /// The source topic does not exist.
97    SourceMissing,
98    /// The source partition heads could not be read.
99    SourceUnreadable,
100    /// The destination's progress could not be read.
101    DestinationUnreadable,
102    /// The destination refuses writes for now, so the projector waits.
103    Backpressure,
104    #[serde(other)]
105    Unknown,
106}
107
108/// One source partition's progress into a binding's destination.
109#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
110pub struct PartitionProgress {
111    pub partition_id: u32,
112    /// The last offset the source partition holds.
113    #[serde(default, skip_serializing_if = "Option::is_none")]
114    pub head_offset: Option<u64>,
115    /// The last offset the destination committed.
116    #[serde(default, skip_serializing_if = "Option::is_none")]
117    pub applied_offset: Option<u64>,
118    /// The first offset the binding materialized for this incarnation, when
119    /// the answering node observed it. Above zero, the history started late.
120    /// `None` means unknown, never complete.
121    #[serde(default, skip_serializing_if = "Option::is_none")]
122    pub first_offset: Option<u64>,
123}
124
125/// Why a partition does not advance.
126#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
127pub struct Hold {
128    pub partition_id: u32,
129    pub offset: u64,
130    pub cause: HoldCause,
131    pub reason: String,
132}
133
134/// What holds a partition.
135#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
136#[serde(rename_all = "snake_case")]
137#[non_exhaustive]
138pub enum HoldCause {
139    Extraction,
140    Schema,
141    Destination,
142    Oversize,
143    MissingHistory,
144    /// The binding was applied for an earlier incarnation of its source
145    /// stream. Apply it again to materialize the live one.
146    RetiredIncarnation,
147    #[serde(other)]
148    Unknown,
149}
150
151/// How far a materialization setup has come, weakest first.
152#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
153#[serde(rename_all = "snake_case")]
154#[non_exhaustive]
155pub enum MaterializationReadiness {
156    /// The definitions are durably appended.
157    Accepted,
158    /// The answering node applied the definitions.
159    Applied,
160    /// The destination table exists.
161    Realized,
162    /// The destination reached the source heads captured after apply.
163    CaughtUp,
164}
165
166impl BindingStatus {
167    /// The level this status proves on its own: `Realized` once the table
168    /// exists, else `Applied`. Catch-up needs a captured target, see
169    /// [`caught_up_to`](Self::caught_up_to).
170    pub fn readiness(&self) -> MaterializationReadiness {
171        if self.realization == RealizationState::Realized {
172            MaterializationReadiness::Realized
173        } else {
174            MaterializationReadiness::Applied
175        }
176    }
177
178    /// Whether this status reached every head `target` captured, on the same
179    /// source incarnation. A partition the target saw empty needs nothing.
180    pub fn caught_up_to(&self, target: &BindingStatus) -> bool {
181        self.realization == RealizationState::Realized
182            && !self.is_blocked()
183            && self.source_incarnation.is_some()
184            && self.source_incarnation == target.source_incarnation
185            && self.source == target.source
186            && self.index == target.index
187            && self.history == target.history
188            && target.partitions.iter().all(|wanted| {
189                wanted.head_offset.is_none_or(|head| {
190                    self.partitions.iter().any(|seen| {
191                        seen.partition_id == wanted.partition_id
192                            && seen.applied_offset.is_some_and(|applied| applied >= head)
193                    })
194                })
195            })
196    }
197
198    /// Whether a record or a history gap stops a partition.
199    pub fn is_blocked(&self) -> bool {
200        !self.holds.is_empty()
201    }
202
203    /// Whether the destination table failed in a way no retry recovers, so
204    /// the binding or its backend must change.
205    pub fn is_failed(&self) -> bool {
206        self.failure.is_some_and(FailureClass::is_permanent)
207    }
208
209    /// Whether the destination provably holds every record of the source
210    /// incarnation it has passed: every partition with progress started at
211    /// offset zero, no gap was accepted, no record was skipped and, under
212    /// `oversize`, no body is kept only by reference. An unknown first offset
213    /// is never complete, and a partition the binding never read is complete
214    /// only while no first offset above zero says its history expired unread.
215    pub fn history_complete(&self, oversize: OversizePolicy) -> bool {
216        self.history.is_append()
217            && self.realization == RealizationState::Realized
218            && self.source_incarnation.is_some()
219            && !self.is_blocked()
220            && oversize != OversizePolicy::Reference
221            && self.accepted_gaps.is_empty()
222            && self.skipped_records == 0
223            && self.partitions.iter().all(|partition| {
224                partition.first_offset == Some(0)
225                    || (partition.applied_offset.is_none() && partition.first_offset.is_none())
226            })
227    }
228}
229
230impl FailureClass {
231    /// Whether the failure lasts until the binding or its backend changes.
232    pub fn is_permanent(self) -> bool {
233        matches!(
234            self,
235            Self::Configuration | Self::Schema | Self::Poison | Self::Fatal
236        )
237    }
238}
239
240/// A writer schema plus its lifecycle state. A `dropped` schema is hidden
241/// from the active set and its id rejects re-registration with a different
242/// definition, but records stamped with the id keep decoding (ids are
243/// permanent). Wire mirrors stay constructible (wire-stability-bound, not
244/// API-stability-bound).
245#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
246pub struct SchemaInfo {
247    pub schema: SchemaDef,
248    #[serde(default)]
249    pub dropped: bool,
250}
251
252/// Request to read one projection's details by id.
253#[derive(Clone, Debug, Serialize, Deserialize)]
254pub struct GetProjection {
255    pub v: u32,
256    pub id: String,
257    /// Also report each binding's materialization status.
258    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
259    pub status: bool,
260}
261
262/// Request to list registered projections, optionally filtered. Empty filters
263/// list every projection. `topics` keeps projections bound to any of the named
264/// source topics. `name_contains` keeps those whose name contains the substring.
265/// `id_prefix` keeps those whose id starts with it. `search` is the single-box
266/// convenience that matches the substring against the name OR the id. `kinds`
267/// keeps projections of the named kinds. The filters compose (AND). The reply is one
268/// page of the matches in id order.
269#[derive(Clone, Debug, Default, Serialize, Deserialize)]
270pub struct ListProjections {
271    pub v: u32,
272    #[serde(default, skip_serializing_if = "Vec::is_empty")]
273    pub topics: Vec<String>,
274    #[serde(default, skip_serializing_if = "Option::is_none")]
275    pub name_contains: Option<String>,
276    #[serde(default, skip_serializing_if = "Option::is_none")]
277    pub id_prefix: Option<String>,
278    #[serde(default, skip_serializing_if = "Option::is_none")]
279    pub search: Option<String>,
280    #[serde(default, skip_serializing_if = "Vec::is_empty")]
281    pub kinds: Vec<ProjectionKind>,
282    /// The stream whose resources the listing covers. Required in stream
283    /// tenancy, where the server authorizes the stream and lists only its
284    /// resources.
285    #[serde(default, skip_serializing_if = "Option::is_none")]
286    pub stream: Option<String>,
287    /// The 1-based page. Zero reads the first page.
288    #[serde(default, skip_serializing_if = "is_zero")]
289    pub page: u32,
290    /// The projections per page. Zero leaves it to the server.
291    #[serde(default, skip_serializing_if = "is_zero")]
292    pub results: u32,
293}
294
295/// One page of a projection listing. `page` counts from 1.
296#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
297pub struct ProjectionPage {
298    pub items: Vec<ProjectionInfo>,
299    pub page: u32,
300    pub total_pages: u32,
301    pub total_results: u32,
302}
303
304/// Request to read one registered writer schema by id.
305#[derive(Clone, Debug, Serialize, Deserialize)]
306pub struct GetSchema {
307    pub v: u32,
308    pub id: u32,
309    /// The stream whose registry holds `id`. Absent for the deployment-wide
310    /// registry.
311    #[serde(default, skip_serializing_if = "Option::is_none")]
312    pub stream: Option<String>,
313}
314
315/// Request to list registered writer schemas, optionally filtered.
316/// `name_contains` keeps those whose optional name contains the substring, and
317/// an absent filter lists every schema. The filter lives here, not only on the
318/// HTTP query, so the server pushes it down rather than the client paging the
319/// whole set.
320#[derive(Clone, Debug, Serialize, Deserialize)]
321pub struct ListSchemas {
322    pub v: u32,
323    #[serde(default, skip_serializing_if = "Option::is_none")]
324    pub name_contains: Option<String>,
325    /// The stream whose registry is listed. Absent for the deployment-wide
326    /// registry.
327    #[serde(default, skip_serializing_if = "Option::is_none")]
328    pub stream: Option<String>,
329    /// The 1-based page. Zero reads the first page.
330    #[serde(default, skip_serializing_if = "is_zero")]
331    pub page: u32,
332    /// The schemas per page. Zero leaves it to the server.
333    #[serde(default, skip_serializing_if = "is_zero")]
334    pub results: u32,
335}
336
337/// One page of a writer-schema listing, ordered by id. `page` counts from 1.
338#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
339pub struct SchemaPage {
340    pub items: Vec<SchemaInfo>,
341    pub page: u32,
342    pub total_pages: u32,
343    pub total_results: u32,
344}
345
346/// The synchronous register request: no id, LaserData Cloud validates the
347/// definition, allocates the next free id, durably appends the control
348/// event, and replies `SchemaRegistered(id)`. Distinct from
349/// `ControlCommand::RegisterSchema`, the durable log form that carries the
350/// allocated id.
351#[derive(Clone, Debug, Serialize, Deserialize)]
352pub struct RegisterSchema {
353    pub v: u32,
354    pub source: SchemaSource,
355    #[serde(default, skip_serializing_if = "Option::is_none")]
356    pub name: Option<String>,
357    #[serde(default, skip_serializing_if = "Option::is_none")]
358    pub version: Option<u32>,
359    /// The stream whose registry allocates the id. Absent for the
360    /// deployment-wide registry.
361    #[serde(default, skip_serializing_if = "Option::is_none")]
362    pub stream: Option<String>,
363}
364
365/// Request to decode one record body under the schema registered for `id`,
366/// the same decode the projector runs on `agdx.sid`-stamped records. Read-only
367/// convenience for consoles and tools that hold a schema-first (Avro/Protobuf)
368/// payload and want its JSON form without re-implementing the codecs.
369#[derive(Clone, Debug, Serialize, Deserialize)]
370pub struct DecodeRecord {
371    pub v: u32,
372    pub id: u32,
373    #[serde(with = "crate::encoding::bin_bytes")]
374    pub payload: Vec<u8>,
375    /// The stream the record was read from: `id` resolves in its registry,
376    /// then in the deployment-wide one. Absent for the deployment-wide
377    /// registry alone.
378    #[serde(default, skip_serializing_if = "Option::is_none")]
379    pub stream: Option<String>,
380}
381
382/// Reply to a registry browse: `Ok` with the result, or `Err`.
383#[derive(Clone, Debug, Serialize, Deserialize)]
384#[non_exhaustive]
385pub enum BrowseReply {
386    Ok(BrowseOutcome),
387    Err(QueryError),
388}
389
390/// The result of a registry browse, shaped per request.
391#[derive(Clone, Debug, Serialize, Deserialize)]
392#[non_exhaustive]
393pub enum BrowseOutcome {
394    /// `list_projections`: one page of the matching projections.
395    Projections(ProjectionPage),
396    /// `get projection`: the projection with the requested id, or `None`.
397    Projection(Option<Box<ProjectionInfo>>),
398    /// `list schemas`: one page of writer schemas, active and tombstoned.
399    Schemas(SchemaPage),
400    /// `get schema`: the schema occupying the requested id, or `None`.
401    Schema(Option<SchemaInfo>),
402    /// `register schema`: the LaserData-Cloud-allocated id, already durably appended
403    /// to the control topic (visibility follows within the apply latency).
404    SchemaRegistered(u32),
405    /// `decode record`: the payload's JSON form under the requested schema,
406    /// or `None` when the body does not decode under it. A `serde_json::Value`
407    /// because the decoded shape is arbitrary (object, array, scalar) and the
408    /// LaserData Cloud encodes exactly that value into the reply.
409    Decoded(Option<serde_json::Value>),
410}
411
412fn is_zero(value: &u32) -> bool {
413    *value == 0
414}
415
416#[cfg(all(test, feature = "cbor"))]
417mod tests {
418    use super::*;
419    use crate::codes::QUERY_OP_VERSION;
420    use crate::content::ContentType;
421    use crate::control::ProjectionBinding;
422    use crate::framing::{decode_named, encode_named};
423
424    fn status(
425        realization: RealizationState,
426        partitions: &[(u32, Option<u64>, Option<u64>)],
427    ) -> BindingStatus {
428        BindingStatus {
429            history: ProjectionHistory::Append,
430            source: SourceSelector::new("fleet", "readings"),
431            index: "readings_rows".to_owned(),
432            realization,
433            realization_reason: None,
434            failure: None,
435            stall: None,
436            source_incarnation: Some(7),
437            partitions: partitions
438                .iter()
439                .map(
440                    |(partition_id, head_offset, applied_offset)| PartitionProgress {
441                        partition_id: *partition_id,
442                        head_offset: *head_offset,
443                        applied_offset: *applied_offset,
444                        first_offset: applied_offset.map(|_| 0),
445                    },
446                )
447                .collect(),
448            holds: Vec::new(),
449            accepted_gaps: Vec::new(),
450            skipped_records: 0,
451        }
452    }
453
454    #[test]
455    fn given_a_missing_incarnation_or_a_hold_when_checking_catch_up_then_should_refuse() {
456        let target = status(RealizationState::Realized, &[(0, Some(10), Some(10))]);
457        let mut unknown = target.clone();
458        unknown.source_incarnation = None;
459        assert!(!unknown.caught_up_to(&unknown));
460        let mut blocked = target.clone();
461        blocked.holds.push(Hold {
462            partition_id: 0,
463            offset: 11,
464            cause: HoldCause::Schema,
465            reason: "invalid".to_owned(),
466        });
467        assert!(!blocked.caught_up_to(&target));
468    }
469
470    #[test]
471    fn given_a_captured_target_when_the_destination_reaches_it_then_should_be_caught_up() {
472        let target = status(
473            RealizationState::Realized,
474            &[(0, Some(10), Some(4)), (1, None, None)],
475        );
476        let behind = status(RealizationState::Realized, &[(0, Some(12), Some(9))]);
477        assert!(!behind.caught_up_to(&target));
478        let reached = status(RealizationState::Realized, &[(0, Some(15), Some(10))]);
479        assert!(
480            reached.caught_up_to(&target),
481            "a newer head does not move the target"
482        );
483        let unrealized = status(RealizationState::AwaitingOwner, &[(0, Some(15), Some(10))]);
484        assert!(!unrealized.caught_up_to(&target));
485        let mut recreated = reached.clone();
486        recreated.source_incarnation = Some(8);
487        assert!(
488            !recreated.caught_up_to(&target),
489            "offsets of another incarnation never compare"
490        );
491        assert_eq!(reached.readiness(), MaterializationReadiness::Realized);
492        assert_eq!(unrealized.readiness(), MaterializationReadiness::Applied);
493    }
494
495    #[test]
496    fn given_a_late_start_a_gap_a_skip_or_references_when_checked_then_should_not_claim_complete_history()
497     {
498        let complete = status(RealizationState::Realized, &[(0, Some(3), Some(3))]);
499        assert!(complete.history_complete(OversizePolicy::Block));
500        assert!(!complete.history_complete(OversizePolicy::Reference));
501        let mut late = complete.clone();
502        late.partitions[0].first_offset = Some(5);
503        assert!(!late.history_complete(OversizePolicy::Block));
504        let mut unknown = complete.clone();
505        unknown.partitions[0].first_offset = None;
506        assert!(
507            !unknown.history_complete(OversizePolicy::Block),
508            "unknown is never complete"
509        );
510        let mut skipped = complete.clone();
511        skipped.skipped_records = 1;
512        assert!(!skipped.history_complete(OversizePolicy::Block));
513        let mut held = complete.clone();
514        held.holds.push(Hold {
515            partition_id: 0,
516            offset: 4,
517            cause: HoldCause::Oversize,
518            reason: "too large".to_owned(),
519        });
520        assert!(held.is_blocked());
521    }
522
523    #[test]
524    fn given_a_partition_whose_records_expired_unread_when_checked_then_should_not_claim_complete_history()
525     {
526        let mut expired = status(RealizationState::Realized, &[(0, Some(3), Some(3))]);
527        expired.partitions.push(PartitionProgress {
528            partition_id: 1,
529            head_offset: None,
530            applied_offset: None,
531            first_offset: Some(5),
532        });
533        assert!(
534            !expired.history_complete(OversizePolicy::Block),
535            "records 0 to 4 of partition 1 expired before the binding read them"
536        );
537        let mut never_written = status(RealizationState::Realized, &[(0, Some(3), Some(3))]);
538        never_written.partitions.push(PartitionProgress {
539            partition_id: 1,
540            head_offset: None,
541            applied_offset: None,
542            first_offset: None,
543        });
544        assert!(never_written.history_complete(OversizePolicy::Block));
545    }
546
547    #[test]
548    fn given_realization_failures_when_checked_then_should_fail_only_on_a_permanent_class() {
549        let mut failed = status(RealizationState::Pending, &[]);
550        assert!(!failed.is_failed());
551        for (class, permanent) in [
552            (FailureClass::Configuration, true),
553            (FailureClass::Schema, true),
554            (FailureClass::Poison, true),
555            (FailureClass::Fatal, true),
556            (FailureClass::Transient, false),
557            (FailureClass::Capacity, false),
558            (FailureClass::ReadOnly, false),
559            (FailureClass::Unknown, false),
560        ] {
561            failed.failure = Some(class);
562            assert_eq!(failed.is_failed(), permanent, "{class:?}");
563        }
564    }
565
566    #[test]
567    fn given_a_failure_a_stall_and_a_retired_hold_when_round_tripped_then_should_preserve_them() {
568        let mut stalled = status(RealizationState::Pending, &[(0, Some(3), None)]);
569        stalled.failure = Some(FailureClass::ReadOnly);
570        stalled.stall = Some(StallReason::Backpressure);
571        stalled.holds.push(Hold {
572            partition_id: 0,
573            offset: 0,
574            cause: HoldCause::RetiredIncarnation,
575            reason: "applied for an earlier stream".to_owned(),
576        });
577        let back: BindingStatus =
578            decode_named(&encode_named(&stalled).expect("serializes")).expect("deserializes");
579        assert_eq!(back, stalled);
580        let json = serde_json::to_value(&stalled).expect("serializes");
581        assert_eq!(
582            (&json["failure"], &json["stall"], &json["holds"][0]["cause"]),
583            (
584                &serde_json::json!("read_only"),
585                &serde_json::json!("backpressure"),
586                &serde_json::json!("retired_incarnation")
587            )
588        );
589        let newer: BindingStatus = serde_json::from_value(serde_json::json!({
590            "source": {"stream": "fleet", "topic": "readings"},
591            "index": "readings_rows",
592            "realization": "pending",
593            "failure": "quota",
594            "stall": "paused",
595        }))
596        .expect("a newer class decodes");
597        assert_eq!(
598            (newer.failure, newer.stall),
599            (Some(FailureClass::Unknown), Some(StallReason::Unknown))
600        );
601        let quiet =
602            serde_json::to_value(status(RealizationState::Realized, &[])).expect("serializes");
603        assert!(
604            quiet.get("failure").is_none() && quiet.get("stall").is_none(),
605            "absent fields are omitted: {quiet}"
606        );
607    }
608
609    #[test]
610    fn given_a_projection_listing_when_encoded_then_should_send_its_page_only_when_set() {
611        let first = ListProjections {
612            v: QUERY_OP_VERSION,
613            ..ListProjections::default()
614        };
615        let json = serde_json::to_value(&first).expect("serializes");
616        assert!(
617            json.get("page").is_none()
618                && json.get("results").is_none()
619                && json.get("kinds").is_none(),
620            "{json}"
621        );
622        let third = ListProjections {
623            v: QUERY_OP_VERSION,
624            kinds: vec![ProjectionKind::Row, ProjectionKind::Record],
625            page: 3,
626            results: 20,
627            ..ListProjections::default()
628        };
629        let back: ListProjections =
630            decode_named(&encode_named(&third).expect("serializes")).expect("deserializes");
631        assert_eq!(
632            (back.kinds, back.page, back.results),
633            (vec![ProjectionKind::Row, ProjectionKind::Record], 3, 20)
634        );
635    }
636
637    #[test]
638    fn given_a_browse_reply_when_round_tripped_then_should_preserve_projection_details() {
639        let info = ProjectionInfo {
640            projection: Projection::builder("reading.v1")
641                .name("reading")
642                .version(1)
643                .content_type(ContentType::Json)
644                .fields(["reading_id", "cpu"])
645                .build(),
646            bindings: vec![
647                ProjectionBinding::builder()
648                    .source("fleet", "readings")
649                    .allow("reading.v1")
650                    .default_projection("reading.v1")
651                    .index("readings_rows")
652                    .build(),
653            ],
654            status: Vec::new(),
655        };
656        let reply = BrowseReply::Ok(BrowseOutcome::Projections(ProjectionPage {
657            items: vec![info],
658            page: 1,
659            total_pages: 1,
660            total_results: 1,
661        }));
662        let bytes = encode_named(&reply).expect("the reply serializes");
663        let back: BrowseReply = decode_named(&bytes).expect("the reply deserializes");
664        let BrowseReply::Ok(BrowseOutcome::Projections(page)) = back else {
665            panic!("expected an Ok(Projections) browse reply");
666        };
667        assert_eq!((page.page, page.total_pages, page.total_results), (1, 1, 1));
668        let list = page.items;
669        assert_eq!(list.len(), 1);
670        assert_eq!(list[0].projection.id.as_str(), "reading.v1");
671        assert_eq!(list[0].projection.extraction.fields.len(), 2);
672        assert_eq!(list[0].bindings[0].index, "readings_rows");
673    }
674
675    #[test]
676    fn given_a_decode_record_when_round_tripped_then_should_preserve_payload_bytes() {
677        let request = DecodeRecord {
678            v: QUERY_OP_VERSION,
679            id: 7,
680            payload: vec![0xff, 0x00, 0x10],
681            stream: None,
682        };
683        let bytes = encode_named(&request).expect("serializes");
684        let back: DecodeRecord = decode_named(&bytes).expect("deserializes");
685        assert_eq!(back.id, 7);
686        assert_eq!(back.payload, vec![0xff, 0x00, 0x10]);
687        assert_eq!(back.stream, None);
688    }
689
690    #[test]
691    fn given_schema_requests_with_a_stream_when_round_tripped_then_should_preserve_it() {
692        let get = GetSchema {
693            v: QUERY_OP_VERSION,
694            id: 7,
695            stream: Some("acme".to_owned()),
696        };
697        let back: GetSchema =
698            decode_named(&encode_named(&get).expect("serializes")).expect("deserializes");
699        assert_eq!((back.id, back.stream.as_deref()), (7, Some("acme")));
700        let list = ListSchemas {
701            v: QUERY_OP_VERSION,
702            name_contains: None,
703            stream: Some("acme".to_owned()),
704            page: 0,
705            results: 0,
706        };
707        let back: ListSchemas =
708            decode_named(&encode_named(&list).expect("serializes")).expect("deserializes");
709        assert_eq!(back.stream.as_deref(), Some("acme"));
710        let register = RegisterSchema {
711            v: QUERY_OP_VERSION,
712            source: SchemaSource::JsonSchema {
713                schema: "{}".to_owned(),
714            },
715            name: None,
716            version: None,
717            stream: Some("acme".to_owned()),
718        };
719        let back: RegisterSchema =
720            decode_named(&encode_named(&register).expect("serializes")).expect("deserializes");
721        assert_eq!(back.stream.as_deref(), Some("acme"));
722        let decode = DecodeRecord {
723            v: QUERY_OP_VERSION,
724            id: 7,
725            payload: vec![1],
726            stream: Some("acme".to_owned()),
727        };
728        let back: DecodeRecord =
729            decode_named(&encode_named(&decode).expect("serializes")).expect("deserializes");
730        assert_eq!(back.stream.as_deref(), Some("acme"));
731    }
732
733    #[test]
734    fn given_schema_requests_without_a_stream_when_encoded_then_should_omit_it() {
735        let get = GetSchema {
736            v: QUERY_OP_VERSION,
737            id: 7,
738            stream: None,
739        };
740        let list = ListSchemas {
741            v: QUERY_OP_VERSION,
742            name_contains: None,
743            stream: None,
744            page: 0,
745            results: 0,
746        };
747        for json in [
748            serde_json::to_string(&get).expect("serializes"),
749            serde_json::to_string(&list).expect("serializes"),
750        ] {
751            assert!(
752                !json.contains("stream"),
753                "an unset stream is omitted: {json}"
754            );
755        }
756    }
757}
laser_wire/browse.rs

laser_wire/
browse.rs
