oxide_batch_core/
chunk.rs

1//! Chunk sizing, counting, and progress values.
2
3use std::error::Error;
4use std::fmt;
5use std::num::NonZeroU32;
6
7/// A nonzero item limit for one chunk.
8#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
9pub struct ChunkSize(NonZeroU32);
10
11impl ChunkSize {
12    /// Constructs a nonzero chunk size.
13    ///
14    /// # Errors
15    ///
16    /// Returns [`ChunkError::ZeroSize`] when `value` is zero.
17    pub fn new(value: u32) -> Result<Self, ChunkError> {
18        NonZeroU32::new(value).map(Self).ok_or(ChunkError::ZeroSize)
19    }
20
21    /// Returns the configured item limit.
22    #[must_use]
23    pub const fn get(self) -> u32 {
24        self.0.get()
25    }
26}
27
28/// A checked non-negative item or transaction count.
29#[derive(Clone, Copy, Debug, Default, Eq, Hash, Ord, PartialEq, PartialOrd)]
30pub struct ChunkCount(u64);
31
32impl ChunkCount {
33    /// The zero count.
34    pub const ZERO: Self = Self(0);
35
36    /// Constructs a count.
37    #[must_use]
38    pub const fn new(value: u64) -> Self {
39        Self(value)
40    }
41
42    /// Returns the numeric count.
43    #[must_use]
44    pub const fn get(self) -> u64 {
45        self.0
46    }
47
48    /// Adds two counts without wrapping.
49    ///
50    /// # Errors
51    ///
52    /// Returns [`ChunkError::CountOverflow`] when the sum exceeds `u64`.
53    pub fn checked_add(self, other: Self) -> Result<Self, ChunkError> {
54        self.0
55            .checked_add(other.0)
56            .map(Self)
57            .ok_or(ChunkError::CountOverflow)
58    }
59
60    /// Increments this count without wrapping.
61    ///
62    /// # Errors
63    ///
64    /// Returns [`ChunkError::CountOverflow`] at `u64::MAX`.
65    pub fn checked_increment(self) -> Result<Self, ChunkError> {
66        self.checked_add(Self(1))
67    }
68}
69
70/// Validated item counts within one open chunk.
71#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
72pub struct ChunkCounts {
73    read: ChunkCount,
74    processed: ChunkCount,
75    written: ChunkCount,
76    filtered: ChunkCount,
77}
78
79impl ChunkCounts {
80    /// Validates a complete count snapshot.
81    ///
82    /// `processed` counts items that produced writer input; `filtered` counts
83    /// items intentionally producing no output. Their sum cannot exceed
84    /// `read`, and `written` cannot exceed `processed`.
85    ///
86    /// # Errors
87    ///
88    /// Returns a typed overflow or invalid-state classification.
89    pub fn new(
90        read: ChunkCount,
91        processed: ChunkCount,
92        written: ChunkCount,
93        filtered: ChunkCount,
94    ) -> Result<Self, ChunkError> {
95        let classified = processed.checked_add(filtered)?;
96        if classified > read {
97            return Err(ChunkError::ClassifiedExceedsRead);
98        }
99        if written > processed {
100            return Err(ChunkError::WrittenExceedsProcessed);
101        }
102        Ok(Self {
103            read,
104            processed,
105            written,
106            filtered,
107        })
108    }
109
110    /// Returns the read count.
111    #[must_use]
112    pub const fn read(self) -> ChunkCount {
113        self.read
114    }
115
116    /// Returns the successfully processed count.
117    #[must_use]
118    pub const fn processed(self) -> ChunkCount {
119        self.processed
120    }
121
122    /// Returns the acknowledged writer-input count.
123    #[must_use]
124    pub const fn written(self) -> ChunkCount {
125        self.written
126    }
127
128    /// Returns the filtered count.
129    #[must_use]
130    pub const fn filtered(self) -> ChunkCount {
131        self.filtered
132    }
133
134    /// Adds two snapshots and revalidates their aggregate invariants.
135    ///
136    /// # Errors
137    ///
138    /// Returns a typed overflow or invalid-state classification.
139    pub fn checked_add(self, other: Self) -> Result<Self, ChunkError> {
140        Self::new(
141            self.read.checked_add(other.read)?,
142            self.processed.checked_add(other.processed)?,
143            self.written.checked_add(other.written)?,
144            self.filtered.checked_add(other.filtered)?,
145        )
146    }
147}
148
149/// Mutable, invariant-preserving progress for one bounded chunk.
150#[derive(Clone, Copy, Debug, Eq, PartialEq)]
151pub struct ChunkProgress {
152    size: ChunkSize,
153    counts: ChunkCounts,
154}
155
156impl ChunkProgress {
157    /// Starts an empty chunk with a validated nonzero size.
158    #[must_use]
159    pub const fn new(size: ChunkSize) -> Self {
160        Self {
161            size,
162            counts: ChunkCounts {
163                read: ChunkCount::ZERO,
164                processed: ChunkCount::ZERO,
165                written: ChunkCount::ZERO,
166                filtered: ChunkCount::ZERO,
167            },
168        }
169    }
170
171    /// Restores a validated in-memory progress snapshot.
172    ///
173    /// # Errors
174    ///
175    /// Returns [`ChunkError::SizeExceeded`] when `counts.read()` exceeds the
176    /// configured chunk size.
177    pub fn from_counts(size: ChunkSize, counts: ChunkCounts) -> Result<Self, ChunkError> {
178        if counts.read().get() > u64::from(size.get()) {
179            return Err(ChunkError::SizeExceeded);
180        }
181        Ok(Self { size, counts })
182    }
183
184    /// Returns the configured size.
185    #[must_use]
186    pub const fn size(self) -> ChunkSize {
187        self.size
188    }
189
190    /// Returns the current validated counts.
191    #[must_use]
192    pub const fn counts(self) -> ChunkCounts {
193        self.counts
194    }
195
196    /// Returns whether no more items may be read into this chunk.
197    #[must_use]
198    pub fn is_full(self) -> bool {
199        self.counts.read().get() == u64::from(self.size.get())
200    }
201
202    /// Records one successfully read item.
203    ///
204    /// # Errors
205    ///
206    /// Returns [`ChunkError::SizeExceeded`] when the chunk is already full,
207    /// or [`ChunkError::CountOverflow`] on arithmetic exhaustion.
208    pub fn record_read(&mut self) -> Result<(), ChunkError> {
209        if self.is_full() {
210            return Err(ChunkError::SizeExceeded);
211        }
212        self.counts.read = self.counts.read.checked_increment()?;
213        Ok(())
214    }
215
216    /// Classifies one previously read item as successfully processed.
217    ///
218    /// # Errors
219    ///
220    /// Returns [`ChunkError::ClassifiedExceedsRead`] when no unclassified read
221    /// item remains, or [`ChunkError::CountOverflow`] on exhaustion.
222    pub fn record_processed(&mut self) -> Result<(), ChunkError> {
223        let next = self.counts.processed.checked_increment()?;
224        let classified = next.checked_add(self.counts.filtered)?;
225        if classified > self.counts.read {
226            return Err(ChunkError::ClassifiedExceedsRead);
227        }
228        self.counts.processed = next;
229        Ok(())
230    }
231
232    /// Classifies one previously read item as filtered.
233    ///
234    /// # Errors
235    ///
236    /// Returns [`ChunkError::ClassifiedExceedsRead`] when no unclassified read
237    /// item remains, or [`ChunkError::CountOverflow`] on exhaustion.
238    pub fn record_filtered(&mut self) -> Result<(), ChunkError> {
239        let next = self.counts.filtered.checked_increment()?;
240        let classified = self.counts.processed.checked_add(next)?;
241        if classified > self.counts.read {
242            return Err(ChunkError::ClassifiedExceedsRead);
243        }
244        self.counts.filtered = next;
245        Ok(())
246    }
247
248    /// Records writer acknowledgement for `count` processed items.
249    ///
250    /// # Errors
251    ///
252    /// Returns [`ChunkError::WrittenExceedsProcessed`] when the aggregate
253    /// exceeds processed output, or [`ChunkError::CountOverflow`] on
254    /// exhaustion.
255    pub fn record_written(&mut self, count: ChunkCount) -> Result<(), ChunkError> {
256        let next = self.counts.written.checked_add(count)?;
257        if next > self.counts.processed {
258            return Err(ChunkError::WrittenExceedsProcessed);
259        }
260        self.counts.written = next;
261        Ok(())
262    }
263}
264
265/// Stable chunk-size and count failure.
266#[derive(Clone, Copy, Debug, Eq, PartialEq)]
267#[non_exhaustive]
268pub enum ChunkError {
269    /// Chunk size was zero.
270    ZeroSize,
271    /// Count arithmetic exceeded `u64`.
272    CountOverflow,
273    /// Processed plus filtered count exceeded read count.
274    ClassifiedExceedsRead,
275    /// Written count exceeded successfully processed count.
276    WrittenExceedsProcessed,
277    /// Read count exceeded the configured chunk size.
278    SizeExceeded,
279}
280
281impl fmt::Display for ChunkError {
282    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
283        formatter.write_str(match self {
284            Self::ZeroSize => "chunk size must be nonzero",
285            Self::CountOverflow => "chunk count arithmetic overflowed",
286            Self::ClassifiedExceedsRead => "processed and filtered counts exceed the read count",
287            Self::WrittenExceedsProcessed => "written count exceeds the processed count",
288            Self::SizeExceeded => "read count exceeds the configured chunk size",
289        })
290    }
291}
292
293impl Error for ChunkError {}
oxide_batch_core/chunk.rs

oxide_batch_core/
chunk.rs
