fix(email): retry publishing mail delivery events
ZhenYi committed
Modified lib/email/src/consumer.rs lib/email/src/consumer.rs +46 −22 View lib/email/src/consumer.rs
@@ -6,6 +6,7 @@
66use events::{EventHandler, Labels, label};
77use lettre::message::Mailbox;
88use queue::{Consumer, ConsumerConfig, MessageHandler};
9use serde::Serialize;
910use tokio_util::sync::CancellationToken;
1011use tracing::{info, warn};
1112use uuid::Uuid;
@@ -14,6 +15,35 @@
1415use crate::events::{EmailFailed, EmailSent, SendEmailRequest, now_ms};
1516use crate::transport::{EmailTransport, EmailTransportBuilder, build_message};
1617
18const EVENT_PUBLISH_ATTEMPTS: usize = 3;
19
20async fn publish_event<T: Serialize>(
21 events: &EventHandler,
22 kind: &str,
23 labels: Labels,
24 payload: &T,
25 request_id: &str,
26) -> bool {
27 let mut attempt = 1_usize;
28 loop {
29 match events
30 .send_labeled(kind, labels.clone(), payload, Some(request_id))
31 .await
32 {
33 Ok(()) => return true,
34 Err(error) if attempt >= EVENT_PUBLISH_ATTEMPTS => {
35 tracing::error!(kind, attempt, %error, "failed to publish the email event");
36 return false;
37 }
38 Err(error) => {
39 tracing::warn!(kind, attempt, %error, "retrying the email event publish");
40 tokio::time::sleep(std::time::Duration::from_millis(50) * attempt as u32).await;
41 attempt += 1;
42 }
43 }
44 }
45}
46
1747pub struct EmailConsumer {
1848 transport: EmailTransport,
1949 from: Mailbox,
@@ -135,17 +165,14 @@
135165 };
136166 let request_id = sent.request_id.to_string();
137167 let labels = Labels::new().with(label::REQUEST_ID, &request_id);
138 if let Err(error) = events
139 .send_labeled(
140 label::EMAIL_DELIVERY_CREATE,
141 labels,
142 sent,
143 Some(request_id.as_str()),
144 )
145 .await
146 {
147 warn!(error = %error, "failed to publish the email delivery event");
148 }
168 let _ = publish_event(
169 events,
170 label::EMAIL_DELIVERY_CREATE,
171 labels,
172 sent,
173 &request_id,
174 )
175 .await;
149176 }
150177
151178 async fn emit_failed(&self, request: &SendEmailRequest, error: &EmailError) {
@@ -161,16 +188,13 @@
161188 metadata: request.metadata.clone(),
162189 };
163190 let labels = Labels::new().with(label::REQUEST_ID, &request_id);
164 if let Err(publish_error) = events
165 .send_labeled(
166 label::EMAIL_FAILURE_CREATE,
167 labels,
168 &failed,
169 Some(request_id.as_str()),
170 )
171 .await
172 {
173 warn!(error = %publish_error, "failed to publish the email failure event");
174 }
191 let _ = publish_event(
192 events,
193 label::EMAIL_FAILURE_CREATE,
194 labels,
195 &failed,
196 &request_id,
197 )
198 .await;
175199 }
176200}