crabka_replicator/
worker.rs

1//! Per-flow replication worker: drives one directional connect pipeline
2//! (`SourceConsumer` -> `TargetSink`) plus the heartbeat and checkpoint
3//! background tasks, with build-retry resilience and clean shutdown.
4
5use std::sync::Arc;
6use std::time::{Duration, SystemTime, UNIX_EPOCH};
7
8use crabka_connect::{ConnectorRuntime, RuntimeState};
9use tracing::warn;
10
11use crate::checkpoint_store::InternalTopicCheckpointStore;
12use crate::config::{NamingPolicy, PolicyConfig};
13use crate::record::ReplicatedRecord;
14use crate::selector::Selector;
15use crate::sink::{SinkParams, TargetSink};
16use crate::source::SourceConsumer;
17use crate::tasks::checkpoint::{CheckpointParams, CheckpointTask};
18use crate::tasks::heartbeat::{HeartbeatParams, HeartbeatTask};
19
20/// Maximum wall-clock time to keep retrying the pipeline build before giving up.
21const MAX_BUILD_ELAPSED: Duration = Duration::from_secs(30);
22/// Initial backoff between build attempts; doubles each retry up to [`MAX_BACKOFF`].
23const INITIAL_BACKOFF: Duration = Duration::from_millis(250);
24/// Cap on the per-retry backoff interval.
25const MAX_BACKOFF: Duration = Duration::from_secs(8);
26
27/// Parameters to start a [`FlowWorker`]. The supervisor resolves selectors into
28/// a concrete topic list and passes the per-flow cluster addresses, aliases,
29/// naming policy, residency policies, and security here.
30pub struct FlowWorkerParams {
31    /// Unique flow name (e.g. `"us-east__eu-west"`); seeds the consumer group id
32    /// and the checkpoint-store key.
33    pub flow_name: String,
34    /// Bootstrap address of the source cluster.
35    pub source_bootstrap: String,
36    /// Bootstrap address of the target cluster.
37    pub target_bootstrap: String,
38    /// Alias of the source cluster (stamped as provenance / used for MM2 names).
39    pub source_alias: String,
40    /// Alias of the target cluster (written into heartbeat records).
41    pub target_alias: String,
42    /// How source topics are renamed on the target.
43    pub naming: NamingPolicy,
44    /// Already-resolved source topic list (supervisor resolves selectors).
45    pub topics: Vec<String>,
46    /// Compliance zones of the target cluster (used for residency checks).
47    pub target_zones: Vec<String>,
48    /// Residency policies to enforce on the sink.
49    pub policies: Vec<PolicyConfig>,
50    /// Selector for which consumer groups the checkpoint task translates.
51    pub group_selector: Selector,
52    /// Optional TLS/SASL security for the source cluster.
53    pub security_source: Option<crabka_client_core::security::ClientSecurity>,
54    /// Optional TLS/SASL security for the target cluster.
55    pub security_target: Option<crabka_client_core::security::ClientSecurity>,
56}
57
58/// One directional replication flow: the running connect runtime plus the
59/// heartbeat and checkpoint background tasks.
60pub struct FlowWorker {
61    runtime: crabka_connect::ConnectorHandle,
62    heartbeat: HeartbeatTask,
63    checkpoint: CheckpointTask,
64}
65
66/// Current wall-clock time in milliseconds since the Unix epoch.
67fn now_ms() -> i64 {
68    i64::try_from(
69        SystemTime::now()
70            .duration_since(UNIX_EPOCH)
71            .unwrap_or_default()
72            .as_millis(),
73    )
74    .unwrap_or(i64::MAX)
75}
76
77/// Compute the next build-retry backoff: double the current interval, capped at
78/// [`MAX_BACKOFF`].
79fn next_backoff(current: Duration) -> Duration {
80    (current * 2).min(MAX_BACKOFF)
81}
82
83impl FlowWorker {
84    /// Build and start the pipeline, retrying transient build failures with
85    /// bounded exponential backoff (a cluster may be briefly unreachable).
86    ///
87    /// Backoff starts at ~250ms and doubles (capped at ~8s) until the cumulative
88    /// elapsed time exceeds ~30s, after which the last build error is returned.
89    ///
90    /// # Errors
91    ///
92    /// Returns the last build error if the pipeline cannot be constructed within
93    /// the retry budget.
94    // cargo-mutants: retry/backoff loop has no value-asserting unit coverage
95    #[cfg_attr(test, mutants::skip)]
96    pub async fn start(p: FlowWorkerParams) -> crate::Result<Self> {
97        let mut backoff = INITIAL_BACKOFF;
98        let mut elapsed = Duration::ZERO;
99
100        loop {
101            match Self::build(&p).await {
102                Ok(worker) => return Ok(worker),
103                Err(e) => {
104                    if elapsed >= MAX_BUILD_ELAPSED {
105                        return Err(e);
106                    }
107                    warn!(
108                        flow = %p.flow_name,
109                        error = %e,
110                        backoff_ms = backoff.as_millis(),
111                        "flow worker build failed; retrying after backoff"
112                    );
113                    tokio::time::sleep(backoff).await;
114                    elapsed += backoff;
115                    backoff = next_backoff(backoff);
116                }
117            }
118        }
119    }
120
121    /// A single build attempt: stand up the source, sink, checkpoint store,
122    /// connect runtime, and both background tasks.
123    async fn build(p: &FlowWorkerParams) -> crate::Result<Self> {
124        let group_id = format!("crabka-replicator-{}", p.flow_name);
125
126        let source = SourceConsumer::start(
127            &p.source_bootstrap,
128            &group_id,
129            &p.topics,
130            p.security_source.clone(),
131        )
132        .await?;
133
134        let sink = TargetSink::start(SinkParams {
135            target_bootstrap: p.target_bootstrap.clone(),
136            source_alias: p.source_alias.clone(),
137            naming: p.naming,
138            target_zones: p.target_zones.clone(),
139            policies: p.policies.clone(),
140            security: p.security_target.clone(),
141        })
142        .await?;
143
144        let store = InternalTopicCheckpointStore::start(
145            &p.target_bootstrap,
146            &p.flow_name,
147            p.security_target.clone(),
148        )
149        .await?;
150
151        let runtime = ConnectorRuntime::<(), ReplicatedRecord>::new()
152            .add_source(source)
153            .add_sink(sink)
154            .checkpoint_store(Arc::new(store))
155            .commit_interval(Duration::from_millis(500))
156            .max_batch(500)
157            .run()?;
158
159        let heartbeat = HeartbeatTask::start(HeartbeatParams {
160            target_bootstrap: p.target_bootstrap.clone(),
161            source_alias: p.source_alias.clone(),
162            target_alias: p.target_alias.clone(),
163            interval: Duration::from_secs(1),
164            now_ms,
165            security: p.security_target.clone(),
166        })
167        .await?;
168
169        let checkpoint = CheckpointTask::start(
170            CheckpointParams {
171                source_bootstrap: p.source_bootstrap.clone(),
172                target_bootstrap: p.target_bootstrap.clone(),
173                source_alias: p.source_alias.clone(),
174                naming: p.naming,
175                group_selector: p.group_selector.clone(),
176                security: p.security_target.clone(),
177            },
178            Duration::from_secs(5),
179        )
180        .await?;
181
182        Ok(Self {
183            runtime,
184            heartbeat,
185            checkpoint,
186        })
187    }
188
189    /// Current connect-runtime state, for the supervisor to detect `Failed`
190    /// workers and restart them.
191    #[must_use]
192    pub fn state(&self) -> RuntimeState {
193        self.runtime.state()
194    }
195
196    /// Graceful stop: drain and shut down the connect runtime, then stop both
197    /// background tasks.
198    // cargo-mutants: shutdown ordering not asserted by unit tests
199    #[cfg_attr(test, mutants::skip)]
200    pub async fn shutdown(self) {
201        let _ = self.runtime.shutdown().await;
202        self.heartbeat.shutdown().await;
203        self.checkpoint.shutdown().await;
204    }
205}
206
207#[cfg(test)]
208mod tests {
209    use assert2::assert;
210
211    use super::*;
212
213    #[test]
214    fn now_ms_is_a_recent_epoch_millis() {
215        // Epoch millis for 2023-11-14T22:13:20Z; any sane current time exceeds
216        // it. The `-> -1 / 0 / 1` mutants all fall below this floor.
217        assert!(super::now_ms() > 1_700_000_000_000);
218    }
219
220    #[test]
221    fn next_backoff_doubles_and_caps() {
222        // Doubling: 250ms -> 500ms (the `*2`→`/2` mutant gives 125ms).
223        assert!(super::next_backoff(Duration::from_millis(250)) == Duration::from_millis(500));
224        // Cap: never exceeds MAX_BACKOFF even when already at the cap.
225        assert!(super::next_backoff(MAX_BACKOFF) == MAX_BACKOFF);
226    }
227
228    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
229    async fn worker_replicates_one_flow() {
230        let s_dir = tempfile::TempDir::new().unwrap();
231        let t_dir = tempfile::TempDir::new().unwrap();
232        let source = crabka_broker::Broker::start(crabka_broker::BrokerConfig::for_tests(
233            s_dir.path().to_path_buf(),
234        ))
235        .await
236        .unwrap();
237        let target = crabka_broker::Broker::start(crabka_broker::BrokerConfig::for_tests(
238            t_dir.path().to_path_buf(),
239        ))
240        .await
241        .unwrap();
242        let sb = source.listen_addr().to_string();
243        let tb = target.listen_addr().to_string();
244
245        crate::test_util::create_topic(&sb, "orders", 1).await;
246        crate::test_util::produce(&sb, "orders", b"k", b"v").await;
247
248        let worker = FlowWorker::start(FlowWorkerParams {
249            flow_name: "us-east__eu-west".into(),
250            source_bootstrap: sb,
251            target_bootstrap: tb.clone(),
252            source_alias: "us-east".into(),
253            target_alias: "eu-west".into(),
254            naming: crate::config::NamingPolicy::Default,
255            topics: vec!["orders".to_string()],
256            target_zones: vec!["us".into()],
257            policies: vec![],
258            group_selector: crate::selector::Selector::compile(&[], &[]).unwrap(),
259            security_source: None,
260            security_target: None,
261        })
262        .await
263        .unwrap();
264
265        crate::test_util::await_topic_count(
266            &tb,
267            "us-east.orders",
268            1,
269            std::time::Duration::from_secs(15),
270        )
271        .await;
272
273        worker.shutdown().await;
274        assert!(crate::test_util::topic_record_count(&tb, "us-east.orders").await >= 1);
275    }
276}
crabka_replicator/worker.rs

crabka_replicator/
worker.rs
