1use std::collections::BTreeMap;
2use std::sync::Arc;
3use std::sync::RwLock;
4use std::time::Duration;
5use std::time::Instant;
6
7use async_trait::async_trait;
8use reqwest::StatusCode;
9use serde::Serialize;
10
11use crate::domain::errors::{Result, StasisError};
12use crate::domain::runtime::cluster_node::{ClusterForwardCommand, ClusterForwardOutcome};
13use crate::ports::outbound::runtime::cluster_command_forwarder::ClusterCommandForwarder;
14use crate::ports::outbound::runtime::cluster_forward_outcome_store::ClusterForwardOutcomeStore;
15use crate::ports::outbound::runtime::runtime_metrics::RuntimeMetrics;
16
17pub const CLUSTER_FORWARD_ATTEMPTS_TOTAL: &str = "cluster_forward_attempts_total";
18pub const CLUSTER_FORWARD_RETRIES_TOTAL: &str = "cluster_forward_retries_total";
19pub const CLUSTER_FORWARD_SUCCESSES_TOTAL: &str = "cluster_forward_successes_total";
20pub const CLUSTER_FORWARD_FAILURES_TOTAL: &str = "cluster_forward_failures_total";
21pub const CLUSTER_FORWARD_NO_ROUTE_TOTAL: &str = "cluster_forward_no_route_total";
22pub const CLUSTER_FORWARD_REJECTED_TOTAL: &str = "cluster_forward_rejected_total";
23pub const CLUSTER_FORWARD_DURATION_MS: &str = "cluster_forward_duration_ms";
24pub const CLUSTER_FORWARD_IDEMPOTENT_HITS_TOTAL: &str = "cluster_forward_idempotent_hits_total";
25
26#[derive(Clone)]
27struct DedupeEntry {
28 accepted: bool,
29 observed_at: Instant,
30}
31
32#[derive(Clone)]
33pub struct HttpClusterCommandForwarder {
34 client: reqwest::Client,
35 region_targets: BTreeMap<String, String>,
36 authorization_bearer: Option<String>,
37 metrics: Option<Arc<dyn RuntimeMetrics>>,
38 outcome_store: Option<Arc<dyn ClusterForwardOutcomeStore>>,
39 max_attempts: u32,
40 base_backoff_ms: u64,
41 max_backoff_ms: u64,
42 idempotency_ttl: Duration,
43 dedupe_cache: Arc<RwLock<BTreeMap<String, DedupeEntry>>>,
44}
45
46impl HttpClusterCommandForwarder {
47 pub fn new(region_targets: BTreeMap<String, String>) -> Self {
48 Self {
49 client: reqwest::Client::new(),
50 region_targets,
51 authorization_bearer: None,
52 metrics: None,
53 outcome_store: None,
54 max_attempts: 3,
55 base_backoff_ms: 100,
56 max_backoff_ms: 2_000,
57 idempotency_ttl: Duration::from_secs(300),
58 dedupe_cache: Arc::new(RwLock::new(BTreeMap::new())),
59 }
60 }
61
62 pub fn with_region_target(
63 mut self,
64 region: impl Into<String>,
65 endpoint_url: impl Into<String>,
66 ) -> Self {
67 self.region_targets
68 .insert(region.into(), endpoint_url.into());
69 self
70 }
71
72 pub fn with_bearer_token(mut self, token: impl Into<String>) -> Self {
73 self.authorization_bearer = Some(token.into());
74 self
75 }
76
77 pub fn with_metrics(mut self, metrics: Arc<dyn RuntimeMetrics>) -> Self {
78 self.metrics = Some(metrics);
79 self
80 }
81
82 pub fn with_outcome_store(mut self, store: Arc<dyn ClusterForwardOutcomeStore>) -> Self {
83 self.outcome_store = Some(store);
84 self
85 }
86
87 pub fn with_retry_policy(
88 mut self,
89 max_attempts: u32,
90 base_backoff_ms: u64,
91 max_backoff_ms: u64,
92 ) -> Self {
93 self.max_attempts = max_attempts.max(1);
94 self.base_backoff_ms = base_backoff_ms.max(1);
95 self.max_backoff_ms = max_backoff_ms.max(self.base_backoff_ms);
96 self
97 }
98
99 pub fn with_idempotency_ttl(mut self, ttl: Duration) -> Self {
100 self.idempotency_ttl = ttl;
101 self
102 }
103
104 fn endpoint_for_region(&self, region: &str) -> Option<&str> {
105 self.region_targets.get(region).map(String::as_str)
106 }
107
108 fn compute_backoff_millis(&self, attempt: u32) -> u64 {
109 let factor = 2u64.saturating_pow(attempt.saturating_sub(1));
110 let candidate = self.base_backoff_ms.saturating_mul(factor);
111 candidate.min(self.max_backoff_ms)
112 }
113
114 fn should_retry_status(status: StatusCode) -> bool {
115 status.is_server_error() || status == StatusCode::TOO_MANY_REQUESTS
116 }
117
118 fn dedupe_key(command: &ClusterForwardCommand) -> Option<String> {
119 command.correlation_id.as_ref().and_then(|correlation_id| {
120 let correlation_id = correlation_id.trim();
121 if correlation_id.is_empty() {
122 return None;
123 }
124
125 Some(format!(
126 "{}|{}|{}",
127 command.target_region, command.command_name, correlation_id
128 ))
129 })
130 }
131
132 fn load_dedupe_hit(&self, command: &ClusterForwardCommand) -> Option<bool> {
133 let key = Self::dedupe_key(command)?;
134 let now = Instant::now();
135
136 let mut cache = self.dedupe_cache.write().ok()?;
137 let entry = cache.get(&key).cloned()?;
138
139 if now.duration_since(entry.observed_at) > self.idempotency_ttl {
140 cache.remove(&key);
141 return None;
142 }
143
144 Some(entry.accepted)
145 }
146
147 fn remember_dedupe_result(&self, command: &ClusterForwardCommand, accepted: bool) {
148 let Some(key) = Self::dedupe_key(command) else {
149 return;
150 };
151
152 let Ok(mut cache) = self.dedupe_cache.write() else {
153 return;
154 };
155
156 cache.insert(
157 key,
158 DedupeEntry {
159 accepted,
160 observed_at: Instant::now(),
161 },
162 );
163 }
164
165 async fn record_outcome(
166 &self,
167 command: &ClusterForwardCommand,
168 accepted: bool,
169 attempts: u32,
170 error: Option<String>,
171 ) {
172 let Some(store) = &self.outcome_store else {
173 return;
174 };
175
176 let _ = store
177 .record(ClusterForwardOutcome {
178 target_region: command.target_region.clone(),
179 command_name: command.command_name.clone(),
180 correlation_id: command.correlation_id.clone(),
181 accepted,
182 attempts,
183 error,
184 completed_at: chrono::Utc::now(),
185 })
186 .await;
187 }
188}
189
190#[derive(Debug, Serialize)]
191struct ForwardCommandPayload<'a> {
192 target_region: &'a str,
193 command_name: &'a str,
194 payload: &'a str,
195 correlation_id: Option<&'a str>,
196 issued_at: String,
197}
198
199#[async_trait]
200impl ClusterCommandForwarder for HttpClusterCommandForwarder {
201 async fn forward(&self, command: ClusterForwardCommand) -> Result<bool> {
202 let start = Instant::now();
203 if let Some(accepted) = self.load_dedupe_hit(&command) {
204 if let Some(metrics) = &self.metrics {
205 metrics.incr_counter(CLUSTER_FORWARD_IDEMPOTENT_HITS_TOTAL, 1);
206 metrics.observe_duration_ms(CLUSTER_FORWARD_DURATION_MS, 0);
207 }
208 return Ok(accepted);
209 }
210
211 let Some(endpoint_url) = self.endpoint_for_region(&command.target_region) else {
212 if let Some(metrics) = &self.metrics {
213 metrics.incr_counter(CLUSTER_FORWARD_NO_ROUTE_TOTAL, 1);
214 metrics.incr_counter(CLUSTER_FORWARD_FAILURES_TOTAL, 1);
215 }
216 let err_msg = format!(
217 "no cluster forward endpoint configured for region={}",
218 command.target_region
219 );
220 self.record_outcome(&command, false, 0, Some(err_msg.clone()))
221 .await;
222 return Err(StasisError::PortFailure(err_msg));
223 };
224
225 let request_body = ForwardCommandPayload {
226 target_region: &command.target_region,
227 command_name: &command.command_name,
228 payload: &command.payload,
229 correlation_id: command.correlation_id.as_deref(),
230 issued_at: command.issued_at.to_rfc3339(),
231 };
232
233 for attempt in 1..=self.max_attempts {
234 if let Some(metrics) = &self.metrics {
235 metrics.incr_counter(CLUSTER_FORWARD_ATTEMPTS_TOTAL, 1);
236 }
237
238 let mut request = self.client.post(endpoint_url).json(&request_body);
239 if let Some(token) = &self.authorization_bearer {
240 request = request.bearer_auth(token);
241 }
242
243 let send_result = request.send().await;
244 match send_result {
245 Ok(response) if response.status().is_success() => {
246 if let Some(metrics) = &self.metrics {
247 metrics.incr_counter(CLUSTER_FORWARD_SUCCESSES_TOTAL, 1);
248 metrics.observe_duration_ms(
249 CLUSTER_FORWARD_DURATION_MS,
250 start.elapsed().as_millis() as u64,
251 );
252 }
253 self.remember_dedupe_result(&command, true);
254 self.record_outcome(&command, true, attempt, None).await;
255 return Ok(true);
256 }
257 Ok(response) if Self::should_retry_status(response.status()) => {
258 if attempt == self.max_attempts {
259 if let Some(metrics) = &self.metrics {
260 metrics.incr_counter(CLUSTER_FORWARD_FAILURES_TOTAL, 1);
261 metrics.observe_duration_ms(
262 CLUSTER_FORWARD_DURATION_MS,
263 start.elapsed().as_millis() as u64,
264 );
265 }
266 let err_msg = format!(
267 "cluster forward failed after retries with status={} region={} command={}",
268 response.status(),
269 command.target_region,
270 command.command_name
271 );
272 self.record_outcome(&command, false, attempt, Some(err_msg.clone()))
273 .await;
274 return Err(StasisError::PortFailure(err_msg));
275 }
276
277 if let Some(metrics) = &self.metrics {
278 metrics.incr_counter(CLUSTER_FORWARD_RETRIES_TOTAL, 1);
279 }
280 }
281 Ok(response) => {
282 if let Some(metrics) = &self.metrics {
283 metrics.incr_counter(CLUSTER_FORWARD_REJECTED_TOTAL, 1);
284 metrics.incr_counter(CLUSTER_FORWARD_FAILURES_TOTAL, 1);
285 metrics.observe_duration_ms(
286 CLUSTER_FORWARD_DURATION_MS,
287 start.elapsed().as_millis() as u64,
288 );
289 }
290 let err_msg = format!(
291 "cluster forward rejected with status={} region={} command={}",
292 response.status(),
293 command.target_region,
294 command.command_name
295 );
296 self.record_outcome(&command, false, attempt, Some(err_msg.clone()))
297 .await;
298 return Err(StasisError::PortFailure(err_msg));
299 }
300 Err(err) => {
301 if attempt == self.max_attempts {
302 if let Some(metrics) = &self.metrics {
303 metrics.incr_counter(CLUSTER_FORWARD_FAILURES_TOTAL, 1);
304 metrics.observe_duration_ms(
305 CLUSTER_FORWARD_DURATION_MS,
306 start.elapsed().as_millis() as u64,
307 );
308 }
309 let err_msg = format!(
310 "cluster forward request failed region={} command={} error={err}",
311 command.target_region, command.command_name
312 );
313 self.record_outcome(&command, false, attempt, Some(err_msg.clone()))
314 .await;
315 return Err(StasisError::PortFailure(err_msg));
316 }
317
318 if let Some(metrics) = &self.metrics {
319 metrics.incr_counter(CLUSTER_FORWARD_RETRIES_TOTAL, 1);
320 }
321 }
322 }
323
324 let delay_ms = self.compute_backoff_millis(attempt);
325 tokio::time::sleep(Duration::from_millis(delay_ms)).await;
326 }
327
328 Err(StasisError::PortFailure(
329 "cluster forward exhausted retries unexpectedly".to_string(),
330 ))
331 }
332}
333
334#[cfg(test)]
335mod tests {
336 use std::collections::BTreeMap;
337 use std::sync::Arc;
338
339 use tokio::io::{AsyncReadExt, AsyncWriteExt};
340 use tokio::net::TcpListener;
341
342 use chrono::Utc;
343
344 use crate::domain::runtime::cluster_node::ClusterForwardCommand;
345 use crate::infrastructure::runtime::in_memory_runtime_metrics::InMemoryRuntimeMetrics;
346 use crate::ports::outbound::runtime::cluster_command_forwarder::ClusterCommandForwarder;
347
348 use super::HttpClusterCommandForwarder;
349 use super::{
350 CLUSTER_FORWARD_ATTEMPTS_TOTAL, CLUSTER_FORWARD_FAILURES_TOTAL,
351 CLUSTER_FORWARD_IDEMPOTENT_HITS_TOTAL, CLUSTER_FORWARD_NO_ROUTE_TOTAL,
352 CLUSTER_FORWARD_RETRIES_TOTAL, CLUSTER_FORWARD_SUCCESSES_TOTAL,
353 };
354
355 async fn start_sequence_server(status_codes: Vec<u16>) -> String {
356 let listener = TcpListener::bind("127.0.0.1:0")
357 .await
358 .expect("listener should bind");
359 let addr = listener
360 .local_addr()
361 .expect("listener address should be available");
362
363 tokio::spawn(async move {
364 for status_code in status_codes {
365 let Ok((mut socket, _)) = listener.accept().await else {
366 break;
367 };
368
369 let mut buf = [0u8; 2048];
370 let _ = socket.read(&mut buf).await;
371
372 let reason = match status_code {
373 200 => "OK",
374 429 => "Too Many Requests",
375 500 => "Internal Server Error",
376 _ => "Status",
377 };
378
379 let response = format!(
380 "HTTP/1.1 {} {}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
381 status_code, reason
382 );
383 let _ = socket.write_all(response.as_bytes()).await;
384 let _ = socket.shutdown().await;
385 }
386 });
387
388 format!("http://{}", addr)
389 }
390
391 fn sample_command() -> ClusterForwardCommand {
392 ClusterForwardCommand {
393 target_region: "eu-west".to_string(),
394 command_name: "scheduler.pause_queue".to_string(),
395 payload: "{\"queue\":\"default\"}".to_string(),
396 correlation_id: Some("cmd-1".to_string()),
397 issued_at: Utc::now(),
398 }
399 }
400
401 #[test]
402 fn backoff_scales_exponentially_with_cap() {
403 let forwarder =
404 HttpClusterCommandForwarder::new(BTreeMap::new()).with_retry_policy(5, 100, 250);
405
406 assert_eq!(forwarder.compute_backoff_millis(1), 100);
407 assert_eq!(forwarder.compute_backoff_millis(2), 200);
408 assert_eq!(forwarder.compute_backoff_millis(3), 250);
409 assert_eq!(forwarder.compute_backoff_millis(4), 250);
410 }
411
412 #[test]
413 fn retry_policy_normalizes_invalid_inputs() {
414 let forwarder =
415 HttpClusterCommandForwarder::new(BTreeMap::new()).with_retry_policy(0, 0, 0);
416
417 assert_eq!(forwarder.max_attempts, 1);
418 assert_eq!(forwarder.base_backoff_ms, 1);
419 assert_eq!(forwarder.max_backoff_ms, 1);
420 }
421
422 #[tokio::test]
423 async fn returns_error_when_region_target_is_not_configured() {
424 let forwarder = HttpClusterCommandForwarder::new(BTreeMap::new());
425
426 let err = forwarder
427 .forward(sample_command())
428 .await
429 .expect_err("expected configuration error");
430
431 assert!(
432 err.to_string()
433 .contains("no cluster forward endpoint configured for region=eu-west")
434 );
435 }
436
437 #[tokio::test]
438 async fn retries_on_retryable_status_then_succeeds() {
439 let endpoint_url = start_sequence_server(vec![500, 200]).await;
440 let metrics = Arc::new(InMemoryRuntimeMetrics::default());
441 let forwarder = HttpClusterCommandForwarder::new(BTreeMap::new())
442 .with_region_target("eu-west", endpoint_url)
443 .with_retry_policy(3, 1, 2)
444 .with_metrics(metrics.clone());
445
446 let accepted = forwarder
447 .forward(sample_command())
448 .await
449 .expect("forward should succeed after retry");
450 assert!(accepted);
451
452 let snapshot = metrics.snapshot();
453 assert_eq!(
454 snapshot.counters.get(CLUSTER_FORWARD_ATTEMPTS_TOTAL),
455 Some(&2)
456 );
457 assert_eq!(
458 snapshot.counters.get(CLUSTER_FORWARD_RETRIES_TOTAL),
459 Some(&1)
460 );
461 assert_eq!(
462 snapshot.counters.get(CLUSTER_FORWARD_SUCCESSES_TOTAL),
463 Some(&1)
464 );
465 }
466
467 #[tokio::test]
468 async fn records_no_route_and_failure_metrics_when_region_is_missing() {
469 let metrics = Arc::new(InMemoryRuntimeMetrics::default());
470 let forwarder =
471 HttpClusterCommandForwarder::new(BTreeMap::new()).with_metrics(metrics.clone());
472
473 let _ = forwarder
474 .forward(sample_command())
475 .await
476 .expect_err("expected no-route error");
477
478 let snapshot = metrics.snapshot();
479 assert_eq!(
480 snapshot.counters.get(CLUSTER_FORWARD_NO_ROUTE_TOTAL),
481 Some(&1)
482 );
483 assert_eq!(
484 snapshot.counters.get(CLUSTER_FORWARD_FAILURES_TOTAL),
485 Some(&1)
486 );
487 }
488
489 #[tokio::test]
490 async fn deduplicates_repeated_correlation_id_within_ttl() {
491 let endpoint_url = start_sequence_server(vec![200]).await;
492 let metrics = Arc::new(InMemoryRuntimeMetrics::default());
493 let forwarder = HttpClusterCommandForwarder::new(BTreeMap::new())
494 .with_region_target("eu-west", endpoint_url)
495 .with_metrics(metrics.clone());
496
497 let first = forwarder
498 .forward(sample_command())
499 .await
500 .expect("first forward should succeed");
501 assert!(first);
502
503 let second = forwarder
504 .forward(sample_command())
505 .await
506 .expect("second forward should use dedupe cache");
507 assert!(second);
508
509 let snapshot = metrics.snapshot();
510 assert_eq!(
511 snapshot.counters.get(CLUSTER_FORWARD_ATTEMPTS_TOTAL),
512 Some(&1)
513 );
514 assert_eq!(
515 snapshot.counters.get(CLUSTER_FORWARD_IDEMPOTENT_HITS_TOTAL),
516 Some(&1)
517 );
518 }
519}