mocra_core/queue/
compensation.rs

1use crate::common::model::{
2    Request, Response,
3    message::{TaskErrorEvent, TaskEvent, TaskParserEvent},
4};
5use crate::errors::Result;
6use crate::utils::logger::LogModel;
7use async_trait::async_trait;
8use std::sync::Arc;
9
10/// Trait for objects that can be uniquely identified for compensation purposes.
11pub trait Identifiable {
12    fn get_id(&self) -> String;
13
14    /// MQ partition key: decides which partition / stream shard / consumer a message is routed
15    /// to, used for **account affinity** (session stickiness + tasks for the same account landing
16    /// on the same node within the cluster).
17    ///
18    /// Defaults to the same value as [`get_id`](Self::get_id) (no affinity, random distribution);
19    /// `TaskEvent` overrides it with the **account**, so that tasks for one account land on the
20    /// same partition consistently via `hash(account)`, reusing Kafka/NATS consumer group
21    /// assignment to achieve cross-node consumer affinity. It is independent of `get_id`, which
22    /// is used for deduplication / compensation.
23    fn partition_key(&self) -> String {
24        self.get_id()
25    }
26}
27
28impl Identifiable for LogModel {
29    fn get_id(&self) -> String {
30        self.request_id
31            .map(|id| id.to_string())
32            .unwrap_or_else(|| self.task_id.clone())
33    }
34}
35
36impl Identifiable for Request {
37    fn get_id(&self) -> String {
38        self.id.to_string()
39    }
40}
41
42impl Identifiable for Response {
43    fn get_id(&self) -> String {
44        self.id.to_string()
45    }
46}
47
48impl Identifiable for TaskParserEvent {
49    fn get_id(&self) -> String {
50        self.id.to_string()
51    }
52}
53
54impl Identifiable for TaskErrorEvent {
55    fn get_id(&self) -> String {
56        self.id.to_string()
57    }
58}
59
60impl Identifiable for TaskEvent {
61    fn get_id(&self) -> String {
62        self.run_id.to_string()
63    }
64
65    /// Account affinity: tasks for one account land on the same partition (session stickiness +
66    /// same account on the same node across the cluster).
67    fn partition_key(&self) -> String {
68        self.account.clone()
69    }
70}
71
72#[async_trait]
73pub trait Compensator: Send + Sync {
74    /// Add a task to the compensation queue.
75    async fn add_task(&self, topic: &str, id: &str, payload: Arc<Vec<u8>>) -> Result<()>;
76    /// Remove a task from the compensation queue.
77    async fn remove_task(&self, topic: &str, id: &str) -> Result<()>;
78}
mocra_core/queue/compensation.rs

mocra_core/queue/
compensation.rs
