fix(events): harden outbox delivery and envelope metadata

Modified lib/events/src/emit.rs +1 −0 View lib/events/src/emit.rs

@@ -54,6 +54,7 @@

5454 }

5555 if let Some(notification) = payload.get_mut("notification")

5656 && notification.is_object()

57 && notification.get("actor").is_none_or(Value::is_null)

5758 {

5859 notification["actor"] = snapshot;

5960 }

Modified lib/events/src/handler.rs +18 −68 View lib/events/src/handler.rs

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

1414 "notify",

1515 "organization",

1616 "pull",

17 "pull_request",

1817];

1918

2019#[derive(Clone, Debug)]

@@ -41,28 +40,11 @@

4140 }

4241 }

4342

44 pub fn publisher(&self) -> &Publisher {

45 &self.publisher

46 }

47

48 pub fn max_payload(&self) -> usize {

49 self.publisher.max_payload()

50 }

51

5243 pub fn envelope<T>(&self, kind: &str, payload: T) -> Envelope<T> {

5344 Envelope::new(kind, payload)

5445 }

5546

5647 #[tracing::instrument(skip_all, fields(kind = kind, request_id = %request_id.unwrap_or_default()))]

57 pub async fn send<T>(&self, kind: &str, payload: T, request_id: Option<&str>) -> QueueResult<()>

58 where

59 T: Serialize,

60 {

61 self.send_labeled(kind, Labels::default(), payload, request_id)

62 .await

63 }

64

65 #[tracing::instrument(skip_all, fields(kind = kind, request_id = %request_id.unwrap_or_default()))]

6648 pub async fn send_labeled<T>(

6749 &self,

6850 kind: &str,

@@ -89,10 +71,20 @@

8971 where

9072 T: Serialize,

9173 {

92 let mut envelope = self.envelope(kind, payload).with_labels(labels);

74 let mut envelope = self

75 .envelope(kind, payload)

76 .with_source(domain_of(kind))

77 .with_labels(labels);

9378 if let Some(key) = idempotency_key {

9479 envelope = envelope.with_idempotency_key(key);

9580 }

81 if let Some(trace_id) = envelope::context::request_id().or_else(|| {

82 request_id

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

84 .map(str::to_owned)

85 }) {

86 envelope = envelope.with_trace_id(trace_id);

87 }

9688 self.send_envelope(&envelope, request_id).await

9789 }

9890

@@ -121,11 +113,6 @@

121113 self.dispatch(envelope, request_id).await

122114 }

123115

124 pub async fn health(&self) -> QueueResult<()> {

125 let topic = self.publisher.topic(topic::HEALTH);

126 self.publisher.conn().health(&topic).await

127 }

128

129116 pub async fn shutdown(&self) -> QueueResult<()> {

130117 self.publisher.conn().shutdown().await;

131118 Ok(())

@@ -141,7 +128,11 @@

141128 let payload = serde_json::to_value(&envelope.payload).map_err(QueueError::from)?;

142129 let area = topic::of(&envelope.kind);

143130 let tag = label::tag(&envelope.kind, Some(&payload));

144 let request_id = envelope::context::request_id().or_else(|| request_id.map(str::to_owned));

131 let request_id = envelope::context::request_id().or_else(|| {

132 request_id

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

134 .map(str::to_owned)

135 });

145136 let properties = envelope.properties(Some(&payload), request_id.as_deref());

146137 let topic = self.publisher.topic(area);

147138 let body = envelope.encode()?;

@@ -165,47 +156,6 @@

165156 Ok(())

166157}

167158

168const PUBLISH_ATTEMPTS: usize = 3;

169const PUBLISH_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);

170const PUBLISH_BACKOFF: std::time::Duration = std::time::Duration::from_millis(50);

171

172/// 统一的事件转发重试：单次尝试带超时，失败按退避重试，

173/// 最终仍失败打 ERROR，便于告警发现事件链断了。

174/// `context` 仅用于日志文案（如 "auth" / "email"）。

175#[tracing::instrument(skip_all, fields(kind = kind, context = context, request_id = %_request_id.unwrap_or_default()))]

176pub async fn retry_publish<T>(

177 events: &EventHandler,

178 kind: &str,

179 labels: Labels,

180 payload: T,

181 context: &str,

182 _request_id: Option<&str>,

183) where

184 T: Serialize + Clone,

185{

186 for attempt in 1..=PUBLISH_ATTEMPTS {

187 let published = events.send_labeled(kind, labels.clone(), payload.clone(), _request_id);

188 match tokio::time::timeout(PUBLISH_TIMEOUT, published).await {

189 Ok(Ok(())) => {

190 if attempt > 1 {

191 tracing::info!(kind, attempt, "{context} event forwarded after retry");

192 }

193 return;

194 }

195 Ok(Err(error)) => {

196 tracing::warn!(kind, attempt, %error, "failed to publish {context} event");

197 }

198 Err(_) => {

199 tracing::warn!(kind, attempt, "publishing {context} event timed out");

200 }

201 }

202 if attempt < PUBLISH_ATTEMPTS {

203 tokio::time::sleep(PUBLISH_BACKOFF * attempt as u32).await;

204 }

205 }

206 tracing::error!(

207 kind,

208 attempts = PUBLISH_ATTEMPTS,

209 "{context} event could not be forwarded to rocketmq"

210 );

159fn domain_of(kind: &str) -> &str {

160 kind.split('.').next().unwrap_or_default()

211161}

Modified lib/events/src/label.rs +2 −9 View lib/events/src/label.rs

@@ -47,7 +47,6 @@

4747pub const ADD_SSH_KEY: &str = "add-sshkey";

4848pub const DELETE_SSH_KEY: &str = "delete-sshkey";

4949pub const CREATE_ACCESS_KEY: &str = "create-accesskey";

50pub const ROTATE_ACCESS_KEY: &str = "rotate-accesskey";

5150pub const DELETE_ACCESS_KEY: &str = "delete-accesskey";

5251pub const USE_ACCESS_KEY: &str = "use-accesskey";

5352pub const CHALLENGE: &str = "challenge";

@@ -104,13 +103,7 @@

104103 ("auth", "password", "update") => RESET_PASSWORD,

105104 ("auth", "ssh_key", "create") => ADD_SSH_KEY,

106105 ("auth", "ssh_key", "delete") => DELETE_SSH_KEY,

107 ("auth", "access_key", "create") => {

108 if has_field(payload, "rotated") {

109 ROTATE_ACCESS_KEY

110 } else {

111 CREATE_ACCESS_KEY

112 }

113 }

106 ("auth", "access_key", "create") => CREATE_ACCESS_KEY,

114107 ("auth", "access_key", "delete") => DELETE_ACCESS_KEY,

115108 ("auth", "access_key", "used") => USE_ACCESS_KEY,

116109 ("auth", "challenge", "create") => CHALLENGE,

@@ -156,5 +149,5 @@

156149}

157150

158151fn has_field(payload: Option<&Value>, key: &str) -> bool {

159 payload.is_some_and(|value| value.get(key).is_some())

152 payload.is_some_and(|value| value.get(key).is_some_and(|value| !value.is_null()))

160153}

Modified lib/events/src/lib.rs +2 −1 View lib/events/src/lib.rs

@@ -5,5 +5,6 @@

55pub mod outbox;

66pub mod topic;

77

8pub use handler::{EventHandler, retry_publish};

8pub use handler::EventHandler;

9pub use outbox::emit_retry;

910pub use queue::{Envelope, Labels};

Modified lib/events/src/outbox.rs +139 −150 View lib/events/src/outbox.rs

@@ -4,14 +4,11 @@

44use chrono::{DateTime, Utc};

55use futures_util::future;

66use models::event_outbox;

7use opentelemetry::metrics::Gauge;

7use opentelemetry::metrics::{Counter, Gauge};

88use opentelemetry::{KeyValue, global};

9use queue::{DeliveryClass, EVENT_PREFIX, Envelope, Publisher, SCHEMA_VERSION};

9use queue::{DeliveryClass, EVENT_PREFIX, Envelope, Publisher, QueueError, SCHEMA_VERSION};

1010use sea_orm::ActiveValue::Set;

11use sea_orm::{

12 ColumnTrait, ConnectionTrait, DatabaseConnection, DatabaseTransaction, DbBackend, DbErr,

13 EntityTrait, QueryFilter, Statement,

14};

11use sea_orm::{ConnectionTrait, DatabaseConnection, DbBackend, DbErr, EntityTrait, Statement};

1512use serde::Serialize;

1613use serde_json::Value;

1714use tokio::task::{JoinHandle, JoinSet};

@@ -26,7 +23,6 @@

2623const PUBLISH_TIMEOUT: Duration = Duration::from_secs(10);

2724const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(30);

2825const POLL_INTERVAL: Duration = Duration::from_millis(500);

29const MAX_ENVELOPE_BYTES: usize = queue::DEFAULT_MAX_MESSAGE_BYTES - 4096;

3026const BACKLOG_SAMPLE_INTERVAL: Duration = Duration::from_secs(10);

3127/// 单条事件的重试上限。超过后丢弃并打 ERROR：否则一条毒消息会永久堵住它所在的

3228/// ordering_key（`auth:{user_id}` 之类），后面的事件永远发不出去。

@@ -44,6 +40,34 @@

4440 .build()

4541});

4642

43static PUBLISHED: LazyLock<Counter<u64>> = LazyLock::new(|| {

44 global::meter("events")

45 .u64_counter("event.outbox.published")

46 .build()

47});

48static RETRIED: LazyLock<Counter<u64>> = LazyLock::new(|| {

49 global::meter("events")

50 .u64_counter("event.outbox.retried")

51 .build()

52});

53static DISCARDED: LazyLock<Counter<u64>> = LazyLock::new(|| {

54 global::meter("events")

55 .u64_counter("event.outbox.discarded")

56 .build()

57});

58static LEASE_LOST: LazyLock<Counter<u64>> = LazyLock::new(|| {

59 global::meter("events")

60 .u64_counter("event.outbox.lease_lost")

61 .build()

62});

63static ENQUEUE_FAILED: LazyLock<Counter<u64>> = LazyLock::new(|| {

64 global::meter("events")

65 .u64_counter("event.outbox.enqueue_failed")

66 .build()

67});

68

69const ENQUEUE_ATTEMPTS: usize = 3;

70

4771const CLAIM: &str = "

4872WITH pending AS (

4973 SELECT candidate.id FROM event_outbox AS candidate

@@ -122,9 +146,6 @@

122146 .get("pull_id")

123147 .or_else(|| labels.get(label::PULL_REQUEST_ID))

124148 .or_else(|| labels.get(label::REPOSITORY_ID)),

125 "pull_request" => labels

126 .get(label::PULL_REQUEST_ID)

127 .or_else(|| labels.get(label::REPOSITORY_ID)),

128149 "issue" => labels

129150 .get(label::ISSUE_ID)

130151 .or_else(|| labels.get(label::BOARD_ID))

@@ -150,27 +171,6 @@

150171 "event `{kind}` requires a user, organization or request id label"

151172 ))

152173 })

153}

154

155#[tracing::instrument(skip_all, fields(kind, request_id = %_request_id.unwrap_or_default()))]

156pub async fn enqueue<T: Serialize>(

157 txn: &DatabaseTransaction,

158 kind: &str,

159 labels: Labels,

160 payload: T,

161 trace_id: &str,

162 _request_id: Option<&str>,

163) -> Result<Uuid, DbErr> {

164 enqueue_at(

165 txn,

166 kind,

167 labels,

168 payload,

169 trace_id,

170 Utc::now(),

171 _request_id,

172 )

173 .await

174174}

175175

176176#[tracing::instrument(skip_all, fields(kind, request_id = %_request_id.unwrap_or_default()))]

@@ -215,6 +215,30 @@

215215 .await

216216}

217217

218#[tracing::instrument(skip_all, fields(kind))]

219pub async fn emit_retry<T: Serialize>(

220 conn: &impl ConnectionTrait,

221 kind: &str,

222 labels: Labels,

223 payload: &T,

224) -> Result<Uuid, DbErr> {

225 let mut attempt = 1_usize;

226 loop {

227 match crate::emit::emit(conn, kind, labels.clone(), payload).await {

228 Ok(id) => return Ok(id),

229 Err(error) if attempt >= ENQUEUE_ATTEMPTS => {

230 ENQUEUE_FAILED.add(1, &[KeyValue::new("domain", domain(kind).to_owned())]);

231 return Err(error);

232 }

233 Err(error) => {

234 tracing::warn!(kind, attempt, %error, "retrying the outbox enqueue");

235 tokio::time::sleep(Duration::from_millis(50) * attempt as u32).await;

236 attempt += 1;

237 }

238 }

239 }

240}

241

218242fn build_model<T: Serialize>(

219243 kind: &str,

220244 labels: Labels,

@@ -236,11 +260,13 @@

236260 if !trace_id.is_empty() {

237261 envelope = envelope.with_trace_id(trace_id);

238262 }

239 let subject = format!("{EVENT_PREFIX}{}", envelope.kind);

263 let subject = envelope

264 .bound_subject(EVENT_PREFIX)

265 .map_err(|error| DbErr::Custom(format!("invalid outbox event kind: {error}")))?;

240266 let encoded = envelope

241267 .encode()

242268 .map_err(|error| DbErr::Custom(format!("encoding outbox event: {error}")))?;

243 if encoded.len() > MAX_ENVELOPE_BYTES {

269 if encoded.len() > queue::MAX_ENVELOPE_BYTES {

244270 return Err(DbErr::Custom(

245271 "outbox event exceeds the size limit".to_owned(),

246272 ));

@@ -260,87 +286,6 @@

260286 Ok((envelope.id, model))

261287}

262288

263fn chat_labels(room_id: Uuid, request_id: Option<&str>) -> Labels {

264 let mut labels = Labels::new()

265 .with(label::ROOM_ID, room_id)

266 .with(label::ORDERING_KEY, format!("chat:room:{room_id}"));

267 if let Some(request_id) = request_id {

268 labels.insert(label::REQUEST_ID, request_id);

269 }

270 labels

271}

272

273pub async fn enqueue_room<T: Serialize>(

274 conn: &impl ConnectionTrait,

275 kind: &str,

276 room_id: Uuid,

277 payload: T,

278 request_id: Option<&str>,

279) -> Result<Uuid, DbErr> {

280 enqueue_at(

281 conn,

282 kind,

283 chat_labels(room_id, request_id),

284 payload,

285 request_id.unwrap_or_default(),

286 Utc::now(),

287 request_id,

288 )

289 .await

290}

291

292pub async fn enqueue_chat_users<T: Serialize>(

293 conn: &impl ConnectionTrait,

294 kind: &str,

295 uids: &[i64],

296 room_id: Uuid,

297 payload: &T,

298 request_id: Option<&str>,

299) -> Result<(), DbErr> {

300 if uids.is_empty() {

301 return Ok(());

302 }

303 if !is_durable_kind(kind) {

304 return Err(DbErr::Custom(format!(

305 "event `{kind}` requires an outbox-enabled business domain"

306 )));

307 }

308 let mut models = Vec::with_capacity(uids.len());

309 for uid in uids {

310 let labels = chat_labels(room_id, request_id)

311 .with(label::USER_ID, *uid)

312 .with(label::ORDERING_KEY, format!("chat:user:{uid}"));

313 let (_, model) = build_model(

314 kind,

315 labels,

316 payload,

317 request_id.unwrap_or_default(),

318 Utc::now(),

319 )?;

320 models.push(model);

321 }

322 event_outbox::Entity::insert_many(models).exec(conn).await?;

323 Ok(())

324}

325

326pub async fn release(db: &DatabaseConnection, ids: &[Uuid]) -> Result<(), DbErr> {

327 if ids.is_empty() {

328 return Ok(());

329 }

330 event_outbox::Entity::update_many()

331 .set(event_outbox::ActiveModel {

332 available_at: Set(Utc::now()),

333 lease_id: Set(None),

334 leased_until: Set(None),

335 ..Default::default()

336 })

337 .filter(event_outbox::Column::Id.is_in(ids.to_vec()))

338 .filter(event_outbox::Column::PublishedAt.is_null())

339 .exec(db)

340 .await?;

341 Ok(())

342}

343

344289struct PendingEvent {

345290 id: Uuid,

346291 subject: String,

@@ -356,8 +301,9 @@

356301async fn publish_identity(

357302 handler: Option<&EventHandler>,

358303 row: &PendingEvent,

359) -> Result<(), String> {

360 let handler = handler.ok_or_else(|| "identity event handler is not configured".to_owned())?;

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

305 let handler =

306 handler.ok_or_else(|| QueueError::publish("identity event handler is not configured"))?;

361307 let request_id = row

362308 .envelope

363309 .get("labels")

@@ -369,29 +315,32 @@

369315 })

370316 .and_then(Value::as_str)

371317 .unwrap_or_default();

372 let envelope: Envelope<Value> = serde_json::from_value(row.envelope.clone())

373 .map_err(|error| format!("invalid outbox envelope: {error}"))?;

318 let mut envelope: Envelope<Value> = serde_json::from_value(row.envelope.clone())?;

374319 if envelope.id != row.id || envelope.schema_version != SCHEMA_VERSION {

375 return Err("outbox event identity or schema is invalid".to_owned());

320 return Err(QueueError::payload(

321 "outbox event identity or schema is invalid",

322 ));

376323 }

324 envelope.attempt = row.attempts.max(1) as u32;

377325 handler

378326 .publish_durable(&row.subject, &envelope, Some(request_id))

379327 .await

380 .map_err(|error| error.to_string())

381328}

382329

383async fn publish_app(publisher: Option<&Publisher>, row: &PendingEvent) -> Result<(), String> {

384 let publisher = publisher.ok_or_else(|| "app event publisher is not configured".to_owned())?;

385 let envelope: Envelope<Value> = serde_json::from_value(row.envelope.clone())

386 .map_err(|error| format!("invalid outbox envelope: {error}"))?;

330async fn publish_app(publisher: Option<&Publisher>, row: &PendingEvent) -> Result<(), QueueError> {

331 let publisher =

332 publisher.ok_or_else(|| QueueError::publish("app event publisher is not configured"))?;

333 let mut envelope: Envelope<Value> = serde_json::from_value(row.envelope.clone())?;

387334 if envelope.id != row.id

388335 || envelope.kind != row.kind

389336 || envelope.schema_version != SCHEMA_VERSION

390337 {

391 return Err("outbox event identity is invalid".to_owned());

338 return Err(QueueError::payload("outbox event identity is invalid"));

392339 }

340 envelope.verify_event_subject(&row.subject)?;

341 envelope.attempt = row.attempts.max(1) as u32;

393342 let topic = publisher.topic(topic::of(&envelope.kind));

394 let body = envelope.encode().map_err(|error| error.to_string())?;

343 let body = envelope.encode()?;

395344 let payload = serde_json::to_value(&envelope.payload).ok();

396345 let tag = label::tag(&envelope.kind, payload.as_ref());

397346 publisher

@@ -403,9 +352,14 @@

403352 body,

404353 )

405354 .await

406 .map_err(|error| error.to_string())

407355}

408356

357enum PublishOutcome {

358 Published,

359 Retried,

360 Discarded(&'static str),

361}

362

409363async fn publish(

410364 db: &DatabaseConnection,

411365 identity: Option<&EventHandler>,

@@ -445,48 +399,83 @@

445399 .instrument(span);

446400 let result = tokio::time::timeout(PUBLISH_TIMEOUT, result)

447401 .await

448 .unwrap_or_else(|_| Err("outbox publish timed out".to_owned()));

449 let statement = match result {

450 Ok(()) => Statement::from_sql_and_values(

451 DbBackend::Postgres,

452 "UPDATE event_outbox SET published_at = CURRENT_TIMESTAMP, lease_id = NULL, leased_until = NULL, last_error = NULL WHERE id = $1 AND lease_id = $2 AND published_at IS NULL",

453 [row.id.into(), lease_id.into()],

402 .unwrap_or_else(|_| Err(QueueError::publish("outbox publish timed out")));

403 let (statement, outcome) = match result {

404 Ok(()) => (

405 Statement::from_sql_and_values(

406 DbBackend::Postgres,

407 "UPDATE event_outbox SET published_at = CURRENT_TIMESTAMP, lease_id = NULL, leased_until = NULL, last_error = NULL WHERE id = $1 AND lease_id = $2 AND published_at IS NULL",

408 [row.id.into(), lease_id.into()],

409 ),

410 PublishOutcome::Published,

454411 ),

455412 Err(error) => {

456 let error_text: String = error.chars().take(2048).collect();

457 if row.attempts >= MAX_PUBLISH_ATTEMPTS {

413 let permanent = !error.is_retryable();

414 let error_text: String = error.to_string().chars().take(2048).collect();

415 if permanent || row.attempts >= MAX_PUBLISH_ATTEMPTS {

458416 tracing::error!(

459417 event_id = %row.id,

460418 subject = %row.subject,

461419 kind = %row.kind,

462420 attempt = row.attempts,

421 permanent,

463422 error = %error_text,

464 "outbox event exceeded the retry budget and was discarded"

423 "outbox event was discarded"

465424 );

466 Statement::from_sql_and_values(

467 DbBackend::Postgres,

468 "UPDATE event_outbox SET published_at = CURRENT_TIMESTAMP, lease_id = NULL, leased_until = NULL, last_error = $3 WHERE id = $1 AND lease_id = $2 AND published_at IS NULL",

469 [row.id.into(), lease_id.into(), error_text.into()],

425 (

426 Statement::from_sql_and_values(

427 DbBackend::Postgres,

428 "UPDATE event_outbox SET published_at = CURRENT_TIMESTAMP, lease_id = NULL, leased_until = NULL, last_error = $3 WHERE id = $1 AND lease_id = $2 AND published_at IS NULL",

429 [row.id.into(), lease_id.into(), error_text.into()],

430 ),

431 PublishOutcome::Discarded(if permanent {

432 "permanent"

433 } else {

434 "retry_budget"

435 }),

470436 )

471437 } else {

472438 tracing::warn!(event_id = %row.id, subject = %row.subject, kind = %row.kind, attempt = row.attempts, error = %error_text, "outbox publish will retry");

473439 let delay = retry_delay(row.attempts) + f64::from(row.id.as_bytes()[15] % 5);

474 Statement::from_sql_and_values(

475 DbBackend::Postgres,

476 "UPDATE event_outbox SET available_at = CURRENT_TIMESTAMP + ($3 * INTERVAL '1 second'), lease_id = NULL, leased_until = NULL, last_error = $4 WHERE id = $1 AND lease_id = $2 AND published_at IS NULL",

477 [

478 row.id.into(),

479 lease_id.into(),

480 delay.into(),

481 error_text.into(),

482 ],

440 (

441 Statement::from_sql_and_values(

442 DbBackend::Postgres,

443 "UPDATE event_outbox SET available_at = CURRENT_TIMESTAMP + ($3 * INTERVAL '1 second'), lease_id = NULL, leased_until = NULL, last_error = $4 WHERE id = $1 AND lease_id = $2 AND published_at IS NULL",

444 [

445 row.id.into(),

446 lease_id.into(),

447 delay.into(),

448 error_text.into(),

449 ],

450 ),

451 PublishOutcome::Retried,

483452 )

484453 }

485454 }

486455 };

456 let event_domain = domain(&row.kind).to_owned();

487457 let updated = db.execute_raw(statement).await?;

488458 if updated.rows_affected() == 0 {

459 LEASE_LOST.add(1, &[KeyValue::new("domain", event_domain)]);

489460 tracing::warn!(event_id = %row.id, "outbox lease was reclaimed by another relay");

461 } else {

462 match outcome {

463 PublishOutcome::Published => {

464 PUBLISHED.add(1, &[KeyValue::new("domain", event_domain)]);

465 }

466 PublishOutcome::Retried => {

467 RETRIED.add(1, &[KeyValue::new("domain", event_domain)]);

468 }

469 PublishOutcome::Discarded(reason) => {

470 DISCARDED.add(

471 1,

472 &[

473 KeyValue::new("domain", event_domain),

474 KeyValue::new("reason", reason),

475 ],

476 );

477 }

478 }

490479 }

491480 Ok(())

492481}

@@ -532,7 +521,7 @@

532521async fn purge_published(db: &DatabaseConnection) {

533522 match db

534523 .execute_unprepared(&format!(

535 "DELETE FROM event_outbox WHERE published_at < CURRENT_TIMESTAMP - INTERVAL '{PURGED_RETENTION_SECS} seconds'"

524 "DELETE FROM event_outbox WHERE id IN (SELECT id FROM event_outbox WHERE published_at < CURRENT_TIMESTAMP - INTERVAL '{PURGED_RETENTION_SECS} seconds' LIMIT 10000)"

536525 ))

537526 .await

538527 {

Modified lib/events/src/topic.rs +2 −3 View lib/events/src/topic.rs

@@ -5,7 +5,6 @@

55pub const VERIFICATION: &str = "verification";

66pub const ORGANIZATION: &str = "org";

77pub const EMAIL: &str = "email";

8pub const HEALTH: &str = "login";

98

109pub const AGENT: &str = "agent";

1110pub const CHAT: &str = "chat";

@@ -14,7 +13,7 @@

1413pub const ISSUE: &str = "issue";

1514pub const NOTIFY: &str = "notify";

1615pub const OBJECT: &str = "object";

17pub const PULL_REQUEST: &str = "pull_request";

16pub const PULL: &str = "pull";

1817pub const REALTIME: &str = "realtime";

1918pub const USER: &str = "user";

2019pub const WEBHOOK: &str = "webhook";

@@ -38,7 +37,7 @@

3837 ("issue", _) => ISSUE,

3938 ("notify", _) => NOTIFY,

4039 ("object", _) => OBJECT,

41 ("pull_request", _) => PULL_REQUEST,

40 ("pull", _) => PULL,

4241 ("realtime", _) => REALTIME,

4342 ("user", _) => USER,

4443 ("webhook", _) => WEBHOOK,

Modified lib/queue/src/conn.rs +15 −1 View lib/queue/src/conn.rs

@@ -8,6 +8,8 @@

88use crate::error::{QueueError, QueueResult};

99

1010pub const DEFAULT_MAX_MESSAGE_BYTES: usize = 4 * 1024 * 1024;

11/// 事件信封的字节预算：为消息属性与协议头预留 4 KiB。

12pub const MAX_ENVELOPE_BYTES: usize = DEFAULT_MAX_MESSAGE_BYTES - 4096;

1113

1214#[derive(Clone)]

1315pub struct RocketMqConn {

@@ -29,6 +31,12 @@

2931impl RocketMqConn {

3032 pub async fn init(config: &RocketMqConfig) -> QueueResult<Self> {

3133 config.validate().map_err(QueueError::config)?;

34 if config.max_message_bytes < MAX_ENVELOPE_BYTES {

35 return Err(QueueError::config(format!(

36 "rocketmq.max_message_bytes ({}) is smaller than the event envelope budget ({MAX_ENVELOPE_BYTES})",

37 config.max_message_bytes

38 )));

39 }

3240 probe_nameservers(config).await?;

3341

3442 let mut producer = DefaultMQProducer::builder()

@@ -118,7 +126,13 @@

118126/// dead nameserver must fail startup instead of silently running without an

119127/// event channel.

120128async fn probe_nameservers(config: &RocketMqConfig) -> QueueResult<()> {

121 let timeout = std::time::Duration::from_secs(3);

129 probe_nameservers_with(config, std::time::Duration::from_secs(3)).await

130}

131

132pub async fn probe_nameservers_with(

133 config: &RocketMqConfig,

134 timeout: std::time::Duration,

135) -> QueueResult<()> {

122136 for server in config.name_servers() {

123137 match tokio::time::timeout(timeout, tokio::net::TcpStream::connect(server)).await {

124138 Ok(Ok(_)) => return Ok(()),

Modified lib/queue/src/envelope.rs +30 −2 View lib/queue/src/envelope.rs

@@ -13,6 +13,8 @@

1313pub const PROPERTY_USER: &str = "user";

1414pub const PROPERTY_ORG: &str = "org";

1515pub const PROPERTY_REQUEST_ID: &str = "request_id";

16pub const PROPERTY_ACTOR: &str = "actor";

17pub const PROPERTY_RECIPIENT: &str = "recipient";

1618pub const SCHEMA_VERSION: u16 = 1;

1719

1820pub const EVENT_PREFIX: &str = "events.";

@@ -191,7 +193,7 @@

191193 payload: Option<&Value>,

192194 request_id: Option<&str>,

193195 ) -> Vec<(String, String)> {

194 let mut properties = Vec::with_capacity(3);

196 let mut properties = Vec::with_capacity(5);

195197

196198 let user = self

197199 .labels

@@ -218,11 +220,37 @@

218220 .get("request_id")

219221 .map(str::to_owned)

220222 .or_else(|| payload.and_then(|value| field_text(value, "request_id")))

221 .or_else(|| request_id.map(str::to_owned));

223 .or_else(|| request_id.map(str::to_owned))

224 .filter(|request_id| !request_id.is_empty());

222225 if let Some(request_id) = request_id {

223226 properties.push((PROPERTY_REQUEST_ID.to_string(), request_id));

224227 }

225228

229 let actor = self

230 .labels

231 .get("actor_user_id")

232 .map(str::to_owned)

233 .or_else(|| payload.and_then(|value| field_text(value, "actor_id")))

234 .or_else(|| {

235 payload

236 .and_then(|value| value.get("actor"))

237 .and_then(|actor| field_text(actor, "id"))

238 })

239 .filter(|actor| !actor.is_empty());

240 if let Some(actor) = actor {

241 properties.push((PROPERTY_ACTOR.to_string(), actor));

242 }

243

244 let recipient = self

245 .labels

246 .get("recipient_user_id")

247 .map(str::to_owned)

248 .or_else(|| payload.and_then(|value| field_text(value, "recipient_id")))

249 .filter(|recipient| !recipient.is_empty());

250 if let Some(recipient) = recipient {

251 properties.push((PROPERTY_RECIPIENT.to_string(), recipient));

252 }

253

226254 properties

227255 }

228256}

Modified lib/queue/src/lib.rs +3 −3 View lib/queue/src/lib.rs

@@ -5,12 +5,12 @@

55pub mod envelope;

66pub mod error;

77pub mod publisher;

8pub mod route;

98pub mod subject;

109

11pub use conn::{DEFAULT_MAX_MESSAGE_BYTES, RocketMqConn};

10pub use conn::{

11 DEFAULT_MAX_MESSAGE_BYTES, MAX_ENVELOPE_BYTES, RocketMqConn, probe_nameservers_with,

12};

1213pub use consumer::{Consumer, ConsumerConfig, MessageHandler};

1314pub use envelope::{DeliveryClass, EVENT_PREFIX, Envelope, Labels, SCHEMA_VERSION};

1415pub use error::{QueueError, QueueResult};

1516pub use publisher::Publisher;

16pub use route::Route;

Deleted lib/queue/src/route.rs +0 −23

@@ -1,23 +0,0 @@

1use crate::error::{QueueError, QueueResult};

2

3pub struct Route {

4 pub area: String,

5 pub tag: String,

6}

7

8pub fn route(kind: &str) -> QueueResult<Route> {

9 let mut parts = kind.split('.');

10 let domain = parts.next().unwrap_or_default();

11 let name = parts.next().unwrap_or_default();

12 let operate = parts.next().unwrap_or_default();

13 if domain.is_empty() || name.is_empty() || operate.is_empty() || parts.next().is_some() {

14 return Err(QueueError::payload(format!(

15 "event kind `{kind}` must use `domain.name.operate`"

16 )));

17 }

18

19 Ok(Route {

20 area: domain.to_string(),

21 tag: format!("{name}-{operate}"),

22 })

23}