vector_common/
finalization.rs

1#![deny(missing_docs)]
2//! This module contains the event metadata required to track an event
3//! as it flows through transforms, being duplicated and merged, and
4//! then report its status when the last copy is delivered or dropped.
5
6use std::{cmp, future::Future, mem, pin::Pin, sync::Arc, task::Poll};
7
8use crossbeam_utils::atomic::AtomicCell;
9use futures::future::FutureExt;
10use tokio::sync::oneshot;
11
12#[cfg(feature = "byte_size_of")]
13use crate::byte_size_of::ByteSizeOf;
14
15/// A collection of event finalizers.
16#[derive(Clone, Debug, Default)]
17pub struct EventFinalizers(Vec<Arc<EventFinalizer>>);
18
19impl Eq for EventFinalizers {}
20
21impl PartialEq for EventFinalizers {
22    fn eq(&self, other: &Self) -> bool {
23        self.0.len() == other.0.len()
24            && (self.0.iter())
25                .zip(other.0.iter())
26                .all(|(a, b)| Arc::ptr_eq(a, b))
27    }
28}
29
30impl PartialOrd for EventFinalizers {
31    fn partial_cmp(&self, other: &Self) -> Option<cmp::Ordering> {
32        // There is no partial order defined structurally on
33        // `EventFinalizer`. Partial equality is defined on the equality of
34        // `Arc`s. Therefore, partial ordering of `EventFinalizers` is defined
35        // only on the length of the finalizers.
36        self.0.len().partial_cmp(&other.0.len())
37    }
38}
39
40#[cfg(feature = "byte_size_of")]
41impl ByteSizeOf for EventFinalizers {
42    fn allocated_bytes(&self) -> usize {
43        // Don't count the allocated data here, it's not really event
44        // data we're interested in tracking but rather an artifact of
45        // tracking and merging events.
46        0
47    }
48}
49
50impl EventFinalizers {
51    /// Default empty finalizer set for use in `const` contexts.
52    pub const DEFAULT: Self = Self(Vec::new());
53
54    /// Creates a new `EventFinalizers` based on the given event finalizer.
55    #[must_use]
56    pub fn new(finalizer: EventFinalizer) -> Self {
57        Self(vec![Arc::new(finalizer)])
58    }
59
60    /// Returns `true` if the collection contains no event finalizers.
61    #[must_use]
62    pub fn is_empty(&self) -> bool {
63        self.0.is_empty()
64    }
65
66    /// Returns the number of event finalizers in the collection.
67    #[must_use]
68    pub fn len(&self) -> usize {
69        self.0.len()
70    }
71
72    /// Adds a new event finalizer to the collection.
73    pub fn add(&mut self, finalizer: EventFinalizer) {
74        self.0.push(Arc::new(finalizer));
75    }
76
77    /// Merges the event finalizers from `other` into the collection.
78    pub fn merge(&mut self, other: Self) {
79        self.0.extend(other.0);
80    }
81
82    /// Updates the status of all event finalizers in the collection.
83    pub fn update_status(&self, status: EventStatus) {
84        for finalizer in &self.0 {
85            finalizer.update_status(status);
86        }
87    }
88
89    /// Consumes all event finalizers and updates their underlying batches immediately.
90    pub fn update_sources(&mut self) {
91        let finalizers = mem::take(&mut self.0);
92        for finalizer in &finalizers {
93            finalizer.update_batch();
94        }
95    }
96}
97
98impl Finalizable for EventFinalizers {
99    fn take_finalizers(&mut self) -> EventFinalizers {
100        mem::take(self)
101    }
102}
103
104impl MergeFinalizable for EventFinalizers {
105    fn merge_finalizers(&mut self, finalizers: EventFinalizers) {
106        self.merge(finalizers);
107    }
108}
109
110impl std::iter::FromIterator<EventFinalizers> for EventFinalizers {
111    fn from_iter<T: IntoIterator<Item = EventFinalizers>>(iter: T) -> Self {
112        Self(iter.into_iter().flat_map(|f| f.0.into_iter()).collect())
113    }
114}
115
116/// Ordered groups of event finalizers.
117///
118/// This is used when a bufferable record contains multiple independently-finalized events and the
119/// finalizers must be removed before the record is consumed. The group order matches the event
120/// order, allowing finalizers to be reattached to their original event if the record is recovered
121/// for retry.
122#[derive(Clone, Debug, Default, Eq, PartialEq)]
123pub struct EventFinalizerGroups(Vec<EventFinalizers>);
124
125impl EventFinalizerGroups {
126    /// Creates grouped finalizers from a single flat finalizer collection.
127    #[must_use]
128    pub fn from_flat(finalizers: EventFinalizers) -> Self {
129        Self(vec![finalizers])
130    }
131
132    /// Creates grouped finalizers from ordered per-event finalizer collections.
133    #[must_use]
134    pub fn from_groups(groups: Vec<EventFinalizers>) -> Self {
135        Self(groups)
136    }
137
138    /// Returns the number of finalizer groups.
139    #[must_use]
140    pub fn len(&self) -> usize {
141        self.0.len()
142    }
143
144    /// Returns `true` if there are no finalizer groups.
145    #[must_use]
146    pub fn is_empty(&self) -> bool {
147        self.0.is_empty()
148    }
149
150    /// Consumes this value and returns the ordered finalizer groups.
151    #[must_use]
152    pub fn into_groups(self) -> Vec<EventFinalizers> {
153        self.0
154    }
155
156    /// Consumes this value and flattens all groups into one finalizer collection.
157    #[must_use]
158    pub fn flatten(self) -> EventFinalizers {
159        self.0.into_iter().collect()
160    }
161
162    /// Updates the status of all finalizers in all groups.
163    pub fn update_status(&self, status: EventStatus) {
164        for group in &self.0 {
165            group.update_status(status);
166        }
167    }
168}
169
170impl FromIterator<EventFinalizers> for EventFinalizerGroups {
171    fn from_iter<T: IntoIterator<Item = EventFinalizers>>(iter: T) -> Self {
172        EventFinalizerGroups(Vec::from_iter(iter))
173    }
174}
175
176/// An event finalizer is the shared data required to handle tracking the status of an event, and updating the status of
177/// a batch with that when the event is dropped.
178#[derive(Debug)]
179pub struct EventFinalizer {
180    status: AtomicCell<EventStatus>,
181    batch: BatchNotifier,
182}
183
184#[cfg(feature = "byte_size_of")]
185impl ByteSizeOf for EventFinalizer {
186    fn allocated_bytes(&self) -> usize {
187        // Don't count the batch notifier, as it's shared across
188        // events in a batch.
189        0
190    }
191}
192
193impl EventFinalizer {
194    /// Creates a new `EventFinalizer` attached to the given `batch`.
195    #[must_use]
196    pub fn new(batch: BatchNotifier) -> Self {
197        let status = AtomicCell::new(EventStatus::Dropped);
198        Self { status, batch }
199    }
200
201    /// Updates the status of the event finalizer to `status`.
202    pub fn update_status(&self, status: EventStatus) {
203        self.status
204            .fetch_update(|old_status| Some(old_status.update(status)))
205            .unwrap_or_else(|_| unreachable!());
206    }
207
208    /// Updates the underlying batch status with the status of the event finalizer.
209    ///
210    /// In doing so, the event finalizer is marked as "recorded", which prevents any further updates to it.
211    pub fn update_batch(&self) {
212        let status = self
213            .status
214            .fetch_update(|_| Some(EventStatus::Recorded))
215            .unwrap_or_else(|_| unreachable!());
216        self.batch.update_status(status);
217    }
218}
219
220impl Drop for EventFinalizer {
221    fn drop(&mut self) {
222        self.update_batch();
223    }
224}
225
226/// A convenience newtype wrapper for the one-shot receiver for an
227/// individual batch status.
228#[pin_project::pin_project]
229pub struct BatchStatusReceiver(oneshot::Receiver<BatchStatus>);
230
231impl Future for BatchStatusReceiver {
232    type Output = BatchStatus;
233    fn poll(mut self: Pin<&mut Self>, ctx: &mut std::task::Context<'_>) -> Poll<Self::Output> {
234        match self.0.poll_unpin(ctx) {
235            Poll::Pending => Poll::Pending,
236            Poll::Ready(Ok(status)) => Poll::Ready(status),
237            Poll::Ready(Err(error)) => {
238                error!(%error, "Batch status receiver dropped before sending.");
239                Poll::Ready(BatchStatus::Errored)
240            }
241        }
242    }
243}
244
245impl BatchStatusReceiver {
246    /// Wrapper for the underlying `try_recv` function.
247    ///
248    /// # Errors
249    ///
250    /// - `TryRecvError::Empty` if no value has been sent yet.
251    /// - `TryRecvError::Closed` if the sender has dropped without sending a value.
252    pub fn try_recv(&mut self) -> Result<BatchStatus, oneshot::error::TryRecvError> {
253        self.0.try_recv()
254    }
255}
256
257/// A batch notifier contains the status of the current batch along with
258/// a one-shot notifier to send that status back to the source. It is
259/// shared among all events of a batch.
260#[derive(Clone, Debug)]
261pub struct BatchNotifier(Arc<OwnedBatchNotifier>);
262
263impl BatchNotifier {
264    /// Creates a new `BatchNotifier` along with the receiver used to await its finalization status.
265    #[must_use]
266    pub fn new_with_receiver() -> (Self, BatchStatusReceiver) {
267        let (sender, receiver) = oneshot::channel();
268        let notifier = OwnedBatchNotifier {
269            status: AtomicCell::new(BatchStatus::Delivered),
270            notifier: Some(sender),
271        };
272        (Self(Arc::new(notifier)), BatchStatusReceiver(receiver))
273    }
274
275    /// Optionally creates a new `BatchNotifier` along with the receiver used to await its finalization status.
276    #[must_use]
277    pub fn maybe_new_with_receiver(enabled: bool) -> (Option<Self>, Option<BatchStatusReceiver>) {
278        if enabled {
279            let (batch, receiver) = Self::new_with_receiver();
280            (Some(batch), Some(receiver))
281        } else {
282            (None, None)
283        }
284    }
285
286    /// Creates a new `BatchNotifier` and attaches it to a group of events.
287    ///
288    /// The receiver used to await the finalization status of the batch is returned.
289    pub fn apply_to<T: AddBatchNotifier>(items: &mut [T]) -> BatchStatusReceiver {
290        let (batch, receiver) = Self::new_with_receiver();
291        for item in items {
292            item.add_batch_notifier(batch.clone());
293        }
294        receiver
295    }
296
297    /// Optionally creates a new `BatchNotifier` and attaches it to a group of events.
298    ///
299    /// If `enabled`, the receiver used to await the finalization status of the batch is
300    /// returned. Otherwise, `None` is returned.
301    pub fn maybe_apply_to<T: AddBatchNotifier>(
302        enabled: bool,
303        items: &mut [T],
304    ) -> Option<BatchStatusReceiver> {
305        enabled.then(|| Self::apply_to(items))
306    }
307
308    /// Updates the status of the notifier.
309    fn update_status(&self, status: EventStatus) {
310        // The status starts as Delivered and can only change if the new
311        // status is different than that.
312        if status != EventStatus::Delivered && status != EventStatus::Dropped {
313            self.0
314                .status
315                .fetch_update(|old_status| Some(old_status.update(status)))
316                .unwrap_or_else(|_| unreachable!());
317        }
318    }
319}
320
321/// The non-shared data underlying the shared `BatchNotifier`
322#[derive(Debug)]
323pub struct OwnedBatchNotifier {
324    status: AtomicCell<BatchStatus>,
325    notifier: Option<oneshot::Sender<BatchStatus>>,
326}
327
328impl OwnedBatchNotifier {
329    /// Sends the status of the notifier back to the source.
330    fn send_status(&mut self) {
331        if let Some(notifier) = self.notifier.take() {
332            let status = self.status.load();
333            // Ignore the error case, as it will happen during normal
334            // source shutdown and we can't detect that here.
335            _ = notifier.send(status);
336        }
337    }
338}
339
340impl Drop for OwnedBatchNotifier {
341    fn drop(&mut self) {
342        self.send_status();
343    }
344}
345
346/// The status of an individual batch.
347#[derive(Copy, Clone, Debug, Eq, PartialEq)]
348#[repr(u8)]
349#[derive(Default)]
350pub enum BatchStatus {
351    /// All events in the batch were accepted.
352    ///
353    /// This is the default.
354    #[default]
355    Delivered,
356    /// At least one event in the batch had a transient error in delivery.
357    Errored,
358    /// At least one event in the batch had a permanent failure or rejection.
359    Rejected,
360}
361
362impl BatchStatus {
363    /// Updates the delivery status based on another batch's delivery status, returning the result.
364    ///
365    /// As not every status has the same priority, some updates may end up being a no-op either due to not being any
366    /// different or due to being lower priority than the current status.
367    #[allow(clippy::match_same_arms)] // False positive: https://github.com/rust-lang/rust-clippy/issues/860
368    fn update(self, status: EventStatus) -> Self {
369        match (self, status) {
370            // `Dropped` and `Delivered` do not change the status.
371            (_, EventStatus::Dropped | EventStatus::Delivered) => self,
372            // `Rejected` overrides `Errored` and `Delivered`
373            (Self::Rejected, _) | (_, EventStatus::Rejected) => Self::Rejected,
374            // `Errored` overrides `Delivered`
375            (Self::Errored, _) | (_, EventStatus::Errored) => Self::Errored,
376            // No change for `Delivered`
377            _ => self,
378        }
379    }
380}
381
382/// The status of an individual event.
383#[derive(Copy, Clone, Debug, Eq, PartialEq)]
384#[repr(u8)]
385#[derive(Default)]
386pub enum EventStatus {
387    /// All copies of this event were dropped without being finalized.
388    ///
389    /// This is the default.
390    #[default]
391    Dropped,
392    /// All copies of this event were delivered successfully.
393    Delivered,
394    /// At least one copy of this event encountered a retriable error.
395    Errored,
396    /// At least one copy of this event encountered a permanent failure or rejection.
397    Rejected,
398    /// This status has been recorded and should not be updated.
399    Recorded,
400}
401
402impl EventStatus {
403    /// Updates the status based on another event's status, returning the result.
404    ///
405    /// As not every status has the same priority, some updates may end up being a no-op either due to not being any
406    /// different or due to being lower priority than the current status.
407    ///
408    /// # Panics
409    ///
410    /// Passing a new status of `Dropped` is a programming error and will panic in debug/test builds.
411    #[allow(clippy::match_same_arms)] // False positive: https://github.com/rust-lang/rust-clippy/issues/860
412    #[must_use]
413    pub fn update(self, status: Self) -> Self {
414        match (self, status) {
415            // `Recorded` always overwrites existing status and is never updated
416            (_, Self::Recorded) | (Self::Recorded, _) => Self::Recorded,
417            // `Dropped` always updates to the new status.
418            (Self::Dropped, _) => status,
419            // Updates *to* `Dropped` are nonsense.
420            (_, Self::Dropped) => {
421                debug_assert!(false, "Updating EventStatus to Dropped is nonsense");
422                self
423            }
424            // `Rejected` overrides `Errored` or `Delivered`.
425            (Self::Rejected, _) | (_, Self::Rejected) => Self::Rejected,
426            // `Errored` overrides `Delivered`.
427            (Self::Errored, _) | (_, Self::Errored) => Self::Errored,
428            // No change for `Delivered`.
429            (Self::Delivered, Self::Delivered) => Self::Delivered,
430        }
431    }
432}
433
434/// An object to which we can add a batch notifier.
435pub trait AddBatchNotifier {
436    /// Adds a single shared batch notifier to this type.
437    fn add_batch_notifier(&mut self, notifier: BatchNotifier);
438}
439
440/// An object that can be finalized.
441pub trait Finalizable {
442    /// Consumes the finalizers of this object.
443    ///
444    /// Typically used for coalescing the finalizers of multiple items, such as when batching finalizable objects where
445    /// all finalizations will be processed when the batch itself is processed.
446    fn take_finalizers(&mut self) -> EventFinalizers;
447
448    /// Consumes this object's finalizers while preserving independent finalizer groups.
449    ///
450    /// The default implementation treats this object as a single finalization group. Container
451    /// types that can later split into independently-finalized events should override this method
452    /// and return one group per event.
453    fn take_finalizer_groups(&mut self) -> EventFinalizerGroups {
454        EventFinalizerGroups::from_flat(self.take_finalizers())
455    }
456}
457
458/// A scalar [`Finalizable`] object whose finalizers can be reattached after being taken.
459pub trait MergeFinalizable: Finalizable {
460    /// Merges the given finalizers back into this object.
461    fn merge_finalizers(&mut self, finalizers: EventFinalizers);
462}
463
464/// A [`Finalizable`] buffer record whose grouped finalizers can be reattached after being taken.
465///
466/// Used to reattach finalizers to a record that is being returned for retry, such as when a
467/// `disk_v2` write is rejected because the buffer is full. Scalar records receive this behavior
468/// automatically through [`MergeFinalizable`]. Containers that can later split into independently
469/// finalized events must implement this trait directly to preserve each finalizer group's owner.
470pub trait GroupedFinalizable: Finalizable {
471    /// Merges grouped finalizers back into this object.
472    fn merge_finalizer_groups(&mut self, finalizers: EventFinalizerGroups);
473}
474
475impl<T: MergeFinalizable> GroupedFinalizable for T {
476    fn merge_finalizer_groups(&mut self, finalizers: EventFinalizerGroups) {
477        self.merge_finalizers(finalizers.flatten());
478    }
479}
480
481impl<T: Finalizable> Finalizable for Vec<T> {
482    fn take_finalizers(&mut self) -> EventFinalizers {
483        self.iter_mut()
484            .fold(EventFinalizers::default(), |mut acc, x| {
485                acc.merge(x.take_finalizers());
486                acc
487            })
488    }
489}
490
491#[cfg(test)]
492mod tests {
493    use tokio::sync::oneshot::error::TryRecvError::Empty;
494
495    use super::*;
496
497    #[test]
498    fn defaults() {
499        let finalizer = EventFinalizers::default();
500        assert_eq!(finalizer.len(), 0);
501    }
502
503    #[test]
504    fn sends_notification() {
505        let (fin, mut receiver) = make_finalizer();
506        assert_eq!(receiver.try_recv(), Err(Empty));
507        drop(fin);
508        assert_eq!(receiver.try_recv(), Ok(BatchStatus::Delivered));
509    }
510
511    #[test]
512    fn early_update() {
513        let (mut fin, mut receiver) = make_finalizer();
514        fin.update_status(EventStatus::Rejected);
515        assert_eq!(receiver.try_recv(), Err(Empty));
516        fin.update_sources();
517        assert_eq!(fin.len(), 0);
518        assert_eq!(receiver.try_recv(), Ok(BatchStatus::Rejected));
519    }
520
521    #[test]
522    fn clone_events() {
523        let (fin1, mut receiver) = make_finalizer();
524        let fin2 = fin1.clone();
525        assert_eq!(fin1.len(), 1);
526        assert_eq!(fin2.len(), 1);
527        assert_eq!(fin1, fin2);
528
529        assert_eq!(receiver.try_recv(), Err(Empty));
530        drop(fin1);
531        assert_eq!(receiver.try_recv(), Err(Empty));
532        drop(fin2);
533        assert_eq!(receiver.try_recv(), Ok(BatchStatus::Delivered));
534    }
535
536    #[test]
537    fn merge_events() {
538        let mut fin0 = EventFinalizers::default();
539        let (fin1, mut receiver1) = make_finalizer();
540        let (fin2, mut receiver2) = make_finalizer();
541
542        assert_eq!(fin0.len(), 0);
543        fin0.merge(fin1);
544        assert_eq!(fin0.len(), 1);
545        fin0.merge(fin2);
546        assert_eq!(fin0.len(), 2);
547
548        assert_eq!(receiver1.try_recv(), Err(Empty));
549        assert_eq!(receiver2.try_recv(), Err(Empty));
550        drop(fin0);
551        assert_eq!(receiver1.try_recv(), Ok(BatchStatus::Delivered));
552        assert_eq!(receiver2.try_recv(), Ok(BatchStatus::Delivered));
553    }
554
555    #[test]
556    fn scalar_grouped_merge_flattens_groups() {
557        let (fin1, mut receiver1) = make_finalizer();
558        let (fin2, mut receiver2) = make_finalizer();
559        let mut merged = EventFinalizers::default();
560
561        merged.merge_finalizer_groups(EventFinalizerGroups::from_groups(vec![fin1, fin2]));
562        assert_eq!(merged.len(), 2);
563        assert_eq!(receiver1.try_recv(), Err(Empty));
564        assert_eq!(receiver2.try_recv(), Err(Empty));
565
566        drop(merged);
567        assert_eq!(receiver1.try_recv(), Ok(BatchStatus::Delivered));
568        assert_eq!(receiver2.try_recv(), Ok(BatchStatus::Delivered));
569    }
570
571    #[ignore = "The current implementation does not deduplicate finalizers"]
572    #[test]
573    fn clone_and_merge_events() {
574        let (mut fin1, mut receiver) = make_finalizer();
575        let fin2 = fin1.clone();
576        fin1.merge(fin2);
577        assert_eq!(fin1.len(), 1);
578
579        assert_eq!(receiver.try_recv(), Err(Empty));
580        drop(fin1);
581        assert_eq!(receiver.try_recv(), Ok(BatchStatus::Delivered));
582    }
583
584    #[test]
585    fn multi_event_batch() {
586        let (batch, mut receiver) = BatchNotifier::new_with_receiver();
587        let event1 = EventFinalizers::new(EventFinalizer::new(batch.clone()));
588        let mut event2 = EventFinalizers::new(EventFinalizer::new(batch.clone()));
589        let event3 = EventFinalizers::new(EventFinalizer::new(batch.clone()));
590        // Also clone one…
591        let event4 = event1.clone();
592        drop(batch);
593        assert_eq!(event1.len(), 1);
594        assert_eq!(event2.len(), 1);
595        assert_eq!(event3.len(), 1);
596        assert_eq!(event4.len(), 1);
597        assert_ne!(event1, event2);
598        assert_ne!(event1, event3);
599        assert_eq!(event1, event4);
600        assert_ne!(event2, event3);
601        assert_ne!(event2, event4);
602        assert_ne!(event3, event4);
603        // …and merge another
604        event2.merge(event3);
605        assert_eq!(event2.len(), 2);
606
607        assert_eq!(receiver.try_recv(), Err(Empty));
608        drop(event1);
609        assert_eq!(receiver.try_recv(), Err(Empty));
610        drop(event2);
611        assert_eq!(receiver.try_recv(), Err(Empty));
612        drop(event4);
613        assert_eq!(receiver.try_recv(), Ok(BatchStatus::Delivered));
614    }
615
616    fn make_finalizer() -> (EventFinalizers, BatchStatusReceiver) {
617        let (batch, receiver) = BatchNotifier::new_with_receiver();
618        let finalizer = EventFinalizers::new(EventFinalizer::new(batch));
619        assert_eq!(finalizer.len(), 1);
620        (finalizer, receiver)
621    }
622
623    #[test]
624    fn event_status_updates() {
625        use EventStatus::{Delivered, Dropped, Errored, Recorded, Rejected};
626
627        assert_eq!(Dropped.update(Dropped), Dropped);
628        assert_eq!(Dropped.update(Delivered), Delivered);
629        assert_eq!(Dropped.update(Errored), Errored);
630        assert_eq!(Dropped.update(Rejected), Rejected);
631        assert_eq!(Dropped.update(Recorded), Recorded);
632
633        //assert_eq!(Delivered.update(Dropped), Delivered);
634        assert_eq!(Delivered.update(Delivered), Delivered);
635        assert_eq!(Delivered.update(Errored), Errored);
636        assert_eq!(Delivered.update(Rejected), Rejected);
637        assert_eq!(Delivered.update(Recorded), Recorded);
638
639        //assert_eq!(Errored.update(Dropped), Errored);
640        assert_eq!(Errored.update(Delivered), Errored);
641        assert_eq!(Errored.update(Errored), Errored);
642        assert_eq!(Errored.update(Rejected), Rejected);
643        assert_eq!(Errored.update(Recorded), Recorded);
644
645        //assert_eq!(Rejected.update(Dropped), Rejected);
646        assert_eq!(Rejected.update(Delivered), Rejected);
647        assert_eq!(Rejected.update(Errored), Rejected);
648        assert_eq!(Rejected.update(Rejected), Rejected);
649        assert_eq!(Rejected.update(Recorded), Recorded);
650
651        //assert_eq!(Recorded.update(Dropped), Recorded);
652        assert_eq!(Recorded.update(Delivered), Recorded);
653        assert_eq!(Recorded.update(Errored), Recorded);
654        assert_eq!(Recorded.update(Rejected), Recorded);
655        assert_eq!(Recorded.update(Recorded), Recorded);
656    }
657
658    #[test]
659    fn batch_status_update() {
660        use BatchStatus::{Delivered, Errored, Rejected};
661
662        assert_eq!(Delivered.update(EventStatus::Dropped), Delivered);
663        assert_eq!(Delivered.update(EventStatus::Delivered), Delivered);
664        assert_eq!(Delivered.update(EventStatus::Errored), Errored);
665        assert_eq!(Delivered.update(EventStatus::Rejected), Rejected);
666        assert_eq!(Delivered.update(EventStatus::Recorded), Delivered);
667
668        assert_eq!(Errored.update(EventStatus::Dropped), Errored);
669        assert_eq!(Errored.update(EventStatus::Delivered), Errored);
670        assert_eq!(Errored.update(EventStatus::Errored), Errored);
671        assert_eq!(Errored.update(EventStatus::Rejected), Rejected);
672        assert_eq!(Errored.update(EventStatus::Recorded), Errored);
673
674        assert_eq!(Rejected.update(EventStatus::Dropped), Rejected);
675        assert_eq!(Rejected.update(EventStatus::Delivered), Rejected);
676        assert_eq!(Rejected.update(EventStatus::Errored), Rejected);
677        assert_eq!(Rejected.update(EventStatus::Rejected), Rejected);
678        assert_eq!(Rejected.update(EventStatus::Recorded), Rejected);
679    }
680}
vector_common/finalization.rs

vector_common/
finalization.rs
