zhenyi

Fork 0

Clone

feat(app): run the event system and serve the internal rpc plane

- add the `event` subcommand (`--mode single|consumer|actor`) that loads only
  the event configuration, starts Redis/MQ/actor/socket plus the internal rpc
  server, and shares one shutdown path with the consumer
- start the platform internal rpc server from `serve` and pass the session
  store into `Deps` so `AuthService` can resolve session credentials
- keep the command name in line with the existing `app serve|mail|migrate|
  openapi` subcommands; `cargo run -- event` now runs the event system

ZhenYi committed 3 days ago

parent 16132e1 commit ec0bdb8 Browse files

app/src/cli.rs +14 −1 View

@@ -1,4 +1,4 @@

1use clap::{Parser, Subcommand};

1use clap::{Parser, Subcommand, ValueEnum};

22

33#[derive(Parser)]

44#[command(name = "app", version)]

@@ -16,4 +16,17 @@

1616 #[arg(long)]

1717 check: bool,

1818 },

19 /// 事件系统：RocketMQ 消费 + socket.io 通道 + 内部 RPC

20 Event {

21 #[arg(long, value_enum, default_value_t = EventMode::Single)]

22 mode: EventMode,

23 },

24}

25

26/// 事件系统的运行角色；拆分布署时用 `consumer` / `actor` 单独启动。

27#[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)]

28pub enum EventMode {

29 Single,

30 Consumer,

31 Actor,

1932}

app/src/event.rs +169 −0 View

@@ -0,0 +1,169 @@

1//! `app event` 子命令：事件系统（RocketMQ 消费 + socket.io 通道 + 内部 RPC）。

2//!

3//! 只加载 `event` 段（socket/actor/redis/consumer/rpc/platform），不加载数据库、

4//! S3 或任何业务域配置；身份与权限全部经内部 RPC 问 `serve`。

5

6use std::sync::Arc;

7use std::time::Duration;

8

9use crate::cli::EventMode;

10use anyhow::{Context, anyhow};

11use config::event::EventConfig;

12use events::gateway::event_service::EventService;

13use events::gateway::platform::PlatformClient;

14use events::gateway::socket_runtime::{self, EventActor};

15use events::gateway::subscriptions::Subscriptions;

16use socket::{SocketHub, SocketOptions};

17use tokio::sync::watch;

18use tokio_util::sync::CancellationToken;

19

20pub async fn run(mode: EventMode) -> anyhow::Result<()> {

21 let telemetry = oltp::init(&oltp::TelemetryConfig::from_env("gitdata-event"));

22

23 let cfg = EventConfig::load().context("loading event configuration")?;

24

25 let platform = PlatformClient::new(&cfg.platform).map_err(|error| anyhow!("{error}"))?;

26 let nonce_store = internal::NonceStore::connect(&cfg.redis, "event")

27 .await

28 .map_err(|error| anyhow!("{error}"))?;

29 let private_pem = cfg

30 .rpc

31 .identity_private_key

32 .resolve()

33 .map_err(|error| anyhow!("resolving event identity private key: {error}"))?;

34 let mac_key = cfg

35 .rpc

36 .platform_mac_key

37 .resolve()

38 .map_err(|error| anyhow!("resolving platform mac key: {error}"))?;

39 let verifier = internal::EnvelopeVerifier::new(

40 &cfg.rpc.identity_key_id,

41 &private_pem,

42 &cfg.rpc.platform_mac_key_id,

43 mac_key.as_bytes(),

44 cfg.rpc.clock_skew_secs,

45 cfg.rpc.nonce_ttl_secs,

46 nonce_store,

47 )

48 .map_err(|error| anyhow!("{error}"))?;

49

50 let mut options = actor::ActorHostOptions::from(&cfg.actor);

51 if let Some(node_id) = cfg

52 .actor

53 .node_id

54 .as_deref()

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

56 {

57 options.node_id = node_id.to_string();

58 }

59 let host = actor::ActorHost::builder(options, cfg.redis.clone())

60 .register(|_| EventActor)

61 .start()

62 .await

63 .map_err(|error| anyhow!("starting the actor host: {error}"))?;

64

65 let subscriptions = Subscriptions::connect(&cfg.redis)

66 .await

67 .map_err(|error| anyhow!("connecting the subscription store: {error}"))?;

68 let push_store = host.push_store();

69 let backend = socket_runtime::backend(

70 host.clone(),

71 platform.clone(),

72 subscriptions.clone(),

73 push_store,

74 );

75 let socket_options = SocketOptions::from(&cfg.socket);

76 let (hub, events_rx) = SocketHub::new(socket_options, backend)

77 .map_err(|error| anyhow!("building the socket hub: {error}"))?;

78 let service = EventService::new(

79 host.clone(),

80 platform.clone(),

81 subscriptions.clone(),

82 Arc::new(verifier),

83 hub.clone(),

84 );

85

86 let cancel = CancellationToken::new();

87 let (shutdown_tx, shutdown_rx) = watch::channel(false);

88

89 let server_config = internal::ServerConfig::new(

90 cfg.rpc.listen_address.clone(),

91 cfg.rpc.rpc_key.clone(),

92 vec![

93 "internal.v1.EventIngressService",

94 "internal.v1.SubscriptionService",

95 ],

96 );

97 let router = internal::server::router(&server_config)

98 .map_err(|error| anyhow!("building the internal rpc server: {error}"))?;

99 let gate = server_config.gate();

100 let router = events::rpc::register_ingress(router, &service, gate.clone());

101 let router = events::rpc::register(router, &service, gate);

102 let rpc_listen = server_config.listen_address().to_string();

103 let mut rpc_shutdown = shutdown_rx.clone();

104 let rpc_task = tokio::spawn(async move {

105 let stop = async move {

106 while !*rpc_shutdown.borrow() {

107 if rpc_shutdown.changed().await.is_err() {

108 break;

109 }

110 }

111 };

112 if let Err(error) = internal::server::serve(router, &rpc_listen, stop).await {

113 tracing::error!(%error, "event internal rpc server stopped");

114 }

115 });

116

117 // 信号：oltp 统一安装（同进程只装一次）。

118 let signal = oltp::shutdown::install();

119 let signal_tx = shutdown_tx.clone();

120 let signal_cancel = cancel.clone();

121 let signal_task = tokio::spawn(async move {

122 signal.wait().await;

123 signal_tx.send_replace(true);

124 signal_cancel.cancel();

125 });

126

127 tracing::info!(

128 mode = ?mode,

129 socket = %cfg.socket.listen_address,

130 rpc = %cfg.rpc.listen_address,

131 "event system starting"

132 );

133

134 let socket_future = socket_runtime::run_socket_node(

135 hub.clone(),

136 events_rx,

137 host.clone(),

138 platform.clone(),

139 subscriptions.clone(),

140 &cfg.socket.listen_address,

141 shutdown_rx.clone(),

142 );

143 let consumer_future = socket_runtime::run_consumer(

144 &cfg.consumer,

145 host.clone(),

146 platform.clone(),

147 subscriptions.clone(),

148 shutdown_rx.clone(),

149 );

150

151 let result = match mode {

152 EventMode::Single => {

153 let (socket_result, consumer_result) = tokio::join!(socket_future, consumer_future);

154 socket_result.and(consumer_result)

155 }

156 EventMode::Actor => socket_future.await,

157 EventMode::Consumer => consumer_future.await,

158 };

159

160 shutdown_tx.send_replace(true);

161 signal_task.abort();

162 let _ = rpc_task.await;

163 host.shutdown().await;

164 tokio::time::sleep(Duration::from_millis(10)).await;

165 if let Err(error) = telemetry.shutdown() {

166 tracing::warn!(%error, "flushing telemetry failed");

167 }

168 result.map_err(|error| anyhow!("event system failed: {error}"))

169}

app/src/main.rs +57 −1 View

@@ -1,6 +1,7 @@

11#![recursion_limit = "256"]

22

33mod cli;

4mod event;

45

56use std::net::SocketAddr;

67use std::path::PathBuf;

@@ -16,10 +17,16 @@

1617 Command::Mail => mail(),

1718 Command::Migrate => migrate(),

1819 Command::Openapi { check } => openapi(check),

20 Command::Event { mode } => event_run(mode),

1921 }

2022}

2123

2224#[tokio::main]

25async fn event_run(mode: cli::EventMode) -> anyhow::Result<()> {

26 event::run(mode).await

27}

28

29#[tokio::main]

2330async fn serve() -> anyhow::Result<()> {

2431 use actix_web::middleware::from_fn;

2532 use actix_web::{App, HttpServer, web};

@@ -76,9 +83,11 @@

7683 )

7784 .await

7885 .context("building AppState")?;

79 let deps = webapi::bootstrap::Deps::new(state.clone());

86 let deps = webapi::bootstrap::Deps::new(state.clone(), session_store.clone());

8087 info!("dependencies ready");

8188

89 let internal_rpc = start_internal_rpc(&cfg, &deps, &cancel)?;

90

8291 spawn_queue_depth_reporter(

8392 state.redis.clone(),

8493 &[

@@ -187,6 +196,9 @@

187196 }

188197 };

189198 let (api_result, ()) = tokio::join!(api, signal);

199 if let Some(task) = internal_rpc {

200 let _ = task.await;

201 }

190202 let _ = relay_stop.send(());

191203 let relay_failed = relay_task.await.unwrap_or(true);

192204 for task in push_tasks {

@@ -354,6 +366,50 @@

354366 Ok(Some(auth::config::Config::from_value(value)))

355367}

356368

369/// 启动平台侧内部 RPC 服务端（Identity + Chat/Issue/Pull/Space 授权）。

370///

371/// 未配置 `internal.rpc.listen_address` 时跳过：serve 仍可单独运行。

372fn start_internal_rpc(

373 cfg: &config::AppConfig,

374 deps: &webapi::bootstrap::Deps,

375 cancel: &tokio_util::sync::CancellationToken,

376) -> anyhow::Result<Option<tokio::task::JoinHandle<()>>> {

377 let listen = cfg.internal.rpc.listen_address.trim();

378 if listen.is_empty() {

379 return Ok(None);

380 }

381 if cfg.internal.rpc.rpc_key.trim().is_empty() {

382 return Err(anyhow!(

383 "internal.rpc.rpc_key must be configured when internal.rpc.listen_address is set"

384 ));

385 }

386 let server_config = internal::ServerConfig::new(

387 listen,

388 cfg.internal.rpc.rpc_key.clone(),

389 vec![

390 "internal.v1.IdentityService",

391 "internal.v1.ChatAuthorizationService",

392 "internal.v1.IssueAuthorizationService",

393 "internal.v1.PullAuthorizationService",

394 "internal.v1.SpaceAuthorizationService",

395 ],

396 );

397 let router = internal::server::router(&server_config).map_err(|error| anyhow!("{error}"))?;

398 let gate = server_config.gate();

399 let router = auth::rpc::register(router, &deps.auth, gate.clone());

400 let router = chat::rpc::register(router, &deps.chat, gate.clone());

401 let router = space::rpc::register(router, &deps.space, gate);

402 let address = server_config.listen_address().to_string();

403 let shutdown = cancel.clone();

404 tracing::info!(%address, "internal rpc listening");

405 Ok(Some(tokio::spawn(async move {

406 let stop = async move { shutdown.cancelled().await };

407 if let Err(error) = internal::server::serve(router, &address, stop).await {

408 tracing::error!(%error, "internal rpc server stopped");

409 }

410 })))

411}

412

357413const QUEUE_DEPTH_INTERVAL: std::time::Duration = std::time::Duration::from_secs(30);

358414

359415/// 周期性上报队列与死信队列的积压深度；进程退出时随运行时一起结束。

app/Cargo.toml +5 −1 View

@@ -10,11 +10,15 @@

1010auth = { workspace = true }

1111actix-web = { workspace = true }

1212chat = { workspace = true }

13socket = { workspace = true }

14actor = { workspace = true }

1315clap = { workspace = true }

1416config = { workspace = true }

1517deletion = { workspace = true }

16events = { workspace = true }

18events = { workspace = true, features = ["gateway"] }

1719export = { workspace = true }

20space = { workspace = true }

21internal = { workspace = true }

1822session = { workspace = true }

1923models = { workspace = true }

2024oltp = { workspace = true }

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

@@ -23,7 +23,7 @@

2323}

2424

2525impl Deps {

26 pub fn new(state: AppState) -> Self {

26 pub fn new(state: AppState, sessions: auth::session::RedisStore) -> Self {

2727 let accounts =

2828 auth::Accounts::new(state.database.connection().clone(), state.cache.clone());

2929 let access = space::Access::new(state.database.connection().clone(), state.cache.clone());

@@ -33,6 +33,7 @@

3333 accounts.clone(),

3434 auth::services::AuthSettings::from_config(&state.config),

3535 state.mail.clone(),

36 sessions,

3637 );

3738 let issue_audience = issues::services::audience::Audience::new(

3839 accounts.clone(),