fix(async): keep request ids across background work

Modified lib/chat/src/services/audience.rs +11 −3 View lib/chat/src/services/audience.rs

@@ -59,8 +59,16 @@

5959 text: &str,

6060 html: &str,

6161 ) -> Result<(), ChatError> {

62 email::send_optional(self.mail.as_ref(), to, subject, text, html)

63 .await

64 .map_err(|error| ChatError::Internal(format!("email send failed: {error}")))

62 let request_id = export::middleware::request_id();

63 email::send_optional(

64 self.mail.as_ref(),

65 to,

66 subject,

67 text,

68 html,

69 request_id.as_deref(),

70 )

71 .await

72 .map_err(|error| ChatError::Internal(format!("email send failed: {error}")))

6573 }

6674}

Modified lib/chat/src/services/messages.rs +35 −25 View lib/chat/src/services/messages.rs

@@ -736,18 +736,22 @@

736736 let explicit = mention_uids.clone();

737737 let is_private = ctx.room.is_private;

738738 let space_id = ctx.room.space_id;

739 let request_id = export::middleware::request_id();

739740 tokio::spawn(async move {

740 fanout_everyone_mentions(

741 &svc,

742 crate::share::notify::EveryoneMention {

743 room_id,

744 space_id,

745 is_private,

746 message_id,

747 body,

748 actor: caller,

749 explicit: &explicit,

750 },

741 export::middleware::with_request_id(

742 request_id.as_deref(),

743 fanout_everyone_mentions(

744 &svc,

745 crate::share::notify::EveryoneMention {

746 room_id,

747 space_id,

748 is_private,

749 message_id,

750 body,

751 actor: caller,

752 explicit: &explicit,

753 },

754 ),

751755 )

752756 .await;

753757 });

@@ -788,17 +792,21 @@

788792 } else {

789793 content_for_email.clone()

790794 };

795 let request_id = export::middleware::request_id();

791796 if !recipients.is_empty() {

792797 tokio::spawn(async move {

793 broadcast_direct_message_push(

794 &svc,

795 kind,

796 room_id,

797 space_id,

798 message_id,

799 caller,

800 body,

801 &recipients,

798 export::middleware::with_request_id(

799 request_id.as_deref(),

800 broadcast_direct_message_push(

801 &svc,

802 kind,

803 room_id,

804 space_id,

805 message_id,

806 caller,

807 body,

808 &recipients,

809 ),

802810 )

803811 .await;

804812 });

@@ -818,6 +826,7 @@

818826 content_for_email.clone()

819827 };

820828 let email_svc = self.clone();

829 let request_id = export::middleware::request_id();

821830 tokio::spawn(async move {

822831 let base_url = email_svc.settings.web.web_base_url.clone();

823832 for uid in email_recipients {

@@ -846,11 +855,12 @@

846855 r#"<p>{subject}</p><p>{email_body}</p><p><a href="{link}">{}</a></p>"#,

847856 i18n::catalog::view_channel(locale)

848857 );

849 if let Err(err) = email_svc

850 .audience

851 .send_email(&addr, &subject, &text, &html)

852 .await

853 {

858 let sent = export::middleware::with_request_id(

859 request_id.as_deref(),

860 email_svc.audience.send_email(&addr, &subject, &text, &html),

861 )

862 .await;

863 if let Err(err) = sent {

854864 tracing::warn!(error = %err, "failed to send mention email");

855865 }

856866 }

Modified lib/chat/src/services/outbound.rs +13 −9 View lib/chat/src/services/outbound.rs

@@ -374,15 +374,19 @@

374374 let client = client_builder.build().map_err(|e| e.to_string())?;

375375

376376 let span = oltp::client::http_client_span("POST", &webhook.url);

377 let response = oltp::client::send(

378 span,

379 client

380 .post(&webhook.url)

381 .header("content-type", "application/json")

382 .header("x-gitdata-event", &delivery.event_type)

383 .header("x-gitdata-delivery", delivery.id.to_string())

384 .header("x-gitdata-signature", &signature)

385 .body(body),

377 let delivery_request_id = delivery.id.to_string();

378 let response = export::middleware::with_request_id(

379 Some(delivery_request_id.as_str()),

380 oltp::client::send(

381 span,

382 client

383 .post(&webhook.url)

384 .header("content-type", "application/json")

385 .header("x-gitdata-event", &delivery.event_type)

386 .header("x-gitdata-delivery", delivery.id.to_string())

387 .header("x-gitdata-signature", &signature)

388 .body(body),

389 ),

386390 )

387391 .await

388392 .map_err(|e| {

Modified lib/chat/src/services/unfurl.rs +6 −1 View lib/chat/src/services/unfurl.rs

@@ -136,8 +136,13 @@

136136 return;

137137 }

138138 let service = service.clone();

139 let request_id = export::middleware::request_id();

139140 tokio::spawn(async move {

140 service.unfurl_message(room_id, message_id, content).await;

141 export::middleware::with_request_id(

142 request_id.as_deref(),

143 service.unfurl_message(room_id, message_id, content),

144 )

145 .await;

141146 });

142147}

143148

Modified lib/chat/src/worker.rs +28 −4 View lib/chat/src/worker.rs

@@ -3,6 +3,7 @@

33use models::users::user_recent_mentions;

44use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};

55use time::OffsetDateTime;

6use tracing::Instrument;

67

78use crate::errors::ChatError;

89use crate::services::service::ChatService;

@@ -27,18 +28,41 @@

2728 let mut tick: u32 = 0;

2829 loop {

2930 interval.tick().await;

30 if let Err(error) = service.deliver_due_reminders().await {

31 let request_id = uuid::Uuid::now_v7().to_string();

32 let span = tracing::info_span!("chat.background", request_id = %request_id);

33 if let Err(error) = export::middleware::with_request_id(

34 Some(request_id.as_str()),

35 service.deliver_due_reminders().instrument(span),

36 )

37 .await

38 {

3139 tracing::warn!(%error, "chat reminder delivery failed");

3240 }

33 if let Err(error) = service.deliver_pending_webhook_events().await {

41 let span = tracing::info_span!("chat.background", request_id = %request_id);

42 if let Err(error) = export::middleware::with_request_id(

43 Some(request_id.as_str()),

44 service.deliver_pending_webhook_events().instrument(span),

45 )

46 .await

47 {

3448 tracing::warn!(%error, "chat outbound webhook delivery failed");

3549 }

3650 tick = tick.wrapping_add(1);

3751 if tick.is_multiple_of(120) {

38 if let Err(error) = service.prune_old_mentions().await {

52 if let Err(error) = export::middleware::with_request_id(

53 Some(request_id.as_str()),

54 service.prune_old_mentions(),

55 )

56 .await

57 {

3958 tracing::warn!(%error, "chat mention prune failed");

4059 }

41 if let Err(error) = service.cleanup_stale_attachments().await {

60 if let Err(error) = export::middleware::with_request_id(

61 Some(request_id.as_str()),

62 service.cleanup_stale_attachments(),

63 )

64 .await

65 {

4266 tracing::warn!(%error, "chat attachment cleanup failed");

4367 }

4468 }

Modified lib/deletion/src/lib.rs +10 −1 View lib/deletion/src/lib.rs

@@ -21,6 +21,7 @@

2121};

2222use serde::Serialize;

2323use time::OffsetDateTime;

24use tracing::Instrument;

2425use uuid::Uuid;

2526

2627const RETRY_BASE_SECS: u64 = 30;

@@ -221,7 +222,14 @@

221222 interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);

222223 loop {

223224 interval.tick().await;

224 if let Err(error) = purge_due(&state, &accounts).await {

225 let request_id = Uuid::now_v7().to_string();

226 let span = tracing::info_span!("deletion.worker", request_id = %request_id);

227 let result = export::middleware::with_request_id(

228 Some(&request_id),

229 purge_due(&state, &accounts).instrument(span),

230 )

231 .await;

232 if let Err(error) = result {

225233 tracing::warn!(%error, "deletion worker pass failed");

226234 }

227235 }

@@ -286,6 +294,7 @@

286294 &cfg.kenshell_rpc_key,

287295 Duration::from_secs(cfg.kenshell_timeout_secs.max(1)),

288296 )

297 .map(|client| client.with_request_id(export::middleware::request_id))

289298 .map_err(ken_shell)

290299 }?;

291300 let repos = shell

Modified lib/email/src/consumer.rs +21 −12 View lib/email/src/consumer.rs

@@ -8,7 +8,7 @@

88use queue::{Consumer, ConsumerConfig, MessageHandler};

99use serde::Serialize;

1010use tokio_util::sync::CancellationToken;

11use tracing::{info, warn};

11use tracing::{Instrument, info, warn};

1212use uuid::Uuid;

1313

1414use crate::error::EmailError;

@@ -106,11 +106,22 @@

106106 continue;

107107 }

108108 };

109 match handle.block_on(handler_consumer.deliver(request.clone())) {

110 Ok(sent) => handle.block_on(handler_consumer.emit_sent(&sent)),

109 let request_id = request

110 .trace

111 .as_deref()

112 .filter(|value| !value.is_empty())

113 .map(str::to_owned)

114 .unwrap_or_else(|| request.request_id.to_string());

115 let span = tracing::info_span!("email.delivery", request_id = %request_id);

116 match handle.block_on(handler_consumer.deliver(request.clone()).instrument(span)) {

117 Ok(sent) => handle.block_on(handler_consumer.emit_sent(&sent, &request_id)),

111118 Err(error) => {

112119 success = false;

113 handle.block_on(handler_consumer.emit_failed(&request, &error));

120 handle.block_on(handler_consumer.emit_failed(

121 &request,

122 &error,

123 &request_id,

124 ));

114125 }

115126 }

116127 }

@@ -159,27 +170,25 @@

159170 })

160171 }

161172

162 async fn emit_sent(&self, sent: &EmailSent) {

173 async fn emit_sent(&self, sent: &EmailSent, request_id: &str) {

163174 let Some(events) = self.events.as_ref() else {

164175 return;

165176 };

166 let request_id = sent.request_id.to_string();

167 let labels = Labels::new().with(label::REQUEST_ID, &request_id);

177 let labels = Labels::new().with(label::REQUEST_ID, request_id);

168178 let _ = publish_event(

169179 events,

170180 label::EMAIL_DELIVERY_CREATE,

171181 labels,

172182 sent,

173 &request_id,

183 request_id,

174184 )

175185 .await;

176186 }

177187

178 async fn emit_failed(&self, request: &SendEmailRequest, error: &EmailError) {

188 async fn emit_failed(&self, request: &SendEmailRequest, error: &EmailError, request_id: &str) {

179189 let Some(events) = self.events.as_ref() else {

180190 return;

181191 };

182 let request_id = request.request_id.to_string();

183192 let failed = EmailFailed {

184193 request_id: request.request_id,

185194 error: error.to_string(),

@@ -187,13 +196,13 @@

187196 failed_at_ms: now_ms(),

188197 metadata: request.metadata.clone(),

189198 };

190 let labels = Labels::new().with(label::REQUEST_ID, &request_id);

199 let labels = Labels::new().with(label::REQUEST_ID, request_id);

191200 let _ = publish_event(

192201 events,

193202 label::EMAIL_FAILURE_CREATE,

194203 labels,

195204 &failed,

196 &request_id,

205 request_id,

197206 )

198207 .await;

199208 }

Modified lib/email/src/lib.rs +4 −1 View lib/email/src/lib.rs

@@ -19,10 +19,13 @@

1919 subject: &str,

2020 text: &str,

2121 html: &str,

22 request_id: Option<&str>,

2223) -> Result<(), EmailError> {

2324 let Some(publisher) = publisher else {

2425 tracing::warn!(to, "mail publisher disabled; skipping email");

2526 return Ok(());

2627 };

27 publisher.send_mail(to, subject, text, html, None).await

28 publisher

29 .send_mail(to, subject, text, html, request_id)

30 .await

2831}

Modified lib/email/src/publisher.rs +5 −2 View lib/email/src/publisher.rs

@@ -35,9 +35,10 @@

3535 request.queued_at_ms = now_ms();

3636 }

3737 let key = request.request_id.to_string();

38 let trace_id = request_id.filter(|request_id| !request_id.is_empty());

3839 let mut properties = Vec::with_capacity(2);

3940 properties.push(("request_id".to_owned(), key.clone()));

40 if let Some(request_id) = request_id {

41 if let Some(request_id) = trace_id {

4142 properties.push(("trace_id".to_owned(), request_id.to_owned()));

4243 }

4344 let body = Bytes::from(serde_json::to_vec(&request)?);

@@ -67,7 +68,9 @@

6768 attachments: Vec::new(),

6869 metadata: Default::default(),

6970 queued_at_ms: now_ms(),

70 trace: request_id.map(str::to_owned),

71 trace: request_id

72 .filter(|request_id| !request_id.is_empty())

73 .map(str::to_owned),

7174 };

7275 self.send(request, request_id).await

7376 }

Modified lib/issues/src/services/audience.rs +11 −3 View lib/issues/src/services/audience.rs

@@ -61,9 +61,17 @@

6161 text: &str,

6262 html: &str,

6363 ) -> Result<(), IssueError> {

64 email::send_optional(self.mail.as_ref(), to, subject, text, html)

65 .await

66 .map_err(|error| IssueError::Internal(format!("email send failed: {error}")))

64 let request_id = export::middleware::request_id();

65 email::send_optional(

66 self.mail.as_ref(),

67 to,

68 subject,

69 text,

70 html,

71 request_id.as_deref(),

72 )

73 .await

74 .map_err(|error| IssueError::Internal(format!("email send failed: {error}")))

6775 }

6876

6977 pub(crate) async fn notify_user(

Modified lib/proto/src/client.rs +20 −2 View lib/proto/src/client.rs

@@ -16,6 +16,7 @@

1616#[derive(Clone, Default)]

1717pub struct RpcInterceptor {

1818 key: Option<MetadataValue<Ascii>>,

19 request_id: Option<fn() -> Option<String>>,

1920}

2021

2122impl RpcInterceptor {

@@ -28,7 +29,15 @@

2829 KenShellError::Config("rpc_key is not valid header value".into())

2930 })?)

3031 };

31 Ok(Self { key })

32 Ok(Self {

33 key,

34 request_id: None,

35 })

36 }

37

38 pub fn with_request_id(mut self, provider: fn() -> Option<String>) -> Self {

39 self.request_id = Some(provider);

40 self

3241 }

3342}

3443

@@ -39,7 +48,11 @@

3948 .metadata_mut()

4049 .insert(crate::RPC_KEY_HEADER, key.clone());

4150 }

42 if let Ok(value) = MetadataValue::try_from(Uuid::now_v7().to_string()) {

51 let request_id = self

52 .request_id

53 .and_then(|provider| provider())

54 .unwrap_or_else(|| Uuid::now_v7().to_string());

55 if let Ok(value) = MetadataValue::try_from(request_id) {

4356 request

4457 .metadata_mut()

4558 .insert(crate::REQUEST_ID_HEADER, value);

@@ -99,6 +112,11 @@

99112 self.channel.clone()

100113 }

101114

115 pub fn with_request_id(mut self, provider: fn() -> Option<String>) -> Self {

116 self.interceptor = self.interceptor.with_request_id(provider);

117 self

118 }

119

102120 pub fn connect_timeout(&self) -> Duration {

103121 self.connect_timeout

104122 }

Modified lib/push/src/worker.rs +13 −7 View lib/push/src/worker.rs

@@ -338,14 +338,20 @@

338338 return;

339339 };

340340 let user_id = job.user_id;

341 let request_id = job.request_id.clone();

341342 tokio::spawn(async move {

342 tokio::select! {

343 _ = cancel.cancelled() => return,

344 _ = tokio::time::sleep(Duration::from_secs(backoff)) => {}

345 }

346 if let Err(error) = redis.lpush_bytes(export::push::QUEUE_TOPIC, &payload).await {

347 tracing::warn!(%error, %user_id, "push requeue failed");

348 }

343 let span = tracing::info_span!("push.requeue", request_id = %request_id.as_deref().unwrap_or_default());

344 let requeue = async move {

345 tokio::select! {

346 _ = cancel.cancelled() => return,

347 _ = tokio::time::sleep(Duration::from_secs(backoff)) => {}

348 }

349 if let Err(error) = redis.lpush_bytes(export::push::QUEUE_TOPIC, &payload).await {

350 tracing::warn!(%error, %user_id, "push requeue failed");

351 }

352 };

353 export::middleware::with_request_id(request_id.as_deref(), requeue.instrument(span))

354 .await;

349355 });

350356 }

351357

Modified lib/webapi/src/bootstrap.rs +2 −1 View lib/webapi/src/bootstrap.rs

@@ -87,7 +87,8 @@

8787 "invalid git.kenshell_endpoint `{}`: {error}",

8888 git_config.kenshell_endpoint

8989 )

90 });

90 })

91 .with_request_id(export::middleware::request_id);

9192 let git = git::service::GitService::new(

9293 state.database.clone(),

9394 accounts.clone(),