laser_wire/
source.rs

1use crate::agent::wire_id;
2use crate::error::InvalidError;
3use crate::validate::Validate;
4use serde::{Deserialize, Serialize};
5use std::collections::BTreeSet;
6
7pub const MAX_SOURCE_PARTITIONS: usize = 65_536;
8
9wire_id!(
10    /// Deployment-provisioned identity of one physical Iggy cluster.
11    PhysicalClusterIncarnation
12);
13
14#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
15pub struct SourceIncarnation {
16    pub cluster: PhysicalClusterIncarnation,
17    pub stream_id: u32,
18    pub topic_id: u32,
19    pub partition_id: u32,
20    pub partition_created_revision: u64,
21}
22
23impl Validate for SourceIncarnation {
24    fn validate(&self) -> Result<(), InvalidError> {
25        if self.cluster.as_u128() == 0 {
26            return Err(InvalidError::new(
27                "physical cluster incarnation must be nonzero",
28            ));
29        }
30        if self.partition_created_revision == 0 {
31            return Err(InvalidError::new(
32                "partition created revision must be nonzero",
33            ));
34        }
35        Ok(())
36    }
37}
38
39#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
40pub struct SourceScope {
41    pub stream: String,
42    pub topic: String,
43}
44
45impl SourceScope {
46    pub fn new(stream: impl Into<String>, topic: impl Into<String>) -> Self {
47        Self {
48            stream: stream.into(),
49            topic: topic.into(),
50        }
51    }
52}
53
54impl Validate for SourceScope {
55    fn validate(&self) -> Result<(), InvalidError> {
56        validate_source_name("stream", &self.stream)?;
57        validate_source_name("topic", &self.topic)
58    }
59}
60
61#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
62pub struct SourcePartitionCut {
63    pub incarnation: SourceIncarnation,
64    pub retained_start: u64,
65    pub end_exclusive: u64,
66}
67
68impl Validate for SourcePartitionCut {
69    fn validate(&self) -> Result<(), InvalidError> {
70        self.incarnation.validate()?;
71        if self.retained_start > self.end_exclusive {
72            return Err(InvalidError::new(format!(
73                "retained start {} is above end-exclusive head {}",
74                self.retained_start, self.end_exclusive
75            )));
76        }
77        Ok(())
78    }
79}
80
81#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
82pub struct SourceCut {
83    pub partitions: Vec<SourcePartitionCut>,
84}
85
86impl Validate for SourceCut {
87    fn validate(&self) -> Result<(), InvalidError> {
88        if self.partitions.is_empty() || self.partitions.len() > MAX_SOURCE_PARTITIONS {
89            return Err(InvalidError::new(format!(
90                "source cut must contain 1..={MAX_SOURCE_PARTITIONS} partitions"
91            )));
92        }
93        let mut partitions = BTreeSet::new();
94        let mut previous_partition = None;
95        let mut namespace = None;
96        for partition in &self.partitions {
97            partition.validate()?;
98            let current = (
99                partition.incarnation.cluster,
100                partition.incarnation.stream_id,
101                partition.incarnation.topic_id,
102            );
103            if namespace.is_some_and(|expected| expected != current) {
104                return Err(InvalidError::new(
105                    "source cut partitions must share one cluster, stream, and topic",
106                ));
107            }
108            namespace = Some(current);
109            if !partitions.insert(partition.incarnation.partition_id) {
110                return Err(InvalidError::new(format!(
111                    "source cut repeats partition {}",
112                    partition.incarnation.partition_id
113                )));
114            }
115            if previous_partition
116                .is_some_and(|previous| previous >= partition.incarnation.partition_id)
117            {
118                return Err(InvalidError::new(
119                    "source cut partitions must be ordered by ascending partition id",
120                ));
121            }
122            previous_partition = Some(partition.incarnation.partition_id);
123        }
124        Ok(())
125    }
126}
127
128fn validate_source_name(label: &str, value: &str) -> Result<(), InvalidError> {
129    if value.is_empty() {
130        return Err(InvalidError::new(format!("{label} must not be empty")));
131    }
132    if value.len() > 255 {
133        return Err(InvalidError::new(format!(
134            "{label} is {} bytes, exceeds cap 255",
135            value.len()
136        )));
137    }
138    if value.chars().any(char::is_control) {
139        return Err(InvalidError::new(format!(
140            "{label} contains a control character"
141        )));
142    }
143    Ok(())
144}
145
146#[cfg(test)]
147mod tests {
148    use super::*;
149
150    fn incarnation(partition_id: u32, created_revision: u64) -> SourceIncarnation {
151        SourceIncarnation {
152            cluster: PhysicalClusterIncarnation::from_u128(1),
153            stream_id: 10,
154            topic_id: 20,
155            partition_id,
156            partition_created_revision: created_revision,
157        }
158    }
159
160    #[test]
161    fn given_a_recreated_partition_when_incarnations_are_compared_then_should_differ() {
162        assert_ne!(incarnation(1, 7), incarnation(1, 8));
163        incarnation(1, 7).validate().expect("incarnation is valid");
164    }
165
166    #[test]
167    fn given_duplicate_partitions_or_inverted_heads_when_a_source_cut_is_validated_then_should_reject()
168     {
169        let duplicate = SourceCut {
170            partitions: vec![
171                SourcePartitionCut {
172                    incarnation: incarnation(1, 7),
173                    retained_start: 0,
174                    end_exclusive: 10,
175                },
176                SourcePartitionCut {
177                    incarnation: incarnation(1, 7),
178                    retained_start: 2,
179                    end_exclusive: 10,
180                },
181            ],
182        };
183        assert!(duplicate.validate().is_err());
184
185        let inverted = SourcePartitionCut {
186            incarnation: incarnation(1, 7),
187            retained_start: 11,
188            end_exclusive: 10,
189        };
190        assert!(inverted.validate().is_err());
191    }
192}
laser_wire/source.rs

laser_wire/
source.rs
