Skip to content
zhenyi

gitdata-core

0
Clone

HTTPS

SSH

SSH needs a key in your settings.

Download ZIP

fix(events): harden socket delivery and shutdown handling

- Bound push backlog with XADD MAXLEN instead of rejecting full streams
- Cache subscription authorization to reduce repeated checks
- Drain workers and pending results during shutdown with a timeout
- Deliver org events even when membership is empty
- Handle full completion queues and close users after revocation delivery
- Use shared session cookie config and remove unused subscription token
ZhenYi committed
parent 98e06cb commit 072c704 Browse files

9 files changed

+58 −29

File tree 9
Modified lib/infra/actor/src/push_store.rs View lib/infra/actor/src/push_store.rs
@@ -73,7 +73,6 @@
7373 command.arg(r#"
7474local existing = redis.call('GET', KEYS[3])
7575if existing then return existing end
76if redis.call('XLEN', KEYS[1]) >= tonumber(ARGV[3]) then return redis.error_reply('push backlog is full') end
7776local prior = redis.call('HGET', KEYS[4], ARGV[5])
7877local resync = ARGV[4] == 'unknown' or (ARGV[4] ~= '' and prior and ARGV[4] < prior) or false
7978if ARGV[4] ~= '' and ARGV[4] ~= 'unknown' and (not prior or ARGV[4] > prior) then
@@ -82,7 +81,7 @@
8281redis.call('INCR', KEYS[2])
8382local sequence = redis.call('GET', KEYS[2])
8483local delivery = sequence .. '-0'
85redis.call('XADD', KEYS[1], delivery, 'payload', ARGV[1], 'sequence', sequence, 'event_id', ARGV[2], 'resync_required', resync and '1' or '0')
84redis.call('XADD', KEYS[1], 'MAXLEN', '~', ARGV[3], delivery, 'payload', ARGV[1], 'sequence', sequence, 'event_id', ARGV[2], 'resync_required', resync and '1' or '0')
8685redis.call('SET', KEYS[3], delivery, 'EX', ARGV[6])
8786return delivery
8887"#).arg(4).arg(self.key(user_id, "events")).arg(self.key(user_id, "sequence"))
Modified lib/infra/socket/src/options.rs View lib/infra/socket/src/options.rs
@@ -6,7 +6,7 @@
66pub const DEFAULT_AUTH_TIMEOUT: Duration = Duration::from_secs(5);
77pub const DEFAULT_MAX_PAYLOAD: usize = 1_000_000;
88/// 浏览器握手带不了 `Authorization`，所以允许从会话 cookie 取同一个 access token。
9pub const DEFAULT_SESSION_COOKIE: &str = "gitdata_session";
9pub const DEFAULT_SESSION_COOKIE: &str = config::event::socket::DEFAULT_SESSION_COOKIE;
1010
1111#[derive(Debug, Clone)]
1212pub struct SocketOptions {
Modified lib/share/events/src/gateway/action/platform_events.rs View lib/share/events/src/gateway/action/platform_events.rs
@@ -73,8 +73,8 @@
7373 return ConsumeResult::Ack { message };
7474 };
7575 let members = match context.platform.member_ids(organization_id).await {
76 Ok(members) if !members.is_empty() => members,
77 _ => return retry(message),
76 Ok(members) => members,
77 Err(_) => return retry(message),
7878 };
7979 let body = body(&message);
8080 let properties = &properties;
Modified lib/share/events/src/gateway/event_service.rs View lib/share/events/src/gateway/event_service.rs
@@ -72,17 +72,9 @@
7272 device_id: &str,
7373 namespace: &str,
7474 request: SubscriptionRequest,
75 access_token: &str,
7675 ) -> Result<String, String> {
7776 self.subscriptions
78 .subscribe(
79 &self.platform,
80 user_id,
81 device_id,
82 namespace,
83 request,
84 access_token,
85 )
77 .subscribe(&self.platform, user_id, device_id, namespace, request)
8678 .await
8779 }
8880
@@ -209,15 +201,15 @@
209201 "topic": "event",
210202 "tag": "access-revoked",
211203 "control": true,
204 "event_id": uuid::Uuid::now_v7().to_string(),
212205 "recipient_user_id": identity.user_id,
213206 "properties": {"user_id": identity.user_id},
214207 "body": {"reason": reason},
215208 });
216 self.host
217 .publish_event(identity.user_id, payload)
218 .await
219 .map_err(|_| tonic::Status::unavailable("delivery store is unavailable"))?;
220 Ok(self.hub.close_user(identity.user_id).await as u32)
209 let delivered = self.host.publish_event(identity.user_id, payload).await;
210 let closed = self.hub.close_user(identity.user_id).await as u32;
211 delivered.map_err(|_| tonic::Status::unavailable("delivery store is unavailable"))?;
212 Ok(closed)
221213 }
222214
223215 pub fn consume_result(_: ConsumeResult) {}
Modified lib/share/events/src/gateway/pipe.rs View lib/share/events/src/gateway/pipe.rs
@@ -12,6 +12,9 @@
1212use super::result::ConsumeResult;
1313
1414pub const QUEUE_CAPACITY: usize = 1024;
15/// 有序消费线程等待单条事件完成的上限；必须远小于 MQ 侧 `consume_timeout`，
16/// 否则 MQ 先判超时重投，同一条消息会被两个消费线程并发处理。
17const MAX_LISTENER_WAIT: Duration = Duration::from_secs(60);
1518
1619pub struct EventsPipe {
1720 consumer: DefaultMQPushConsumer,
@@ -43,7 +46,7 @@
4346 .build();
4447 consumer.register_message_listener_orderly(Forwarder::new(
4548 events_tx,
46 Duration::from_millis(config.consume_timeout_ms),
49 Duration::from_millis(config.consume_timeout_ms).min(MAX_LISTENER_WAIT),
4750 ));
4851
4952 Ok(Self {
@@ -80,8 +83,16 @@
8083 };
8184 let key = message.msg_id().to_string();
8285 if let Some(queue) = self.pending.get_mut(&key) {
83 if let Some(completion) = queue.pop_front() {
84 let _ = completion.send(success);
86 while let Some(completion) = queue.pop_front() {
87 let retry = completion.clone();
88 match completion.try_send(success) {
89 Ok(()) => break,
90 Err(std::sync::mpsc::TrySendError::Disconnected(_)) => continue,
91 Err(std::sync::mpsc::TrySendError::Full(_)) => {
92 queue.push_front(retry);
93 break;
94 }
95 }
8596 }
8697 if queue.is_empty() {
8798 self.pending.remove(&key);
Modified lib/share/events/src/gateway/socket_backend.rs View lib/share/events/src/gateway/socket_backend.rs
@@ -1,3 +1,4 @@
1use std::collections::HashMap;
12use std::time::Duration;
23
34use actor::{ActorHost, PushStore};
@@ -11,6 +12,8 @@
1112
1213const IDLE_FALLBACK: Duration = Duration::from_secs(30);
1314const STANDBY_WAIT: Duration = Duration::from_secs(2);
15/// 单条订阅的投递授权结论缓存时长；与身份复核的 60 秒节奏一致。
16const AUTHORIZATION_TTL: Duration = Duration::from_secs(60);
1417
1518/// socket 通道与事件业务之间的适配层：身份、订阅、投递策略都落在这里。
1619#[derive(Clone)]
@@ -99,6 +102,7 @@
99102 let notification = hub.notification(user).await;
100103 let work = async {
101104 let mut verified: Option<tokio::time::Instant> = None;
105 let mut authorized: HashMap<String, tokio::time::Instant> = HashMap::new();
102106 loop {
103107 let changed = notification.notified();
104108 tokio::pin!(changed);
@@ -152,6 +156,12 @@
152156 {
153157 continue;
154158 }
159 if authorized
160 .get(&subscription.id)
161 .is_some_and(|at| at.elapsed() < AUTHORIZATION_TTL)
162 {
163 return Ok(true);
164 }
155165 if !subscriptions
156166 .contains(&subscription)
157167 .await
@@ -170,6 +180,7 @@
170180 continue;
171181 }
172182 }
183 authorized.insert(subscription.id.clone(), tokio::time::Instant::now());
173184 return Ok(true);
174185 }
175186 return Ok(false);
Modified lib/share/events/src/gateway/socket_runtime.rs View lib/share/events/src/gateway/socket_runtime.rs
@@ -1,4 +1,5 @@
11use std::sync::Arc;
2use std::time::Duration;
23
34use actor::ActorHost;
45use serde_json::Value;
@@ -12,6 +13,8 @@
1213use crate::gateway::subscriptions::{SubscriptionRequest, Subscriptions};
1314use crate::gateway::{ConsumeResult, EventsPipe, action, is_personal};
1415
16const SHUTDOWN_DRAIN: Duration = Duration::from_secs(10);
17
1518pub struct EventActor;
1619
1720impl actor::VirtualActor for EventActor {
@@ -98,7 +101,6 @@
98101 &event.device_id,
99102 &event.namespace,
100103 request,
101 &event.access_token,
102104 )
103105 .await?;
104106 Ok(serde_json::json!({"id":id}))
@@ -320,9 +322,26 @@
320322 }
321323 }
322324 }
323 workers.abort_all();
324 while workers.join_next().await.is_some() {}
325325 pipe.shutdown().await;
326 drop(lanes);
327 let drained = tokio::time::timeout(SHUTDOWN_DRAIN, async {
328 loop {
329 tokio::select! {
330 result = completed.recv() => {
331 let Some(result) = result else { break };
332 let _ = pipe.apply_result(result).await;
333 }
334 joined = workers.join_next() => {
335 if joined.is_none() { break }
336 }
337 }
338 }
339 })
340 .await;
341 if drained.is_err() {
342 workers.abort_all();
343 while workers.join_next().await.is_some() {}
344 }
326345 Ok(())
327346}
328347
@@ -337,7 +356,7 @@
337356 .map(ToString::to_string)
338357 .unwrap_or_default();
339358 let fields: &[&str] = if is_personal(topic.as_str(), &tag) {
340 &["user"]
359 &["recipient", "user"]
341360 } else {
342361 &["room_id", "issue_id", "pull_id", "repo_id", "org", "user"]
343362 };
Modified lib/share/events/src/gateway/subscriptions.rs View lib/share/events/src/gateway/subscriptions.rs
@@ -56,7 +56,6 @@
5656 device_id: &str,
5757 namespace: &str,
5858 request: SubscriptionRequest,
59 token: &str,
6059 ) -> Result<String, String> {
6160 Self::validate(&request.topic, &request.tag)?;
6261 if user_id <= 0 || device_id.is_empty() {
@@ -75,7 +74,6 @@
7574 }
7675 filter.user_id = Some(user_id.to_string());
7776 }
78 let _ = token;
7977 let path = resource_path(&filter, request.permission_path.as_deref())?;
8078 let has_resource = filter.room_id.is_some()
8179 || filter.issue_id.is_some()
Modified lib/share/events/src/rpc/mod.rs View lib/share/events/src/rpc/mod.rs
@@ -134,7 +134,6 @@
134134 Some(request.permission_path)
135135 },
136136 },
137 "",
138137 )
139138 .await
140139 .map_err(|error| Status::invalid_argument(error))?;