Skip to content
zhenyi

gitdata-core

0
Clone

HTTPS

SSH

SSH needs a key in your settings.

Download ZIP

feat(fanout): implement bounded fanout mechanism for message delivery

ZhenYi committed
parent 213a71b commit 3353252 Browse files

11 files changed

+538 −208

File tree 11
Modified app/src/main.rs View app/src/main.rs
@@ -86,6 +86,7 @@
8686 .context("building AppState")?;
8787 let session_store = auth::session::RedisStore::new(state.cache.clone());
8888 let deps = webapi::bootstrap::Deps::new(state.clone(), session_store.clone());
89 chat::share::fanout::start(deps.chat.clone(), cancel.clone());
8990 info!("dependencies ready");
9091
9192 let internal_rpc = start_internal_rpc(&cfg, &deps, &cancel)?;
Modified config/default/push.json View config/default/push.json
@@ -7,5 +7,7 @@
77 "max_attempts": 3,
88 "retry_backoff_base_secs": 30,
99 "request_timeout_secs": 10,
10 "delivery_concurrency": 4,
11 "job_timeout_secs": 25,
1012 "ttl_secs": 86400
1113}
Modified lib/infra/config/src/push.rs View lib/infra/config/src/push.rs
@@ -20,6 +20,10 @@
2020 pub max_attempts: u32,
2121 pub retry_backoff_base_secs: u64,
2222 pub request_timeout_secs: u64,
23 /// 单个任务内并发投递的设备数。
24 pub delivery_concurrency: usize,
25 /// 单个任务的总时间预算；超出后放弃本次投递。
26 pub job_timeout_secs: u64,
2327 /// 推送消息在推送服务上的存活时间（TTL 头）。
2428 pub ttl_secs: u64,
2529}
@@ -35,6 +39,8 @@
3539 max_attempts: 3,
3640 retry_backoff_base_secs: 30,
3741 request_timeout_secs: 10,
42 delivery_concurrency: 4,
43 job_timeout_secs: 25,
3844 ttl_secs: 86_400,
3945 }
4046 }
@@ -65,10 +71,17 @@
6571 if self.workers == 0
6672 || self.max_attempts == 0
6773 || self.request_timeout_secs == 0
74 || self.delivery_concurrency == 0
75 || self.job_timeout_secs == 0
6876 || self.ttl_secs == 0
6977 {
7078 return Err("push numeric settings must be greater than zero".to_string());
7179 }
80 if self.job_timeout_secs < self.request_timeout_secs {
81 return Err(
82 "push.job_timeout_secs must be at least push.request_timeout_secs".to_string(),
83 );
84 }
7285 Ok(())
7386 }
7487
Modified lib/infra/oltp/src/metrics.rs View lib/infra/oltp/src/metrics.rs
@@ -178,6 +178,13 @@
178178 .build()
179179});
180180
181static CHAT_FANOUT: LazyLock<Counter<u64>> = LazyLock::new(|| {
182 app_meter()
183 .u64_counter("chat.fanout")
184 .with_description("Chat fan-out jobs by outcome")
185 .build()
186});
187
181188static QUEUE_DEPTH: LazyLock<Gauge<i64>> = LazyLock::new(|| {
182189 app_meter()
183190 .i64_gauge("queue.depth")
@@ -232,6 +239,10 @@
232239 CHAT_WEBHOOK_DELIVERIES.add(1, &[KeyValue::new("outcome", outcome)]);
233240}
234241
242pub fn chat_fanout(outcome: &'static str) {
243 CHAT_FANOUT.add(1, &[KeyValue::new("outcome", outcome)]);
244}
245
235246pub fn queue_depth(topic: &'static str, depth: i64) {
236247 QUEUE_DEPTH.record(depth, &[KeyValue::new("topic", topic)]);
237248}
Modified lib/service/chat/src/services/messages.rs View lib/service/chat/src/services/messages.rs
@@ -31,13 +31,13 @@
3131}
3232use crate::services::reactions::ReactionAggDto;
3333use crate::services::service::ChatService;
34use crate::share::fanout::{self, Job};
3435use crate::share::lookup::{
3536 RoomContext, require_room_readable, require_room_writable, space_slug_for,
3637};
3738use crate::share::notify::{
38 NotifyDecision, advance_read_state, current_room_seq, fanout_everyone_mentions,
39 mention_email_allowed, next_room_seq, notification_decision, notify_user,
40 publish_message_event, record_mentions,
39 NotifyDecision, advance_read_state, current_room_seq, next_room_seq, notification_decision,
40 notify_user, publish_message_event, record_mentions,
4141};
4242use crate::share::realtime::RoomEvent;
4343use crate::share::{authz, mention};
@@ -57,7 +57,7 @@
5757/// `notification_decision` the mention path uses, so "muted" and "mentions
5858/// only" mean the same thing here as everywhere else.
5959#[allow(clippy::too_many_arguments)]
60async fn broadcast_direct_message_push(
60pub(crate) async fn broadcast_direct_message_push(
6161 service: &ChatService,
6262 kind: DirectMessageKind,
6363 room_id: Uuid,
@@ -723,33 +723,23 @@
723723 .await;
724724
725725 if mention_everyone {
726 let svc = self.clone();
727726 let body = if content_for_email.chars().count() > 120 {
728727 content_for_email.chars().take(120).collect::<String>()
729728 } else {
730729 content_for_email.clone()
731730 };
732 let explicit = mention_uids.clone();
733 let is_private = ctx.room.is_private;
734 let space_id = ctx.room.space_id;
735 let request_id = api::middleware::request_id();
736 tokio::spawn(async move {
737 api::middleware::with_request_id(
738 request_id.as_deref(),
739 fanout_everyone_mentions(
740 &svc,
741 crate::share::notify::EveryoneMention {
742 room_id,
743 space_id,
744 is_private,
745 message_id,
746 body,
747 actor: caller,
748 explicit: &explicit,
749 },
750 ),
751 )
752 .await;
731 fanout::submit(Job::EveryoneStart {
732 scope: crate::share::notify::EveryoneScope {
733 room_id,
734 space_id: ctx.room.space_id,
735 is_private: ctx.room.is_private,
736 message_id,
737 body,
738 actor: caller,
739 explicit: mention_uids.clone(),
740 request_id: api::middleware::request_id(),
741 trace: oltp::propagation::traceparent(),
742 },
753743 });
754744 }
755745 crate::services::unfurl::spawn_unfurl(self, room_id, message_id, content_for_email.clone());
@@ -781,30 +771,23 @@
781771 _ => None,
782772 }
783773 } {
784 let svc = self.clone();
785774 let recipients = crate::share::notify::dm_recipients(self, room_id, caller).await;
786775 let body = if content_for_email.chars().count() > 120 {
787776 content_for_email.chars().take(120).collect::<String>()
788777 } else {
789778 content_for_email.clone()
790779 };
791 let request_id = api::middleware::request_id();
792780 if !recipients.is_empty() {
793 tokio::spawn(async move {
794 api::middleware::with_request_id(
795 request_id.as_deref(),
796 broadcast_direct_message_push(
797 &svc,
798 kind,
799 room_id,
800 space_id,
801 message_id,
802 caller,
803 body,
804 &recipients,
805 ),
806 )
807 .await;
781 fanout::submit(Job::DirectMessagePush {
782 kind,
783 room_id,
784 space_id,
785 message_id,
786 actor: caller,
787 body,
788 recipients,
789 request_id: api::middleware::request_id(),
790 trace: oltp::propagation::traceparent(),
808791 });
809792 }
810793 }
@@ -816,51 +799,22 @@
816799 .copied()
817800 .filter(|uid| *uid != caller)
818801 .collect();
819 let email_body = if content_for_email.chars().count() > 200 {
820 content_for_email.chars().take(200).collect::<String>()
821 } else {
822 content_for_email.clone()
823 };
824 let email_svc = self.clone();
825 let request_id = api::middleware::request_id();
826 tokio::spawn(async move {
827 let base_url = email_svc.settings.web.web_base_url.clone();
828 for uid in email_recipients {
829 if !mention_email_allowed(&email_svc, room_id, space_id, uid)
830 .await
831 .unwrap_or(false)
832 {
833 continue;
834 }
835 let Ok(mut emails) = email_svc.audience.primary_emails(&[uid]).await else {
836 continue;
837 };
838 let Some(addr) = emails.remove(&uid) else {
839 continue;
840 };
841 // One recipient per iteration, so the mail follows that
842 // person's own language rather than the sender's.
843 let locale = email_svc.audience.recipient_locale(uid).await;
844 let subject = i18n::catalog::mention_email_subject(locale, &room_name);
845 let link = format!("{base_url}/chat/rooms/{room_id}");
846 let text = format!(
847 "{subject}\n\n{email_body}\n\n{}: {link}",
848 i18n::catalog::view_channel(locale)
849 );
850 let html = format!(
851 r#"<p>{subject}</p><p>{email_body}</p><p><a href="{link}">{}</a></p>"#,
852 i18n::catalog::view_channel(locale)
853 );
854 let sent = api::middleware::with_request_id(
855 request_id.as_deref(),
856 email_svc.audience.send_email(&addr, &subject, &text, &html),
857 )
858 .await;
859 if let Err(err) = sent {
860 tracing::warn!(error = %err, "failed to send mention email");
861 }
862 }
863 });
802 if !email_recipients.is_empty() {
803 let email_body = if content_for_email.chars().count() > 200 {
804 content_for_email.chars().take(200).collect::<String>()
805 } else {
806 content_for_email.clone()
807 };
808 fanout::submit(Job::MentionEmails {
809 room_id,
810 space_id,
811 room_name,
812 body: email_body,
813 recipients: email_recipients,
814 request_id: api::middleware::request_id(),
815 trace: oltp::propagation::traceparent(),
816 });
817 }
864818 }
865819
866820 Ok(dto)
Added lib/service/chat/src/share/fanout.rs View lib/service/chat/src/share/fanout.rs
@@ -0,0 +1,312 @@
1//! 消息路径的扇出有界化：固定 worker 池替代裸 `tokio::spawn`。
2//!
3//! 两条 lane：`interactive` 承接 DM 推送与 @mention 邮件（单 job 有界），
4//! `bulk` 承接 @everyone（按 `EVERYONE_CHUNK` 切块）。队列满时 `try_send`
5//! 丢弃并打点，绝不把背压施加到消息写入路径。
6
7use std::sync::{Arc, OnceLock};
8
9use tokio::sync::{Mutex, mpsc};
10use tokio_util::sync::CancellationToken;
11use tracing::Instrument as _;
12use uuid::Uuid;
13
14use crate::services::messages::DirectMessageKind;
15use crate::services::service::ChatService;
16use crate::share::notify::{
17 EveryoneScope, everyone_candidates, fanout_everyone_chunk, mention_email_allowed,
18};
19
20const INTERACTIVE_WORKERS: usize = 2;
21const INTERACTIVE_CAPACITY: usize = 256;
22const BULK_WORKERS: usize = 2;
23const BULK_CAPACITY: usize = 64;
24/// @everyone 单块人数：一块的数据库往返次数有界，块与块之间让出 worker。
25const EVERYONE_CHUNK: usize = 100;
26
27pub(crate) enum Job {
28 MentionEmails {
29 room_id: Uuid,
30 space_id: Option<Uuid>,
31 room_name: String,
32 body: String,
33 recipients: Vec<i64>,
34 request_id: Option<String>,
35 trace: Option<String>,
36 },
37 DirectMessagePush {
38 kind: DirectMessageKind,
39 room_id: Uuid,
40 space_id: Option<Uuid>,
41 message_id: Uuid,
42 actor: i64,
43 body: String,
44 recipients: Vec<i64>,
45 request_id: Option<String>,
46 trace: Option<String>,
47 },
48 EveryoneStart {
49 scope: EveryoneScope,
50 },
51 EveryoneChunk {
52 scope: Arc<EveryoneScope>,
53 recipients: Vec<i64>,
54 },
55}
56
57impl Job {
58 fn kind(&self) -> &'static str {
59 match self {
60 Self::MentionEmails { .. } => "mention_emails",
61 Self::DirectMessagePush { .. } => "direct_message_push",
62 Self::EveryoneStart { .. } => "everyone_start",
63 Self::EveryoneChunk { .. } => "everyone_chunk",
64 }
65 }
66
67 fn request_id(&self) -> Option<&str> {
68 match self {
69 Self::MentionEmails { request_id, .. } | Self::DirectMessagePush { request_id, .. } => {
70 request_id.as_deref()
71 }
72 Self::EveryoneStart { scope } => scope.request_id.as_deref(),
73 Self::EveryoneChunk { scope, .. } => scope.request_id.as_deref(),
74 }
75 }
76
77 fn trace(&self) -> Option<&str> {
78 match self {
79 Self::MentionEmails { trace, .. } | Self::DirectMessagePush { trace, .. } => {
80 trace.as_deref()
81 }
82 Self::EveryoneStart { scope } => scope.trace.as_deref(),
83 Self::EveryoneChunk { scope, .. } => scope.trace.as_deref(),
84 }
85 }
86}
87
88struct Fanout {
89 interactive: mpsc::Sender<Job>,
90 bulk: mpsc::Sender<Job>,
91}
92
93static FANOUT: OnceLock<Fanout> = OnceLock::new();
94
95/// 启动固定 worker 池；必须在 tokio 运行时内调用一次。
96pub fn start(service: ChatService, cancel: CancellationToken) {
97 let (interactive_tx, interactive_rx) = mpsc::channel(INTERACTIVE_CAPACITY);
98 let (bulk_tx, bulk_rx) = mpsc::channel(BULK_CAPACITY);
99 let interactive_rx = Arc::new(Mutex::new(interactive_rx));
100 let bulk_rx = Arc::new(Mutex::new(bulk_rx));
101
102 for _ in 0..INTERACTIVE_WORKERS {
103 tokio::spawn(worker_loop(
104 service.clone(),
105 interactive_rx.clone(),
106 bulk_tx.clone(),
107 cancel.clone(),
108 ));
109 }
110 for _ in 0..BULK_WORKERS {
111 tokio::spawn(worker_loop(
112 service.clone(),
113 bulk_rx.clone(),
114 bulk_tx.clone(),
115 cancel.clone(),
116 ));
117 }
118
119 if FANOUT
120 .set(Fanout {
121 interactive: interactive_tx,
122 bulk: bulk_tx,
123 })
124 .is_err()
125 {
126 tracing::warn!("chat fanout already started");
127 } else {
128 tracing::info!(
129 interactive = INTERACTIVE_WORKERS,
130 bulk = BULK_WORKERS,
131 "chat fanout pool started"
132 );
133 }
134}
135
136/// 投递一个扇出任务；队列满或未启动时丢弃并打点，不阻塞调用方。
137pub(crate) fn submit(job: Job) {
138 let Some(fanout) = FANOUT.get() else {
139 tracing::warn!(kind = job.kind(), "chat fanout not started; dropping job");
140 oltp::metrics::chat_fanout("unstarted");
141 return;
142 };
143 let (tx, lane) = match &job {
144 Job::EveryoneStart { .. } | Job::EveryoneChunk { .. } => (&fanout.bulk, "bulk"),
145 _ => (&fanout.interactive, "interactive"),
146 };
147 match tx.try_send(job) {
148 Ok(()) => oltp::metrics::chat_fanout("queued"),
149 Err(mpsc::error::TrySendError::Full(job)) => {
150 tracing::warn!(
151 kind = job.kind(),
152 lane,
153 "chat fanout queue full; dropping job"
154 );
155 oltp::metrics::chat_fanout("full");
156 }
157 Err(mpsc::error::TrySendError::Closed(job)) => {
158 tracing::warn!(
159 kind = job.kind(),
160 lane,
161 "chat fanout queue closed; dropping job"
162 );
163 oltp::metrics::chat_fanout("closed");
164 }
165 }
166}
167
168async fn worker_loop(
169 service: ChatService,
170 rx: Arc<Mutex<mpsc::Receiver<Job>>>,
171 bulk: mpsc::Sender<Job>,
172 cancel: CancellationToken,
173) {
174 loop {
175 let job = {
176 let mut guard = rx.lock().await;
177 tokio::select! {
178 _ = cancel.cancelled() => return,
179 job = guard.recv() => job,
180 }
181 };
182 match job {
183 Some(job) => run(&service, job, &bulk).await,
184 None => return,
185 }
186 }
187}
188
189async fn run(service: &ChatService, job: Job, bulk: &mpsc::Sender<Job>) {
190 let request_id = job.request_id().map(str::to_owned);
191 let trace = job.trace().map(str::to_owned);
192 let span = tracing::info_span!(
193 "chat.fanout",
194 kind = job.kind(),
195 request_id = tracing::field::Empty
196 );
197 if let Some(request_id) = request_id.as_deref() {
198 span.record("request_id", request_id);
199 }
200 oltp::propagation::set_parent_from_traceparent(&span, trace.as_deref());
201 api::middleware::with_request_id(
202 request_id.as_deref(),
203 run_job(service, job, bulk).instrument(span),
204 )
205 .await;
206 oltp::metrics::chat_fanout("done");
207}
208
209async fn run_job(service: &ChatService, job: Job, bulk: &mpsc::Sender<Job>) {
210 match job {
211 Job::MentionEmails {
212 room_id,
213 space_id,
214 room_name,
215 body,
216 recipients,
217 ..
218 } => mention_emails(service, room_id, space_id, room_name, body, recipients).await,
219 Job::DirectMessagePush {
220 kind,
221 room_id,
222 space_id,
223 message_id,
224 actor,
225 body,
226 recipients,
227 ..
228 } => {
229 crate::services::messages::broadcast_direct_message_push(
230 service,
231 kind,
232 room_id,
233 space_id,
234 message_id,
235 actor,
236 body,
237 &recipients,
238 )
239 .await;
240 }
241 Job::EveryoneStart { scope } => everyone_start(service, scope, bulk).await,
242 Job::EveryoneChunk { scope, recipients } => {
243 fanout_everyone_chunk(service, &scope, &recipients).await;
244 }
245 }
246}
247
248async fn everyone_start(service: &ChatService, scope: EveryoneScope, bulk: &mpsc::Sender<Job>) {
249 let candidates = everyone_candidates(service, &scope).await;
250 let scope = Arc::new(scope);
251 for chunk in candidates.chunks(EVERYONE_CHUNK) {
252 let job = Job::EveryoneChunk {
253 scope: scope.clone(),
254 recipients: chunk.to_vec(),
255 };
256 match bulk.try_send(job) {
257 Ok(()) => {}
258 // 队列已满时自己消化这一块：不丢通知，也不会互相等待而死锁。
259 Err(mpsc::error::TrySendError::Full(job)) => {
260 if let Job::EveryoneChunk { scope, recipients } = job {
261 fanout_everyone_chunk(service, &scope, &recipients).await;
262 }
263 }
264 Err(mpsc::error::TrySendError::Closed(_)) => break,
265 }
266 }
267}
268
269async fn mention_emails(
270 service: &ChatService,
271 room_id: Uuid,
272 space_id: Option<Uuid>,
273 room_name: String,
274 body: String,
275 recipients: Vec<i64>,
276) {
277 let Ok(mut emails) = service.audience.primary_emails(&recipients).await else {
278 return;
279 };
280 let fallback = service.audience.default_locale;
281 let locales = i18n::resolve_user_locales(service.database.read(), &recipients, fallback).await;
282 let base_url = service.settings.web.web_base_url.clone();
283 for uid in recipients {
284 if !mention_email_allowed(service, room_id, space_id, uid)
285 .await
286 .unwrap_or(false)
287 {
288 continue;
289 }
290 let Some(addr) = emails.remove(&uid) else {
291 continue;
292 };
293 let locale = locales.get(&uid).copied().unwrap_or(fallback);
294 let subject = i18n::catalog::mention_email_subject(locale, &room_name);
295 let link = format!("{base_url}/chat/rooms/{room_id}");
296 let text = format!(
297 "{subject}\n\n{body}\n\n{}: {link}",
298 i18n::catalog::view_channel(locale)
299 );
300 let html = format!(
301 r#"<p>{subject}</p><p>{body}</p><p><a href="{link}">{}</a></p>"#,
302 i18n::catalog::view_channel(locale)
303 );
304 if let Err(error) = service
305 .audience
306 .send_email(&addr, &subject, &text, &html)
307 .await
308 {
309 tracing::warn!(%error, "failed to send mention email");
310 }
311 }
312}
Modified lib/service/chat/src/share/mod.rs View lib/service/chat/src/share/mod.rs
@@ -1,5 +1,6 @@
11pub mod authz;
22pub mod events;
3pub mod fanout;
34pub mod guard;
45pub mod lookup;
56pub mod mention;
Modified lib/service/chat/src/share/notify.rs View lib/service/chat/src/share/notify.rs
@@ -164,31 +164,26 @@
164164
165165/// One @everyone fan-out. Grouped into a struct because the room scope, the
166166/// message and the actor are all `Uuid`-shaped and would be easy to transpose.
167pub struct EveryoneMention<'a> {
167#[derive(Clone)]
168pub struct EveryoneScope {
168169 pub room_id: Uuid,
169170 pub space_id: Option<Uuid>,
170171 pub is_private: bool,
171172 pub message_id: Uuid,
172173 pub body: String,
173174 pub actor: i64,
174 pub explicit: &'a [i64],
175 pub explicit: Vec<i64>,
176 pub request_id: Option<String>,
177 pub trace: Option<String>,
175178}
176179
177/// Fan out an @everyone mention to room/space members that were not explicitly
178/// mentioned. Runs best-effort after commit: writes notifications and recent
179/// mentions, then pushes a WS user event.
180pub async fn fanout_everyone_mentions(service: &ChatService, scope: EveryoneMention<'_>) {
181 let EveryoneMention {
182 room_id,
183 space_id,
184 is_private,
185 message_id,
186 body,
187 actor,
188 explicit,
189 } = scope;
190 let candidates: Vec<i64> = if space_id.is_none() {
191 room_members::Entity::find()
180/// The recipients of an @everyone mention: room/space members minus the actor
181/// and the explicitly mentioned users.
182pub async fn everyone_candidates(service: &ChatService, scope: &EveryoneScope) -> Vec<i64> {
183 let room_id = scope.room_id;
184 let actor = scope.actor;
185 let candidates: Vec<i64> = match scope.space_id {
186 None => room_members::Entity::find()
192187 .filter(room_members::Column::RoomId.eq(room_id))
193188 .filter(room_members::Column::LeftAt.is_null())
194189 .all(service.database.read())
@@ -196,48 +191,58 @@
196191 .unwrap_or_default()
197192 .into_iter()
198193 .map(|m| m.user_id)
199 .collect()
200 } else {
201 let space_id = space_id.expect("space channel");
202 let Ok(space) = service.participants.access.by_id(space_id).await else {
203 return;
204 };
205 let mut ids: Vec<i64> = space
206 .members(actor)
207 .await
208 .unwrap_or_default()
209 .into_iter()
210 .filter(|m| !m.suppress_everyone)
211 .map(|m| m.user_id)
212 .collect();
213 if is_private {
214 let dir_id = rooms::Entity::find_by_id(room_id)
215 .one(service.database.read())
194 .collect(),
195 Some(space_id) => {
196 let Ok(space) = service.participants.access.by_id(space_id).await else {
197 return Vec::new();
198 };
199 let mut ids: Vec<i64> = space
200 .members(actor)
216201 .await
217 .ok()
218 .flatten()
219 .and_then(|room| room.dir_id);
220 match dir_id {
221 Some(dir_id) => match space.dir_visible_users(dir_id, &ids).await {
222 Ok(visible) => ids.retain(|uid| visible.contains(uid)),
223 Err(_) => ids.clear(),
224 },
225 None => ids.clear(),
202 .unwrap_or_default()
203 .into_iter()
204 .filter(|m| !m.suppress_everyone)
205 .map(|m| m.user_id)
206 .collect();
207 if scope.is_private {
208 let dir_id = rooms::Entity::find_by_id(room_id)
209 .one(service.database.read())
210 .await
211 .ok()
212 .flatten()
213 .and_then(|room| room.dir_id);
214 match dir_id {
215 Some(dir_id) => match space.dir_visible_users(dir_id, &ids).await {
216 Ok(visible) => ids.retain(|uid| visible.contains(uid)),
217 Err(_) => ids.clear(),
218 },
219 None => ids.clear(),
220 }
226221 }
222 ids
227223 }
228 ids
229224 };
230225
226 candidates
227 .into_iter()
228 .filter(|uid| *uid != actor && !scope.explicit.contains(uid))
229 .collect()
230}
231
232/// Fan out an @everyone mention to one bounded batch of recipients: writes
233/// notifications and recent mentions, then pushes a WS user event.
234pub async fn fanout_everyone_chunk(
235 service: &ChatService,
236 scope: &EveryoneScope,
237 recipients: &[i64],
238) {
231239 let data = serde_json::json!({
232 "room_id": room_id,
233 "message_id": message_id,
240 "room_id": scope.room_id,
241 "message_id": scope.message_id,
234242 "everyone": true,
235243 });
236 for uid in candidates {
237 if uid == actor || explicit.contains(&uid) {
238 continue;
239 }
240 match notification_decision(service, room_id, space_id, uid, true, true).await {
244 for &uid in recipients {
245 match notification_decision(service, scope.room_id, scope.space_id, uid, true, true).await {
241246 Ok(NotifyDecision::Notify) => {}
242247 _ => continue,
243248 }
@@ -245,8 +250,8 @@
245250 let _ = user_recent_mentions::ActiveModel {
246251 id: Set(Uuid::now_v7()),
247252 user_id: Set(uid),
248 channel_id: Set(room_id),
249 message_id: Set(message_id),
253 channel_id: Set(scope.room_id),
254 message_id: Set(scope.message_id),
250255 created_at: Set(now),
251256 }
252257 .insert(service.database.write())
@@ -261,15 +266,15 @@
261266 service.audience.recipient_locale(uid).await,
262267 )
263268 .to_string(),
264 Some(body.clone()),
269 Some(scope.body.clone()),
265270 Some(data.clone()),
266 Some(actor),
271 Some(scope.actor),
267272 )
268273 .await;
269274 service.audience.realtime
270275 .publish_user(
271276 uid,
272 &serde_json::json!({"kind": "mention", "room_id": room_id, "message_id": message_id, "everyone": true}),
277 &serde_json::json!({"kind": "mention", "room_id": scope.room_id, "message_id": scope.message_id, "everyone": true}),
273278 )
274279 .await;
275280 }
Modified lib/service/push/src/worker.rs View lib/service/push/src/worker.rs
@@ -10,10 +10,12 @@
1010
1111use crate::protocol::{PushJob, online_key};
1212use bootstrap::AppState;
13use futures::StreamExt as _;
1314use models::users::user_preferences;
1415use models::users::user_push_subscriptions::{self, PushSubscriptionPlatform};
1516use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};
1617use tracing::Instrument as _;
18use uuid::Uuid;
1719
1820use crate::client::{DeliveryOutcome, NotificationPayload, SubscriptionKeys, VapidKeys, deliver};
1921
@@ -95,11 +97,20 @@
9597 span.record("request_id", request_id);
9698 }
9799 oltp::propagation::set_parent_from_traceparent(&span, job.trace.as_deref());
98 if let Err(error) = api::middleware::with_request_id(
99 job.request_id.as_deref(),
100 self.handle(&job).instrument(span),
101 )
102 .await
100 let budget = Duration::from_secs(self.state.config.push.job_timeout_secs.max(1));
101 let delivery = async {
102 match tokio::time::timeout(budget, self.handle(&job).instrument(span)).await {
103 Ok(result) => result,
104 Err(_) => {
105 oltp::metrics::push_delivery("timeout");
106 tracing::warn!("push job timed out; dropping this attempt");
107 self.forget(&job);
108 Ok(())
109 }
110 }
111 };
112 if let Err(error) =
113 api::middleware::with_request_id(job.request_id.as_deref(), delivery).await
103114 {
104115 tracing::warn!(
105116 %error,
@@ -160,66 +171,20 @@
160171
161172 let payload = self.build_payload(job).await?;
162173 let ttl_secs = self.state.config.push.ttl_secs;
163 let mut retry = false;
164 for subscription in subscriptions {
165 let keys = SubscriptionKeys {
166 endpoint: subscription.endpoint.clone(),
167 p256dh_key: subscription.p256dh_key.clone(),
168 auth_key: subscription.auth_key.clone(),
169 };
170 match deliver(&self.client, &self.keys, &keys, &payload, ttl_secs).await {
171 Ok(DeliveryOutcome::Delivered) => {
172 oltp::metrics::push_delivery("delivered");
173 tracing::debug!(user = %job.user_id, "push delivered");
174 }
175 Ok(DeliveryOutcome::Gone) => {
176 oltp::metrics::push_delivery("gone");
177 // 用户清了浏览器数据或卸载了站点：一次性清理，不是错误。
178 tracing::debug!(user = %job.user_id, "push subscription expired; removing");
179 if let Err(error) =
180 user_push_subscriptions::Entity::delete_by_id(subscription.id)
181 .exec(self.state.database.write())
182 .await
183 {
184 tracing::warn!(%error, "failed to remove expired push subscription");
185 }
186 }
187 Ok(DeliveryOutcome::Rejected) => {
188 oltp::metrics::push_delivery("rejected");
189 tracing::warn!(
190 user = %job.user_id,
191 endpoint = %subscription.endpoint,
192 "push rejected by service; sending to dlq"
193 );
194 self.to_dlq(job).await;
195 self.forget(job);
196 return Ok(());
197 }
198 Ok(DeliveryOutcome::Retry) => {
199 oltp::metrics::push_delivery("retry");
200 retry = true;
201 }
202 Err(PushError::Subscription(message)) => {
203 oltp::metrics::push_delivery("subscription_error");
204 // 订阅本身不可修（endpoint 或密钥解析失败），重投没有意义。
205 tracing::warn!(user = %job.user_id, %message, "dropping broken subscription");
206 if let Err(error) =
207 user_push_subscriptions::Entity::delete_by_id(subscription.id)
208 .exec(self.state.database.write())
209 .await
210 {
211 tracing::warn!(%error, "failed to remove broken push subscription");
212 }
213 }
214 Err(error) => {
215 oltp::metrics::push_delivery("error");
216 tracing::warn!(%error, user = %job.user_id, "push delivery error");
217 retry = true;
218 }
219 }
220 }
174 let concurrency = self.state.config.push.delivery_concurrency.max(1);
175 let outcomes: Vec<DeliveryOutcome> = futures::stream::iter(
176 subscriptions
177 .into_iter()
178 .map(|subscription| self.deliver_one(subscription, &payload, ttl_secs)),
179 )
180 .buffer_unordered(concurrency)
181 .collect()
182 .await;
221183
222 if retry {
184 if outcomes.contains(&DeliveryOutcome::Rejected) {
185 self.to_dlq(job).await;
186 self.forget(job);
187 } else if outcomes.contains(&DeliveryOutcome::Retry) {
223188 self.retry_or_drop(job).await;
224189 } else {
225190 self.forget(job);
@@ -227,6 +192,70 @@
227192 Ok(())
228193 }
229194
195 /// 向一台设备投递；订阅级副作用（清理失效行）就地完成。
196 ///
197 /// 返回 `Gone` 表示订阅已被移除（过期或密钥不可解析），不算投递失败——它
198 /// 与其他设备的结果一起汇总，单台设备的永久拒绝不再中断其余设备。
199 async fn deliver_one(
200 &self,
201 subscription: user_push_subscriptions::Model,
202 payload: &NotificationPayload,
203 ttl_secs: u64,
204 ) -> DeliveryOutcome {
205 let keys = SubscriptionKeys {
206 endpoint: subscription.endpoint.clone(),
207 p256dh_key: subscription.p256dh_key.clone(),
208 auth_key: subscription.auth_key.clone(),
209 };
210 match deliver(&self.client, &self.keys, &keys, payload, ttl_secs).await {
211 Ok(DeliveryOutcome::Delivered) => {
212 oltp::metrics::push_delivery("delivered");
213 tracing::debug!("push delivered");
214 DeliveryOutcome::Delivered
215 }
216 Ok(DeliveryOutcome::Gone) => {
217 oltp::metrics::push_delivery("gone");
218 // 用户清了浏览器数据或卸载了站点：一次性清理，不是错误。
219 tracing::debug!("push subscription expired; removing");
220 self.remove_subscription(subscription.id).await;
221 DeliveryOutcome::Gone
222 }
223 Ok(DeliveryOutcome::Rejected) => {
224 oltp::metrics::push_delivery("rejected");
225 tracing::warn!(
226 endpoint = %subscription.endpoint,
227 "push rejected by service; sending to dlq"
228 );
229 DeliveryOutcome::Rejected
230 }
231 Ok(DeliveryOutcome::Retry) => {
232 oltp::metrics::push_delivery("retry");
233 DeliveryOutcome::Retry
234 }
235 Err(PushError::Subscription(message)) => {
236 oltp::metrics::push_delivery("subscription_error");
237 // 订阅本身不可修（endpoint 或密钥解析失败），重投没有意义。
238 tracing::warn!(%message, "dropping broken subscription");
239 self.remove_subscription(subscription.id).await;
240 DeliveryOutcome::Gone
241 }
242 Err(error) => {
243 oltp::metrics::push_delivery("error");
244 tracing::warn!(%error, "push delivery error");
245 DeliveryOutcome::Retry
246 }
247 }
248 }
249
250 async fn remove_subscription(&self, id: Uuid) {
251 if let Err(error) = user_push_subscriptions::Entity::delete_by_id(id)
252 .exec(self.state.database.write())
253 .await
254 {
255 tracing::warn!(%error, "failed to remove push subscription");
256 }
257 }
258
230259 /// 用户偏好里的禁用位；`Other` 类不受位控制，只受"是否有订阅"约束。
231260 async fn skip_by_preferences(&self, job: &PushJob) -> Result<bool, PushError> {
232261 let Some(flags) = user_preferences::Entity::find_by_id(job.user_id)
Modified lib/service/push/Cargo.toml View lib/service/push/Cargo.toml
@@ -23,6 +23,7 @@
2323reqwest = { workspace = true }
2424tokio = { workspace = true, features = ["time", "sync", "rt"] }
2525tokio-util = { workspace = true }
26futures = { workspace = true }
2627time = { workspace = true }
2728# `web-push-native` 只提供无 io 的"生成密文请求"能力，HTTP 用本仓既有的
2829# reqwest（rustls）；两者都必须落在 http 1.x 上，否则 `http::Uri` 不是同一个类型。
Modified Cargo.lock View Cargo.lock
@@ -5751,6 +5751,7 @@
57515751 "config 1.0.0",
57525752 "db",
57535753 "events",
5754 "futures",
57545755 "http 1.5.0",
57555756 "models",
57565757 "oltp",