laser_wire/filter/
read.rs

1use crate::codes::FILTER_OP_VERSION;
2use crate::error::InvalidError;
3use crate::filter::expr::ConsumerFilter;
4use crate::limits::{
5    MAX_FILTER_PREVIEW_EXAMINED, MAX_FILTER_PREVIEW_RECORDS, MAX_FILTER_SAMPLE_BYTES,
6    MAX_FILTER_SAMPLE_HEADERS, MAX_FILTER_SOURCE_NAME_BYTES, MAX_FILTERED_PAGE_BYTES,
7    MAX_FILTERED_PAGE_RECORDS,
8};
9use crate::result::ResultCode;
10use crate::schema::Digest32;
11use crate::validate::Validate;
12use serde::ser::SerializeStruct;
13use serde::{Deserialize, Deserializer, Serialize, Serializer};
14use std::fmt;
15
16/// The stream and topic a filtered read or preview addresses, by name.
17#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
18pub struct FilterSource {
19    pub stream: String,
20    pub topic: String,
21}
22
23/// Whose progress a filtered read follows: an independent consumer or a
24/// consumer group. Groups accept either a name or a numeric ID.
25#[derive(Clone, Debug, PartialEq, Eq)]
26pub enum FilterConsumer {
27    Consumer(String),
28    Group(String),
29    GroupId(u64),
30}
31
32impl fmt::Display for FilterConsumer {
33    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
34        match self {
35            Self::Consumer(name) | Self::Group(name) => formatter.write_str(name),
36            Self::GroupId(id) => write!(formatter, "{id}"),
37        }
38    }
39}
40
41impl FilterConsumer {
42    pub const fn is_group(&self) -> bool {
43        matches!(self, Self::Group(_) | Self::GroupId(_))
44    }
45}
46
47impl Serialize for FilterConsumer {
48    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
49        let mut state = serializer.serialize_struct("FilterConsumer", 2)?;
50        match self {
51            Self::Consumer(name) => {
52                state.serialize_field("kind", "consumer")?;
53                state.serialize_field("name", name)?;
54            }
55            Self::Group(name) => {
56                state.serialize_field("kind", "group")?;
57                state.serialize_field("name", name)?;
58            }
59            Self::GroupId(id) => {
60                state.serialize_field("kind", "group_id")?;
61                state.serialize_field("id", id)?;
62            }
63        }
64        state.end()
65    }
66}
67
68impl<'de> Deserialize<'de> for FilterConsumer {
69    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
70        #[derive(Deserialize)]
71        #[serde(tag = "kind", rename_all = "snake_case")]
72        enum Selector {
73            Consumer { name: String },
74            Group { name: String },
75            GroupId { id: u64 },
76        }
77        Ok(match Selector::deserialize(deserializer)? {
78            Selector::Consumer { name } => Self::Consumer(name),
79            Selector::Group { name } => Self::Group(name),
80            Selector::GroupId { id } => Self::GroupId(id),
81        })
82    }
83}
84
85/// Which filter a read executes.
86#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
87#[serde(rename_all = "snake_case")]
88pub enum FilterRef {
89    /// The full filter in the request. Needs no catalog. A consumer group
90    /// still needs a binding with the same digest.
91    Inline(ConsumerFilter),
92    /// A saved revision from the catalog.
93    Revision { filter_id: u32, revision: u32 },
94    /// Whatever revision the consumer group is bound to. Groups only.
95    Bound,
96}
97
98/// Where a filtered read starts.
99#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
100#[serde(rename_all = "snake_case")]
101pub enum FilteredStart {
102    /// After the consumer's stored offset.
103    Next,
104    First,
105    Last,
106    Offset(u64),
107    /// Records at or after this broker timestamp, in microseconds.
108    Timestamp(u64),
109    /// Continue exactly where the previous page stopped.
110    Continue(Continuation),
111}
112
113/// Which replica serves the read.
114#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize, strum::Display)]
115#[serde(rename_all = "snake_case")]
116#[strum(serialize_all = "snake_case")]
117pub enum ReadMode {
118    /// The partition primary, through an attached consumer session. The owner
119    /// proves the page came from the generation the reply names. Required for
120    /// acknowledgments.
121    #[default]
122    Primary,
123    /// Whatever replica the connection reaches. It may lag without bound, so it
124    /// serves diagnostics and previews, never acknowledgments.
125    Local,
126}
127
128impl ReadMode {
129    pub const fn is_primary(&self) -> bool {
130        matches!(self, Self::Primary)
131    }
132}
133
134/// The exact source history a page was read from. A purge or a recreated
135/// stream, topic, or partition changes it, so a continuation or acknowledgment
136/// from an older history is rejected rather than applied to new records.
137#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
138pub struct SourceGeneration {
139    pub stream_id: u32,
140    pub stream_created_at_micros: u64,
141    pub topic_id: u32,
142    pub topic_created_at_micros: u64,
143    pub partition_id: u32,
144    pub partition_created_revision: u64,
145    pub purge_generation: u64,
146}
147
148/// The resume point a page hands to the next request.
149#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
150pub struct Continuation {
151    #[serde(default, skip_serializing_if = "Option::is_none")]
152    pub group_id: Option<u64>,
153    pub next_scan_offset: u64,
154    pub generation: SourceGeneration,
155    pub digest: Digest32,
156    /// The route the page was read on. A continuation from a local page never
157    /// resumes a primary read, because a local replica can lag a purge.
158    #[serde(default, skip_serializing_if = "ReadMode::is_primary")]
159    pub read_mode: ReadMode,
160}
161
162/// Read one bounded page of matching records from one partition.
163#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
164pub struct FilteredPollRequest {
165    pub v: u32,
166    pub source: FilterSource,
167    pub partition_id: u32,
168    pub consumer: FilterConsumer,
169    pub filter: FilterRef,
170    pub start: FilteredStart,
171    /// Most matching records to return.
172    pub count: u32,
173    /// Most record bytes to return. The server may lower it.
174    pub max_reply_bytes: u32,
175    #[serde(default)]
176    pub read_mode: ReadMode,
177}
178
179/// Why a page stopped scanning.
180#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, strum::Display)]
181#[serde(rename_all = "snake_case")]
182#[strum(serialize_all = "snake_case")]
183pub enum StopReason {
184    /// The requested number of matches was reached.
185    Filled,
186    /// A record, byte, or round budget was spent. More records may follow.
187    Budget,
188    /// No more records are visible on the serving replica right now.
189    EndOfVisible,
190    /// A record failed to decode under the `stop` fault policy.
191    Fault,
192    /// One matching record alone exceeds the reply cap.
193    OversizedRecord,
194}
195
196/// Why a record could not be evaluated.
197#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, strum::Display)]
198#[serde(rename_all = "snake_case")]
199#[strum(serialize_all = "snake_case")]
200pub enum FaultReason {
201    /// No valid uint32 agdx.sid header identifies the writer schema.
202    MissingSchema,
203    /// The record schema is not permitted by this filter.
204    SchemaNotAllowed,
205    /// The registered schema family differs from the payload codec.
206    SchemaMismatch,
207    /// The payload is not valid for the filter's codec.
208    Malformed,
209    /// The payload exceeds the decoder's size limit.
210    TooLarge,
211    /// The payload nests deeper than the decoder's depth limit.
212    TooDeep,
213    /// The record's agdx.ct header names another codec than the filter's.
214    ForeignCodec,
215    /// A predicate's value exists but has a type the predicate cannot compare.
216    TypeMismatch,
217}
218
219impl FaultReason {
220    /// Whether the record is in another format than the filter reads, which
221    /// follows the filter's foreign policy instead of its fault policy.
222    pub const fn is_foreign(self) -> bool {
223        matches!(
224            self,
225            FaultReason::ForeignCodec
226                | FaultReason::MissingSchema
227                | FaultReason::SchemaNotAllowed
228                | FaultReason::SchemaMismatch
229        )
230    }
231}
232
233/// One record the filter could not evaluate.
234#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
235pub struct RecordFault {
236    pub offset: u64,
237    pub reason: FaultReason,
238}
239
240/// Decoder limits applied by the server to unevaluated records in a page.
241#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
242pub struct FilterDecodeLimits {
243    pub max_payload_bytes: u32,
244    pub max_depth: u32,
245}
246
247/// The filter a page actually executed.
248#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
249pub struct AppliedPolicy {
250    #[serde(default, skip_serializing_if = "Option::is_none")]
251    pub group_id: Option<u64>,
252    pub digest: Digest32,
253    #[serde(default, skip_serializing_if = "Option::is_none")]
254    pub filter_id: Option<u32>,
255    #[serde(default, skip_serializing_if = "Option::is_none")]
256    pub revision: Option<u32>,
257}
258
259/// One page of a filtered read.
260///
261/// `records` is a standard polled-messages body: the matching records keep
262/// their original offsets, ids, timestamps, headers, and payload bytes.
263/// `next_scan_offset` is the first record the server has not examined, `None`
264/// when nothing was examined and the start selector must be sent again.
265/// `safe_ack_offset` is the last examined offset. The reader may acknowledge it
266/// only after every returned record at or below it is fully handled.
267#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
268pub struct FilteredPage {
269    pub v: u32,
270    pub partition_id: u32,
271    pub policy: AppliedPolicy,
272    pub generation: SourceGeneration,
273    pub read_mode: ReadMode,
274    #[serde(default, skip_serializing_if = "Option::is_none")]
275    pub next_scan_offset: Option<u64>,
276    #[serde(default, skip_serializing_if = "Option::is_none")]
277    pub safe_ack_offset: Option<u64>,
278    /// The serving replica's partition head when the page was read. Reported,
279    /// never used as scan progress.
280    pub frontier: u64,
281    pub examined: u32,
282    pub matched: u32,
283    pub stop: StopReason,
284    #[serde(default, skip_serializing_if = "Option::is_none")]
285    pub fault: Option<RecordFault>,
286    /// Offsets of returned records the filter could not evaluate (the `pass`
287    /// fault policy).
288    #[serde(default, skip_serializing_if = "Vec::is_empty")]
289    pub unevaluated: Vec<u64>,
290    #[serde(default, skip_serializing_if = "Option::is_none")]
291    pub evaluation_limits: Option<Box<FilterDecodeLimits>>,
292    #[serde(with = "crate::encoding::bin_bytes")]
293    pub records: Vec<u8>,
294}
295
296impl FilteredPage {
297    /// The start of the next request: continue after this page, or repeat the
298    /// original selector when nothing was examined.
299    pub fn next_start(&self, original: &FilteredStart) -> FilteredStart {
300        match self.next_scan_offset {
301            Some(next_scan_offset) => FilteredStart::Continue(Continuation {
302                group_id: self.policy.group_id,
303                next_scan_offset,
304                generation: self.generation,
305                digest: self.policy.digest.clone(),
306                read_mode: self.read_mode,
307            }),
308            None => original.clone(),
309        }
310    }
311}
312
313/// Acknowledge a filtered page through its safe offset. The server stores the
314/// offset only while the source still has the generation the page was read
315/// from, and only while the consumer still owns the partition.
316#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
317pub struct FilteredAck {
318    #[serde(default, skip_serializing_if = "Option::is_none")]
319    pub group_id: Option<u64>,
320    pub v: u32,
321    pub source: FilterSource,
322    pub partition_id: u32,
323    pub consumer: FilterConsumer,
324    pub generation: SourceGeneration,
325    pub digest: Digest32,
326    pub offset: u64,
327}
328
329/// A stored acknowledgment.
330#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
331pub struct AckReceipt {
332    pub partition_id: u32,
333    pub offset: u64,
334    pub generation: SourceGeneration,
335}
336
337/// Preview a filter over stored records of one partition. Observational: it
338/// joins no group, stores no offset, and changes no consumer state.
339#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
340pub struct FilterPreviewRequest {
341    pub v: u32,
342    pub source: FilterSource,
343    pub partition_id: u32,
344    /// `inline` or `revision`. `bound` is rejected.
345    pub filter: FilterRef,
346    pub from_offset: u64,
347    pub max_examined: u32,
348    pub max_records: u32,
349    /// Include a per-predicate explanation for every returned record.
350    #[serde(default)]
351    pub explain: bool,
352}
353
354/// How the filter judged one previewed record.
355#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, strum::Display)]
356#[serde(rename_all = "snake_case")]
357#[strum(serialize_all = "snake_case")]
358pub enum Verdict {
359    Selected,
360    Rejected,
361    Fault,
362}
363
364/// One previewed record: identity, verdict, and a bounded text view.
365#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
366pub struct PreviewRecord {
367    pub offset: u64,
368    pub timestamp_micros: u64,
369    pub verdict: Verdict,
370    #[serde(default, skip_serializing_if = "Option::is_none")]
371    pub fault: Option<FaultReason>,
372    /// The payload as lossy UTF-8, cut at the preview cap.
373    pub payload_text: String,
374    #[serde(default)]
375    pub payload_truncated: bool,
376    #[serde(default, skip_serializing_if = "Option::is_none")]
377    pub explanation: Option<FilterExplanation>,
378}
379
380/// A preview of one partition range.
381#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
382pub struct FilterPreview {
383    pub v: u32,
384    pub partition_id: u32,
385    pub policy: AppliedPolicy,
386    pub read_mode: ReadMode,
387    pub examined: u32,
388    pub matched: u32,
389    pub faults: u32,
390    pub stop: StopReason,
391    #[serde(default, skip_serializing_if = "Option::is_none")]
392    pub next_offset: Option<u64>,
393    pub frontier: u64,
394    /// Selected records first, capped at `max_records`. Rejected records are
395    /// listed only when `explain` was requested.
396    pub records: Vec<PreviewRecord>,
397}
398
399/// A typed user header value, as the evaluator reads it. Iggy integer kinds up
400/// to 64 bits map to `int` or `uint`. 128-bit integers arrive as `raw` bytes.
401#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
402#[serde(tag = "kind", content = "value", rename_all = "snake_case")]
403pub enum HeaderScalar {
404    Bool(bool),
405    Int(i64),
406    Uint(u64),
407    Float(f64),
408    String(String),
409    Raw(#[serde(with = "crate::encoding::bin_bytes")] Vec<u8>),
410}
411
412/// One user header of a record.
413#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
414pub struct FilterHeader {
415    pub key: String,
416    pub value: HeaderScalar,
417}
418
419/// Evaluate a filter against one supplied record. Nothing is read from a
420/// stream and nothing is stored.
421#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
422pub struct FilterTestRequest {
423    pub v: u32,
424    /// `inline` or `revision`. `bound` is rejected.
425    pub filter: FilterRef,
426    #[serde(with = "crate::encoding::bin_bytes")]
427    pub payload: Vec<u8>,
428    #[serde(default, skip_serializing_if = "Vec::is_empty")]
429    pub headers: Vec<FilterHeader>,
430}
431
432/// Three-valued comparison truth.
433#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, strum::Display)]
434#[serde(rename_all = "snake_case")]
435#[strum(serialize_all = "snake_case")]
436pub enum Truth {
437    Match,
438    NoMatch,
439    Unknown,
440}
441
442/// One node of an explanation tree: what it tested and how it resolved.
443/// `truth` is absent when the payload could not be decoded for this node.
444#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
445pub struct ExplainNode {
446    pub label: String,
447    #[serde(default, skip_serializing_if = "Option::is_none")]
448    pub truth: Option<Truth>,
449    #[serde(default, skip_serializing_if = "Vec::is_empty")]
450    pub children: Vec<ExplainNode>,
451}
452
453/// Why a record was or was not selected.
454#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
455pub struct FilterExplanation {
456    pub verdict: Verdict,
457    #[serde(default, skip_serializing_if = "Option::is_none")]
458    pub fault: Option<FaultReason>,
459    pub root: ExplainNode,
460}
461
462/// The result of a sample test.
463#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
464pub struct FilterTestResult {
465    pub v: u32,
466    pub policy: AppliedPolicy,
467    pub explanation: FilterExplanation,
468}
469
470/// A filter that validated and compiled.
471#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
472pub struct FilterValidation {
473    pub v: u32,
474    pub digest: Digest32,
475    /// Whether evaluating the filter can decode the payload.
476    pub reads_payload: bool,
477    /// Whether evaluating the filter can read user headers.
478    pub reads_headers: bool,
479}
480
481/// The typed cause of a consumer-filter failure. Each maps to one shared
482/// [`ResultCode`], so a generic client can still classify it.
483#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, strum::Display)]
484#[serde(rename_all = "snake_case")]
485#[strum(serialize_all = "snake_case")]
486#[non_exhaustive]
487pub enum FilterErrorReason {
488    /// The filter or request is malformed or out of range.
489    InvalidRequest,
490    /// The server or its backend does not serve this operation.
491    Unsupported,
492    /// The wire version is not accepted.
493    VersionSkew,
494    /// The source, filter, revision, binding, or operation does not exist.
495    NotFound,
496    /// A policy or mutation precondition does not hold (a different digest is
497    /// bound, a name is taken, an expected revision moved).
498    Conflict,
499    /// The source history changed (purge or recreation) since the page or
500    /// continuation was issued.
501    SourceChanged,
502    /// The consumer no longer owns the partition, or its session ended.
503    MembershipStale,
504    /// This node is not the partition primary. Resolve the route again.
505    NotPrimary,
506    /// The catalog cannot confirm the policy right now.
507    CatalogUnavailable,
508    /// The saved revision is paused. Pending records may still be acknowledged.
509    RevisionDisabled,
510    /// The request is too large.
511    TooLarge,
512    Unauthenticated,
513    Forbidden,
514    /// A transient failure. The same request may succeed later.
515    Unavailable,
516    /// A configured catalog resource limit is reached.
517    CapacityExhausted,
518    /// An unexpected failure the caller cannot retry away.
519    Backend,
520    /// A reason this build does not know. The error's `code` still classifies
521    /// it.
522    #[serde(other)]
523    Unknown,
524}
525
526impl FilterErrorReason {
527    pub const fn code(self) -> ResultCode {
528        match self {
529            Self::InvalidRequest => ResultCode::InvalidArgument,
530            Self::Unsupported => ResultCode::Unsupported,
531            Self::VersionSkew => ResultCode::VersionSkew,
532            Self::NotFound => ResultCode::NotFound,
533            Self::Conflict => ResultCode::Conflict,
534            Self::SourceChanged => ResultCode::StaleGeneration,
535            Self::MembershipStale | Self::NotPrimary => ResultCode::Stale,
536            Self::CatalogUnavailable | Self::Unavailable | Self::RevisionDisabled => {
537                ResultCode::Unavailable
538            }
539            Self::TooLarge => ResultCode::TooLarge,
540            Self::Unauthenticated => ResultCode::Unauthenticated,
541            Self::Forbidden => ResultCode::Forbidden,
542            Self::CapacityExhausted => ResultCode::ResourceLimit,
543            Self::Backend | Self::Unknown => ResultCode::Backend,
544        }
545    }
546}
547
548/// A consumer-filter failure.
549#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
550#[error("{reason}: {message}")]
551pub struct FilterError {
552    pub code: ResultCode,
553    pub reason: FilterErrorReason,
554    pub message: String,
555}
556
557impl FilterError {
558    pub fn new(reason: FilterErrorReason, message: impl Into<String>) -> Self {
559        Self {
560            code: reason.code(),
561            reason,
562            message: message.into(),
563        }
564    }
565
566    pub fn invalid(error: InvalidError) -> Self {
567        Self::new(FilterErrorReason::InvalidRequest, error.0)
568    }
569
570    /// Refuse a request of another filter op version as `version_skew`, before
571    /// its shape is validated, so a client of another build can tell a version
572    /// gap from a malformed request.
573    ///
574    /// # Errors
575    ///
576    /// `version_skew` when `v` is not [`FILTER_OP_VERSION`].
577    pub fn check_version(v: u32) -> Result<(), Self> {
578        validate_version(v).map_err(|error| Self::new(FilterErrorReason::VersionSkew, error.0))
579    }
580}
581
582/// A successful reply of a server-native consumer-filter command.
583#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
584#[serde(rename_all = "snake_case")]
585pub enum FilterOutcome {
586    Page(FilteredPage),
587    Acknowledged(AckReceipt),
588    Preview(FilterPreview),
589    Tested(FilterTestResult),
590    Validated(FilterValidation),
591}
592
593/// The reply to every server-native consumer-filter command.
594#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
595#[serde(rename_all = "snake_case")]
596// Keep successful pages inline without another allocation on every read.
597#[allow(clippy::large_enum_variant)]
598pub enum FilterReply {
599    Ok(FilterOutcome),
600    Err(FilterError),
601}
602
603fn validate_version(v: u32) -> Result<(), InvalidError> {
604    if v == FILTER_OP_VERSION {
605        Ok(())
606    } else {
607        Err(InvalidError::new(format!(
608            "filter request version {v} is not supported, expected {FILTER_OP_VERSION}"
609        )))
610    }
611}
612
613pub(crate) fn validate_name(label: &str, name: &str) -> Result<(), InvalidError> {
614    if name.is_empty() || name.len() > MAX_FILTER_SOURCE_NAME_BYTES {
615        return Err(InvalidError::new(format!(
616            "{label} must be 1..={MAX_FILTER_SOURCE_NAME_BYTES} bytes"
617        )));
618    }
619    Ok(())
620}
621
622impl Validate for FilterSource {
623    fn validate(&self) -> Result<(), InvalidError> {
624        validate_name("stream", &self.stream)?;
625        validate_name("topic", &self.topic)
626    }
627}
628
629impl Validate for FilterConsumer {
630    fn validate(&self) -> Result<(), InvalidError> {
631        match self {
632            Self::Consumer(name) => validate_name("consumer", name),
633            Self::Group(name) => validate_name("consumer group", name),
634            Self::GroupId(id) => u32::try_from(*id).map(|_| ()).map_err(|_| {
635                InvalidError::new("consumer group id exceeds the native offset-key range")
636            }),
637        }
638    }
639}
640
641impl Validate for FilterRef {
642    fn validate(&self) -> Result<(), InvalidError> {
643        match self {
644            Self::Inline(filter) => filter.validate(),
645            Self::Revision { .. } | Self::Bound => Ok(()),
646        }
647    }
648}
649
650impl Validate for FilteredPollRequest {
651    fn validate(&self) -> Result<(), InvalidError> {
652        validate_version(self.v)?;
653        self.source.validate()?;
654        self.consumer.validate()?;
655        self.filter.validate()?;
656        if matches!(self.filter, FilterRef::Bound) && !self.consumer.is_group() {
657            return Err(InvalidError::new("a `bound` filter needs a consumer group"));
658        }
659        if self.count == 0 || self.count > MAX_FILTERED_PAGE_RECORDS {
660            return Err(InvalidError::new(format!(
661                "count must be 1..={MAX_FILTERED_PAGE_RECORDS}"
662            )));
663        }
664        if self.max_reply_bytes == 0 || self.max_reply_bytes > MAX_FILTERED_PAGE_BYTES {
665            return Err(InvalidError::new(format!(
666                "max_reply_bytes must be 1..={MAX_FILTERED_PAGE_BYTES}"
667            )));
668        }
669        if let FilteredStart::Continue(continuation) = &self.start {
670            continuation.digest.validate()?;
671            if self.consumer.is_group() != continuation.group_id.is_some() {
672                return Err(InvalidError::new(
673                    "a group continuation must retain its group identity",
674                ));
675            }
676            if continuation.generation.partition_id != self.partition_id {
677                return Err(InvalidError::new(
678                    "a continuation belongs to another partition",
679                ));
680            }
681        }
682        Ok(())
683    }
684}
685
686impl Validate for FilteredAck {
687    fn validate(&self) -> Result<(), InvalidError> {
688        validate_version(self.v)?;
689        self.source.validate()?;
690        self.consumer.validate()?;
691        self.digest.validate()?;
692        if self.consumer.is_group() != self.group_id.is_some() {
693            return Err(InvalidError::new(
694                "a group acknowledgment must retain its group identity",
695            ));
696        }
697        if self.generation.partition_id != self.partition_id {
698            return Err(InvalidError::new(
699                "an acknowledgment names another partition's generation",
700            ));
701        }
702        Ok(())
703    }
704}
705
706impl Validate for FilterPreviewRequest {
707    fn validate(&self) -> Result<(), InvalidError> {
708        validate_version(self.v)?;
709        self.source.validate()?;
710        self.filter.validate()?;
711        if matches!(self.filter, FilterRef::Bound) {
712            return Err(InvalidError::new(
713                "a preview executes an inline filter or a saved revision",
714            ));
715        }
716        if self.max_examined == 0 || self.max_examined > MAX_FILTER_PREVIEW_EXAMINED {
717            return Err(InvalidError::new(format!(
718                "max_examined must be 1..={MAX_FILTER_PREVIEW_EXAMINED}"
719            )));
720        }
721        if self.max_records == 0 || self.max_records > MAX_FILTER_PREVIEW_RECORDS {
722            return Err(InvalidError::new(format!(
723                "max_records must be 1..={MAX_FILTER_PREVIEW_RECORDS}"
724            )));
725        }
726        Ok(())
727    }
728}
729
730impl Validate for FilterTestRequest {
731    fn validate(&self) -> Result<(), InvalidError> {
732        validate_version(self.v)?;
733        self.filter.validate()?;
734        if matches!(self.filter, FilterRef::Bound) {
735            return Err(InvalidError::new(
736                "a sample test executes an inline filter or a saved revision",
737            ));
738        }
739        if self.payload.len() > MAX_FILTER_SAMPLE_BYTES {
740            return Err(InvalidError::new(format!(
741                "a sample payload exceeds {MAX_FILTER_SAMPLE_BYTES}B"
742            )));
743        }
744        if self.headers.len() > MAX_FILTER_SAMPLE_HEADERS {
745            return Err(InvalidError::new(format!(
746                "a sample carries more than {MAX_FILTER_SAMPLE_HEADERS} headers"
747            )));
748        }
749        for header in &self.headers {
750            validate_name("sample header key", &header.key)?;
751            let bytes = match &header.value {
752                HeaderScalar::String(value) => value.len(),
753                HeaderScalar::Raw(value) => value.len(),
754                _ => 0,
755            };
756            if bytes > crate::limits::MAX_FILTER_STRING_BYTES {
757                return Err(InvalidError::new(
758                    "a sample header value exceeds its byte limit",
759                ));
760            }
761        }
762        Ok(())
763    }
764}
765
766#[cfg(test)]
767mod tests {
768    use super::*;
769    use crate::filter::expr::FilterExpr;
770    use crate::query::CmpOp;
771
772    fn generation() -> SourceGeneration {
773        SourceGeneration {
774            stream_id: 1,
775            stream_created_at_micros: 10,
776            topic_id: 2,
777            topic_created_at_micros: 20,
778            partition_id: 0,
779            partition_created_revision: 30,
780            purge_generation: 0,
781        }
782    }
783
784    fn request() -> FilteredPollRequest {
785        FilteredPollRequest {
786            v: FILTER_OP_VERSION,
787            source: FilterSource {
788                stream: "orbit".to_owned(),
789                topic: "fleet_changes".to_owned(),
790            },
791            partition_id: 0,
792            consumer: FilterConsumer::Group("anomaly-desk".to_owned()),
793            filter: FilterRef::Bound,
794            start: FilteredStart::Next,
795            count: 100,
796            max_reply_bytes: 1024 * 1024,
797            read_mode: ReadMode::Primary,
798        }
799    }
800
801    #[test]
802    fn given_a_group_read_when_validated_then_should_pass() {
803        request().validate().expect("valid");
804    }
805
806    #[test]
807    fn given_out_of_range_reads_when_validated_then_should_be_rejected() {
808        let mut bound_consumer = request();
809        bound_consumer.consumer = FilterConsumer::Consumer("one".to_owned());
810        let mut zero = request();
811        zero.count = 0;
812        let mut huge = request();
813        huge.max_reply_bytes = MAX_FILTERED_PAGE_BYTES + 1;
814        let mut other_partition = request();
815        other_partition.start = FilteredStart::Continue(Continuation {
816            group_id: None,
817            next_scan_offset: 7,
818            generation: SourceGeneration {
819                partition_id: 5,
820                ..generation()
821            },
822            digest: Digest32::new([1; 32]),
823            read_mode: ReadMode::Primary,
824        });
825        for request in [bound_consumer, zero, huge, other_partition] {
826            assert!(request.validate().is_err(), "{request:?} must be rejected");
827        }
828    }
829
830    #[test]
831    fn given_a_page_when_asked_for_the_next_start_then_should_continue_or_repeat() {
832        let page = FilteredPage {
833            v: FILTER_OP_VERSION,
834            partition_id: 0,
835            policy: AppliedPolicy {
836                group_id: None,
837                digest: Digest32::new([9; 32]),
838                filter_id: Some(3),
839                revision: Some(1),
840            },
841            generation: generation(),
842            read_mode: ReadMode::Primary,
843            next_scan_offset: Some(1397),
844            safe_ack_offset: Some(1396),
845            frontier: 2000,
846            examined: 397,
847            matched: 3,
848            stop: StopReason::Filled,
849            fault: None,
850            unevaluated: Vec::new(),
851            evaluation_limits: None,
852            records: Vec::new(),
853        };
854        assert_eq!(
855            page.next_start(&FilteredStart::Next),
856            FilteredStart::Continue(Continuation {
857                group_id: None,
858                next_scan_offset: 1397,
859                generation: generation(),
860                digest: Digest32::new([9; 32]),
861                read_mode: ReadMode::Primary,
862            })
863        );
864        let empty = FilteredPage {
865            next_scan_offset: None,
866            safe_ack_offset: None,
867            ..page
868        };
869        assert_eq!(empty.next_start(&FilteredStart::Last), FilteredStart::Last);
870    }
871
872    #[test]
873    fn given_error_reasons_when_classified_then_should_map_to_shared_codes() {
874        assert_eq!(
875            FilterError::new(FilterErrorReason::SourceChanged, "purged").code,
876            ResultCode::StaleGeneration
877        );
878        assert_eq!(FilterErrorReason::NotPrimary.code(), ResultCode::Stale);
879        assert_eq!(FilterErrorReason::Conflict.code(), ResultCode::Conflict);
880        assert_eq!(
881            FilterErrorReason::CatalogUnavailable.code(),
882            ResultCode::Unavailable
883        );
884    }
885
886    #[test]
887    fn given_samples_and_previews_when_validated_then_should_reject_bound_filters() {
888        let inline = FilterRef::Inline(ConsumerFilter::json(FilterExpr::pred(
889            "kind",
890            CmpOp::Eq,
891            "satellites",
892        )));
893        let test = FilterTestRequest {
894            v: FILTER_OP_VERSION,
895            filter: inline.clone(),
896            payload: br#"{"kind":"satellites"}"#.to_vec(),
897            headers: Vec::new(),
898        };
899        test.validate().expect("valid sample");
900        let bound = FilterTestRequest {
901            filter: FilterRef::Bound,
902            ..test
903        };
904        assert!(bound.validate().is_err());
905        let preview = FilterPreviewRequest {
906            v: FILTER_OP_VERSION,
907            source: request().source,
908            partition_id: 0,
909            filter: inline,
910            from_offset: 0,
911            max_examined: MAX_FILTER_PREVIEW_EXAMINED + 1,
912            max_records: 10,
913            explain: false,
914        };
915        assert!(preview.validate().is_err());
916    }
917
918    #[test]
919    fn given_oversized_sample_headers_when_validated_then_should_reject_before_evaluation() {
920        let mut request = FilterTestRequest {
921            v: FILTER_OP_VERSION,
922            filter: FilterRef::Inline(ConsumerFilter::headers_only(FilterExpr::header(
923                "priority",
924                CmpOp::Eq,
925                "critical",
926            ))),
927            payload: Vec::new(),
928            headers: vec![FilterHeader {
929                key: "priority".to_owned(),
930                value: HeaderScalar::Raw(vec![0; crate::limits::MAX_FILTER_STRING_BYTES + 1]),
931            }],
932        };
933        assert!(request.validate().is_err());
934        request.headers[0].value = HeaderScalar::String("critical".to_owned());
935        request.validate().expect("bounded header");
936        request.headers[0].key.clear();
937        assert!(request.validate().is_err());
938    }
939}
laser_wire/filter/read.rs

laser_wire/filter/
read.rs
