feat(fanout): enhance message delivery lanes with dedicated DM and email channels

Modified lib/infra/oltp/src/metrics.rs +8 −2 View lib/infra/oltp/src/metrics.rs

@@ -239,8 +239,14 @@

239239 CHAT_WEBHOOK_DELIVERIES.add(1, &[KeyValue::new("outcome", outcome)]);

240240}

241241

242pub fn chat_fanout(outcome: &'static str) {

243 CHAT_FANOUT.add(1, &[KeyValue::new("outcome", outcome)]);

242pub fn chat_fanout(lane: &'static str, outcome: &'static str) {

243 CHAT_FANOUT.add(

244 1,

245 &[

246 KeyValue::new("lane", lane),

247 KeyValue::new("outcome", outcome),

248 ],

249 );

244250}

245251

246252pub fn queue_depth(topic: &'static str, depth: i64) {

Modified lib/service/chat/src/share/fanout.rs +80 −25 View lib/service/chat/src/share/fanout.rs

@@ -1,8 +1,8 @@

11//! 消息路径的扇出有界化：固定 worker 池替代裸 `tokio::spawn`。

22//!

3//! 两条 lane：`interactive` 承接 DM 推送与 @mention 邮件（单 job 有界），

4//! `bulk` 承接 @everyone（按 `EVERYONE_CHUNK` 切块）。队列满时 `try_send`

5//! 丢弃并打点，绝不把背压施加到消息写入路径。

3//! 三条 lane：`dm` 承接 DM 推送，`email` 承接 @mention 邮件（单 job 有界），

4//! `bulk` 承接 @everyone（按 `EVERYONE_CHUNK` 切块）。队列满时转后台等待

5//! 投递，只有 channel 关闭时才丢弃，绝不把背压施加到消息写入路径。

66

77use std::sync::{Arc, OnceLock};

88

@@ -17,8 +17,10 @@

1717 EveryoneScope, everyone_candidates, fanout_everyone_chunk, mention_email_allowed,

1818};

1919

20const INTERACTIVE_WORKERS: usize = 2;

21const INTERACTIVE_CAPACITY: usize = 256;

20const DM_WORKERS: usize = 2;

21const DM_CAPACITY: usize = 256;

22const EMAIL_WORKERS: usize = 2;

23const EMAIL_CAPACITY: usize = 64;

2224const BULK_WORKERS: usize = 2;

2325const BULK_CAPACITY: usize = 64;

2426/// @everyone 单块人数：一块的数据库往返次数有界，块与块之间让出 worker。

@@ -85,8 +87,34 @@

8587 }

8688}

8789

90#[derive(Clone, Copy)]

91enum Lane {

92 Dm,

93 Email,

94 Bulk,

95}

96

97impl Lane {

98 fn of(job: &Job) -> Self {

99 match job {

100 Job::DirectMessagePush { .. } => Self::Dm,

101 Job::MentionEmails { .. } => Self::Email,

102 Job::EveryoneStart { .. } | Job::EveryoneChunk { .. } => Self::Bulk,

103 }

104 }

105

106 fn name(self) -> &'static str {

107 match self {

108 Self::Dm => "dm",

109 Self::Email => "email",

110 Self::Bulk => "bulk",

111 }

112 }

113}

114

88115struct Fanout {

89 interactive: mpsc::Sender<Job>,

116 dm: mpsc::Sender<Job>,

117 email: mpsc::Sender<Job>,

90118 bulk: mpsc::Sender<Job>,

91119}

92120

@@ -94,19 +122,29 @@

94122

95123/// 启动固定 worker 池；必须在 tokio 运行时内调用一次。

96124pub fn start(service: ChatService, cancel: CancellationToken) {

97 let (interactive_tx, interactive_rx) = mpsc::channel(INTERACTIVE_CAPACITY);

125 let (dm_tx, dm_rx) = mpsc::channel(DM_CAPACITY);

126 let (email_tx, email_rx) = mpsc::channel(EMAIL_CAPACITY);

98127 let (bulk_tx, bulk_rx) = mpsc::channel(BULK_CAPACITY);

99 let interactive_rx = Arc::new(Mutex::new(interactive_rx));

128 let dm_rx = Arc::new(Mutex::new(dm_rx));

129 let email_rx = Arc::new(Mutex::new(email_rx));

100130 let bulk_rx = Arc::new(Mutex::new(bulk_rx));

101131

102 for _ in 0..INTERACTIVE_WORKERS {

132 for _ in 0..DM_WORKERS {

103133 tokio::spawn(worker_loop(

104134 service.clone(),

105 interactive_rx.clone(),

135 dm_rx.clone(),

106136 bulk_tx.clone(),

107137 cancel.clone(),

108138 ));

109139 }

140 for _ in 0..EMAIL_WORKERS {

141 tokio::spawn(worker_loop(

142 service.clone(),

143 email_rx.clone(),

144 bulk_tx.clone(),

145 cancel.clone(),

146 ));

147 }

110148 for _ in 0..BULK_WORKERS {

111149 tokio::spawn(worker_loop(

112150 service.clone(),

@@ -118,7 +156,8 @@

118156

119157 if FANOUT

120158 .set(Fanout {

121 interactive: interactive_tx,

159 dm: dm_tx,

160 email: email_tx,

122161 bulk: bulk_tx,

123162 })

124163 .is_err()

@@ -126,41 +165,56 @@

126165 tracing::warn!("chat fanout already started");

127166 } else {

128167 tracing::info!(

129 interactive = INTERACTIVE_WORKERS,

168 dm = DM_WORKERS,

169 email = EMAIL_WORKERS,

130170 bulk = BULK_WORKERS,

131171 "chat fanout pool started"

132172 );

133173 }

134174}

135175

136/// 投递一个扇出任务；队列满或未启动时丢弃并打点，不阻塞调用方。

176/// 投递一个扇出任务；队列满时转后台等待投递，不阻塞调用方，只有 channel

177/// 关闭（进程退出）或未启动时才会丢弃。

137178pub(crate) fn submit(job: Job) {

179 let lane = Lane::of(&job);

138180 let Some(fanout) = FANOUT.get() else {

139 tracing::warn!(kind = job.kind(), "chat fanout not started; dropping job");

140 oltp::metrics::chat_fanout("unstarted");

181 tracing::warn!(

182 kind = job.kind(),

183 lane = lane.name(),

184 "chat fanout not started; dropping job"

185 );

186 oltp::metrics::chat_fanout(lane.name(), "unstarted");

141187 return;

142188 };

143 let (tx, lane) = match &job {

144 Job::EveryoneStart { .. } | Job::EveryoneChunk { .. } => (&fanout.bulk, "bulk"),

145 _ => (&fanout.interactive, "interactive"),

189 let tx = match lane {

190 Lane::Dm => &fanout.dm,

191 Lane::Email => &fanout.email,

192 Lane::Bulk => &fanout.bulk,

146193 };

147194 match tx.try_send(job) {

148 Ok(()) => oltp::metrics::chat_fanout("queued"),

195 Ok(()) => oltp::metrics::chat_fanout(lane.name(), "queued"),

149196 Err(mpsc::error::TrySendError::Full(job)) => {

150197 tracing::warn!(

151198 kind = job.kind(),

152 lane,

153 "chat fanout queue full; dropping job"

199 lane = lane.name(),

200 "chat fanout queue full; deferring job"

154201 );

155 oltp::metrics::chat_fanout("full");

202 oltp::metrics::chat_fanout(lane.name(), "deferred");

203 let tx = tx.clone();

204 tokio::spawn(async move {

205 if tx.send(job).await.is_err() {

206 tracing::warn!(lane = lane.name(), "chat fanout closed while deferring job");

207 oltp::metrics::chat_fanout(lane.name(), "closed");

208 }

209 });

156210 }

157211 Err(mpsc::error::TrySendError::Closed(job)) => {

158212 tracing::warn!(

159213 kind = job.kind(),

160 lane,

214 lane = lane.name(),

161215 "chat fanout queue closed; dropping job"

162216 );

163 oltp::metrics::chat_fanout("closed");

217 oltp::metrics::chat_fanout(lane.name(), "closed");

164218 }

165219 }

166220}

@@ -187,6 +241,7 @@

187241}

188242

189243async fn run(service: &ChatService, job: Job, bulk: &mpsc::Sender<Job>) {

244 let lane = Lane::of(&job).name();

190245 let request_id = job.request_id().map(str::to_owned);

191246 let trace = job.trace().map(str::to_owned);

192247 let span = tracing::info_span!(

@@ -203,7 +258,7 @@

203258 run_job(service, job, bulk).instrument(span),

204259 )

205260 .await;

206 oltp::metrics::chat_fanout("done");

261 oltp::metrics::chat_fanout(lane, "done");

207262}

208263

209264async fn run_job(service: &ChatService, job: Job, bulk: &mpsc::Sender<Job>) {