krabka_replicator/
ids.rs

1//! Domain newtypes for the replicator crate.
2//!
3//! MirrorMaker-2 offset translation juggles three different `i64` offsets that
4//! all live in different address spaces:
5//!
6//! - [`UpstreamOffset`][]: an offset on the **source** cluster, as recorded in an
7//!   offset-sync or checkpoint.
8//! - [`DownstreamOffset`][]: the corresponding offset on the **target** cluster.
9//! - [`CommittedOffset`][]: a consumer group's committed **source** offset that
10//!   translation moves to the target.
11//!
12//! A call site can transpose these values. For example, it can build an
13//! [`crate::mm2::OffsetSync`] as
14//! `{ upstream: downstream_val, downstream: upstream_val }`, or write the
15//! `translate()` math as `up + (committed - down)`. Such code compiles cleanly
16//! when every value is a bare `i64`, and it silently corrupts offset
17//! translation. A separate type for each meaning turns those swaps into compile
18//! errors.
19//!
20//! The internal MM2 wire codec (`Writer` / `Reader`) still
21//! reads and writes raw `i64`/`i32`, so byte-exactness with the JVM MM2 codecs is
22//! preserved: these newtypes wrap only the in-memory domain values and unwrap via
23//! `.0` at the codec boundary.
24//!
25//! See the [newtype guidance] in the style guide.
26//!
27//! [newtype guidance]: ../../../docs/style_guides/code_style_guide.md
28
29use std::cmp::Ordering;
30
31use derive_more::{Display, From, Into};
32/// The canonical cross-crate `PartitionIndex`, as carried in offset-syncs,
33/// checkpoints, and replicated records. The MM2 offset newtypes below stay
34/// crate-local. They encode source-cluster and target-cluster distinctions that
35/// are specific to MirrorMaker-2 and have no meaning outside this crate.
36pub use krabka_ids::{Offset, PartitionIndex};
37
38/// An offset on the **source** cluster, as recorded in an offset-sync or
39/// checkpoint (`upstream` in the JVM MM2 codecs).
40#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Display, From, Into)]
41pub struct UpstreamOffset(pub i64);
42
43/// An offset on the **target** cluster, paired with an [`UpstreamOffset`] in an
44/// offset-sync or checkpoint (`downstream` in the JVM MM2 codecs).
45#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Display, From, Into)]
46pub struct DownstreamOffset(pub i64);
47
48/// A consumer group's committed **source** offset that translation moves to the
49/// target. It lives in the same address space as [`UpstreamOffset`], because
50/// both are source-cluster offsets. It is a distinct concept at the translation
51/// call site, so it gets its own type.
52#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Display, From, Into)]
53pub struct CommittedOffset(pub i64);
54
55/// A record timestamp in epoch milliseconds.
56#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Display, From, Into)]
57pub struct Timestamp(pub i64);
58
59macro_rules! impl_primitive_cmp {
60    ($ty:ty, $inner:ty) => {
61        impl PartialEq<$inner> for $ty {
62            #[inline]
63            fn eq(&self, other: &$inner) -> bool {
64                self.0 == *other
65            }
66        }
67
68        impl PartialEq<$ty> for $inner {
69            #[inline]
70            fn eq(&self, other: &$ty) -> bool {
71                *self == other.0
72            }
73        }
74
75        impl PartialOrd<$inner> for $ty {
76            #[inline]
77            fn partial_cmp(&self, other: &$inner) -> Option<Ordering> {
78                self.0.partial_cmp(other)
79            }
80        }
81
82        impl PartialOrd<$ty> for $inner {
83            #[inline]
84            fn partial_cmp(&self, other: &$ty) -> Option<Ordering> {
85                self.partial_cmp(&other.0)
86            }
87        }
88    };
89}
90
91impl_primitive_cmp!(UpstreamOffset, i64);
92impl_primitive_cmp!(DownstreamOffset, i64);
93impl_primitive_cmp!(CommittedOffset, i64);
94impl_primitive_cmp!(Timestamp, i64);
95
96impl From<Offset> for UpstreamOffset {
97    #[inline]
98    fn from(value: Offset) -> Self {
99        Self(value.0)
100    }
101}
102
103impl From<CommittedOffset> for UpstreamOffset {
104    #[inline]
105    fn from(value: CommittedOffset) -> Self {
106        Self(value.0)
107    }
108}
krabka_replicator/ids.rs

krabka_replicator/
ids.rs
