refactor(queue): refresh crate across consumer, producer and metrics

Modified lib/queue/consumer.rs +9 −1 View lib/queue/consumer.rs

@@ -14,6 +14,7 @@

1414

1515use crate::config::AppQueueSettings;

1616use crate::error::QueueError;

17use crate::metrics;

1718

1819pub type RawHandler =

1920 Arc<dyn Fn(RawMessage) -> BoxFuture<'static, Result<(), QueueError>> + Send + Sync>;

@@ -181,12 +182,17 @@

181182 key: msg.key().map(|k| k.to_vec()),

182183 payload: msg.payload().map(|p| p.to_vec()).unwrap_or_default(),

183184 };

185 metrics::record_message_received();

184186 let mut attempts = 0usize;

185187 loop {

186188 attempts += 1;

187189 match handler(raw.clone()).await {

188 Ok(()) => break,

190 Ok(()) => {

191 metrics::record_message_processed(true);

192 break;

193 }

189194 Err(e) => {

195 metrics::record_retry();

190196 let delay = backoff_delay(attempts);

191197 warn!(

192198 group = %group_id,

@@ -196,6 +202,8 @@

196202 "handler failed"

197203 );

198204 if attempts > max_retries {

205 metrics::record_message_dropped();

206 metrics::record_message_processed(false);

199207 error!(

200208 group = %group_id,

201209 topic = %raw.topic,

Modified lib/queue/lib.rs +1 −0 View lib/queue/lib.rs

@@ -1,6 +1,7 @@

11pub mod config;

22pub mod consumer;

33pub mod error;

4pub mod metrics;

45pub mod producer;

56pub mod topic;

67pub mod typed;

Added lib/queue/metrics.rs +76 −0 View lib/queue/metrics.rs

@@ -0,0 +1,76 @@

1use std::sync::Once;

2

3use metrics::{counter, describe_counter};

4

5use crate::error::QueueError;

6

7pub(crate) const OUTCOME_SUCCESS: &str = "success";

8

9static DESCRIBE_ONCE: Once = Once::new();

10

11fn ensure_described() {

12 DESCRIBE_ONCE.call_once(describe_all);

13}

14

15fn describe_all() {

16 describe_counter!("queue_publish_total", "Messages published to Kafka partitioned by outcome.");

17 describe_counter!(

18 "queue_message_received_total",

19 "Messages received from Kafka by subscribers."

20 );

21 describe_counter!(

22 "queue_message_processed_total",

23 "Messages handled by subscriber handlers partitioned by outcome."

24 );

25 describe_counter!(

26 "queue_retries_total",

27 "Retries of failed handler executions."

28 );

29 describe_counter!(

30 "queue_message_dropped_total",

31 "Messages dropped after exhausting handler retries."

32 );

33}

34

35fn outcome_for<T>(result: &Result<T, QueueError>) -> &'static str {

36 match result {

37 Ok(_) => OUTCOME_SUCCESS,

38 Err(QueueError::Kafka(_)) => "kafka",

39 Err(QueueError::Config(_)) => "config",

40 Err(QueueError::Topic(_)) => "topic",

41 Err(QueueError::Serde(_)) => "serde",

42 Err(QueueError::Handler(_)) => "handler",

43 Err(QueueError::Join(_)) => "join",

44 Err(QueueError::Disabled) => "disabled",

45 Err(QueueError::Cancelled) => "cancelled",

46 }

47}

48

49pub(crate) fn record_publish<T>(result: &Result<T, QueueError>) {

50 ensure_described();

51 counter!("queue_publish_total", "outcome" => outcome_for(result)).increment(1);

52}

53

54pub(crate) fn record_message_received() {

55 ensure_described();

56 counter!("queue_message_received_total").increment(1);

57}

58

59pub(crate) fn record_message_processed(ok: bool) {

60 ensure_described();

61 counter!(

62 "queue_message_processed_total",

63 "outcome" => if ok { OUTCOME_SUCCESS } else { "error" }

64 )

65 .increment(1);

66}

67

68pub(crate) fn record_retry() {

69 ensure_described();

70 counter!("queue_retries_total").increment(1);

71}

72

73pub(crate) fn record_message_dropped() {

74 ensure_described();

75 counter!("queue_message_dropped_total").increment(1);

76}

No newline at end of file

Modified lib/queue/producer.rs +26 −20 View lib/queue/producer.rs

@@ -10,6 +10,7 @@

1010

1111use crate::config::AppQueueSettings;

1212use crate::error::QueueError;

13use crate::metrics;

1314

1415#[derive(Clone)]

1516pub struct KafkaPublisher {

@@ -47,28 +48,33 @@

4748 key: Option<&[u8]>,

4849 payload: &[u8],

4950 ) -> Result<(), QueueError> {

50 let timeout = Duration::from_millis(self.cfg.publish_queue_timeout_ms as u64);

51 let message_id = Uuid::new_v4().to_string();

52 let headers = OwnedHeaders::new()

53 .insert(Header {

54 key: "message-id",

55 value: Some(&message_id),

56 })

57 .insert(Header {

58 key: "content-type",

59 value: Some("application/octet-stream"),

60 });

61 let mut record = FutureRecord::to(topic).payload(payload).headers(headers);

62 if let Some(k) = key {

63 record = record.key(k);

64 }

65 match self.inner.send(record, timeout).await {

66 Ok(_) => Ok(()),

67 Err((e, _msg)) => {

68 warn!(topic, error = %e, "kafka publish failed");

69 Err(QueueError::Kafka(e))

51 let result: Result<(), QueueError> = async {

52 let timeout = Duration::from_millis(self.cfg.publish_queue_timeout_ms as u64);

53 let message_id = Uuid::new_v4().to_string();

54 let headers = OwnedHeaders::new()

55 .insert(Header {

56 key: "message-id",

57 value: Some(&message_id),

58 })

59 .insert(Header {

60 key: "content-type",

61 value: Some("application/octet-stream"),

62 });

63 let mut record = FutureRecord::to(topic).payload(payload).headers(headers);

64 if let Some(k) = key {

65 record = record.key(k);

66 }

67 match self.inner.send(record, timeout).await {

68 Ok(_) => Ok(()),

69 Err((e, _msg)) => {

70 warn!(topic, error = %e, "kafka publish failed");

71 Err(QueueError::Kafka(e))

72 }

7073 }

7174 }

75 .await;

76 metrics::record_publish(&result);

77 result

7278 }

7379

7480 pub fn producer(&self) -> &FutureProducer {