krabka_replicator/
naming.rs

1//! Remote-topic naming and loop-prevention primitives.
2
3use crate::config::NamingPolicy;
4
5/// Header key that every produced record carries with the origin cluster alias.
6/// The identity-naming loop-guard uses it, and it is also useful for provenance.
7pub const PROVENANCE_HEADER: &str = "__krabka_origin";
8
9/// Maps source topic names to target topic names and enforces loop-prevention
10/// rules for active/active replication.
11#[derive(Debug, Clone)]
12pub struct Renamer {
13    policy: NamingPolicy,
14    source_alias: String,
15}
16
17impl Renamer {
18    /// Create a new [`Renamer`] for the given policy and source cluster alias.
19    #[must_use]
20    pub fn new(policy: NamingPolicy, source_alias: &str) -> Self {
21        Self {
22            policy,
23            source_alias: source_alias.to_owned(),
24        }
25    }
26
27    /// Return the name this topic should have on the target cluster.
28    ///
29    /// - [`NamingPolicy::Default`][]: `<source_alias>.<source_topic>`
30    /// - [`NamingPolicy::Identity`][]: `<source_topic>` (unchanged)
31    #[must_use]
32    pub fn target_name(&self, source_topic: &str) -> String {
33        match self.policy {
34            NamingPolicy::Default => format!("{}.{}", self.source_alias, source_topic),
35            NamingPolicy::Identity => source_topic.to_owned(),
36        }
37    }
38
39    /// Return `true` if this topic looks like a topic that replication already
40    /// produced. This flow should exclude it from its subscription, for loop
41    /// prevention.
42    ///
43    /// Under [`NamingPolicy::Default`] a topic that contains `.`, but does not
44    /// *start* with `.`, counts as remote. This mirrors the "topic contains the
45    /// replication separator" heuristic of `MirrorMaker` 2. The rule is
46    /// deliberately simple for Slice 1, and the two-cluster integration test
47    /// pins the exact behaviour.
48    ///
49    /// Under [`NamingPolicy::Identity`] the loop guard uses the header. See
50    /// [`PROVENANCE_HEADER`]. This method then always returns `false`.
51    #[must_use]
52    pub fn is_remote(&self, topic: &str) -> bool {
53        match self.policy {
54            NamingPolicy::Default => topic.contains('.') && !topic.starts_with('.'),
55            NamingPolicy::Identity => false,
56        }
57    }
58
59    /// The source cluster alias stamped in [`PROVENANCE_HEADER`] on produced
60    /// records, used for identity-policy loop prevention.
61    #[must_use]
62    pub fn provenance_alias(&self) -> &str {
63        &self.source_alias
64    }
65
66    /// The naming policy in effect.
67    #[must_use]
68    pub fn policy(&self) -> NamingPolicy {
69        self.policy
70    }
71}
72
73#[cfg(test)]
74mod tests {
75    use assert2::check;
76
77    use super::*;
78    use crate::config::NamingPolicy;
79
80    #[test]
81    fn default_policy_prefixes_with_source_alias() {
82        let n = Renamer::new(NamingPolicy::Default, "us-east");
83        assert2::assert!(n.target_name("orders") == "us-east.orders");
84        check!(n.is_remote("eu-west.billing"));
85        check!(!n.is_remote("orders"));
86    }
87
88    #[test]
89    fn identity_policy_keeps_name_and_uses_provenance() {
90        let n = Renamer::new(NamingPolicy::Identity, "us-east");
91        assert2::assert!(n.target_name("orders") == "orders");
92        check!(!n.is_remote("orders"));
93        assert2::assert!(n.provenance_alias() == "us-east");
94        // `policy()` returns the configured policy verbatim, not the Default.
95        assert2::assert!(n.policy() == NamingPolicy::Identity);
96    }
97}
krabka_replicator/naming.rs

krabka_replicator/
naming.rs
