feat(redis): enhance connection handling with optional connection reuse

Modified app/src/event.rs +6 −5 View app/src/event.rs

@@ -23,9 +23,11 @@

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

2424

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

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

26 let conn = cache::CacheConn::init(&cfg.redis)

2727 .await

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

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

29 let actor_conn = actor::RedisConnection::from_conn(conn.clone());

30 let nonce_store = internal::NonceStore::from_conn(conn);

2931 let private_pem = cfg

3032 .rpc

3133 .identity_private_key

@@ -57,14 +59,13 @@

5759 options.node_id = node_id.to_string();

5860 }

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

62 .connection(actor_conn.clone())

6063 .register(|_| EventActor)

6164 .start()

6265 .await

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

6467

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

66 .await

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

68 let subscriptions = Subscriptions::from_connection(actor_conn);

6869 let push_store = host.push_store();

6970 let backend = socket_runtime::backend(

7071 host.clone(),

Modified app/src/main.rs +1 −3 View app/src/main.rs

@@ -48,9 +48,6 @@

4848 let cancel = CancellationToken::new();

4949 let app_config = cfg.clone();

5050 let session_config = app_config.session.clone();

51 let session_store = auth::session::RedisStore::connect(&app_config.redis)

52 .await

53 .context("connecting the session store")?;

5451 let database = std::sync::Arc::new(

5552 db::AppDatabase::connect(app_config.database.clone())

5653 .await

@@ -84,6 +81,7 @@

8481 )

8582 .await

8683 .context("building AppState")?;

84 let session_store = auth::session::RedisStore::new(state.cache.clone());

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

8886 info!("dependencies ready");

8987

Modified lib/infra/actor/src/actor/host/builder.rs +12 −1 View lib/infra/actor/src/actor/host/builder.rs

@@ -19,6 +19,7 @@

1919 pub(super) options: ActorHostOptions,

2020 pub(super) redis: RedisConfig,

2121 pub(super) factory: Option<ActorFactory>,

22 pub(super) connection: Option<redis_conn::RedisConnection>,

2223}

2324

2425impl ActorHostBuilder {

@@ -27,9 +28,16 @@

2728 options,

2829 redis,

2930 factory: None,

31 connection: None,

3032 }

3133 }

3234

35 /// Reuses an already established pool instead of opening another one.

36 pub fn connection(mut self, connection: redis_conn::RedisConnection) -> Self {

37 self.connection = Some(connection);

38 self

39 }

40

3341 pub fn register<A>(mut self, factory: impl Fn(ActorId) -> A + Send + Sync + 'static) -> Self

3442 where

3543 A: VirtualActor,

@@ -51,7 +59,10 @@

5159 .factory

5260 .ok_or_else(|| ActorError::InvalidOptions("no actor type registered".to_string()))?;

5361

54 let (commands, reads) = redis_conn::connect(&self.redis).await?;

62 let (commands, reads) = match self.connection.clone() {

63 Some(connection) => (connection.clone(), connection),

64 None => redis_conn::connect(&self.redis).await?,

65 };

5566 let directory = RedisDirectory::new(

5667 commands.clone(),

5768 options.key_prefix.clone(),

Modified lib/infra/actor/src/actor/redis_conn.rs +5 −1 View lib/infra/actor/src/actor/redis_conn.rs

@@ -14,11 +14,15 @@

1414}

1515

1616impl RedisConnection {

17 pub fn from_conn(conn: CacheConn) -> Self {

18 Self { conn }

19 }

20

1721 pub async fn connect(config: &RedisConfig) -> Result<Self, ActorError> {

1822 let conn = CacheConn::init(config)

1923 .await

2024 .map_err(|error| ActorError::InvalidOptions(error.to_string()))?;

21 Ok(Self { conn })

25 Ok(Self::from_conn(conn))

2226 }

2327

2428 pub async fn query<T: FromRedisValue>(&self, cmd: &Cmd) -> RedisResult<T> {

Modified lib/infra/bootstrap/src/health.rs +2 −6 View lib/infra/bootstrap/src/health.rs

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

66use sea_orm::{ConnectionTrait, DbBackend, Statement};

77

88use api::http::{Router, State, get};

9use cache::{AppRedis, RedisConnection};

9use cache::AppRedis;

1010use db::AppDatabase;

1111

1212use crate::AppState;

@@ -35,11 +35,7 @@

3535 .await

3636 .is_ok(),

3737 };

38 let mut conn = state.redis.connection_clone();

39 let redis_ok: bool = match &mut conn {

40 RedisConnection::Standalone(c) => redis::cmd("PING").query_async::<String>(c).await.is_ok(),

41 RedisConnection::Cluster(c) => redis::cmd("PING").query_async::<String>(c).await.is_ok(),

42 };

38 let redis_ok = state.redis.ping().await.is_ok();

4339 let rocketmq_ok = queue::probe_nameservers_with(&state.rocketmq, Duration::from_secs(1))

4440 .await

4541 .is_ok();

Modified lib/infra/bootstrap/src/lib.rs +3 −2 View lib/infra/bootstrap/src/lib.rs

@@ -37,8 +37,9 @@

3737 mail: Option<email::MailPublisher>,

3838 events: Option<events::EventHandler>,

3939 ) -> Result<Self, BootstrapError> {

40 let redis = cache::AppRedis::connect(config.redis.clone()).await?;

41 let cache = cache::Cache::connect(&config.redis).await?;

40 let conn = cache::CacheConn::init(&config.redis).await?;

41 let redis = cache::AppRedis::from_conn(conn.clone());

42 let cache = cache::Cache::from_conn(conn, &config.redis);

4243 let storage: Arc<storage::S3Client> = Arc::new(storage::S3Client::new(&config.s3)?);

4344

4445 Ok(Self::from_parts(

Modified lib/infra/cache/src/client.rs +26 −292 View lib/infra/cache/src/client.rs

@@ -1,118 +1,50 @@

1use std::sync::Arc;

2use std::sync::atomic::{AtomicUsize, Ordering};

3

41use config::redis::{RedisConfig, RedisMode};

5use redis::aio::{ConnectionManager, ConnectionManagerConfig, MultiplexedConnection};

6use redis::cluster_async::ClusterConnection;

7use redis::{ConnectionAddr, ConnectionInfo, Value};

82

3use crate::conn::{BlockingConnection, CacheConn, CacheConnection};

94use crate::error::CacheError;

105

11#[derive(Clone)]

12pub enum RedisConnection {

13 Standalone(ConnectionManager),

14 Cluster(ClusterConnection),

15}

6pub type RedisConnection = CacheConnection;

167

17impl RedisConnection {

18 pub fn is_cluster(&self) -> bool {

19 matches!(self, RedisConnection::Cluster(_))

20 }

21}

22

8/// Raw-key facade over the shared Redis pool.

9///

10/// Unlike [`crate::Cache`] it does not apply `redis.key_prefix`, because queue,

11/// quota and lock keys are consumed verbatim by other processes and workers.

2312#[derive(Clone)]

2413pub struct AppRedis {

25 config: RedisConfig,

26 /// One multiplexed connection per pool slot; every slot is an independent

27 /// connection (not a clone of another slot) so requests spread over

28 /// multiple sockets.

29 connections: Vec<RedisConnection>,

30 /// Round-robin cursor shared across clones of this handle.

31 cursor: Arc<AtomicUsize>,

32 /// Connection parameters (auth, tls) of the first configured node,

33 /// reused to open per-node connections for cluster-wide operations.

34 node_info: Option<ConnectionInfo>,

14 conn: CacheConn,

3515}

3616

3717impl AppRedis {

3818 pub async fn connect(config: RedisConfig) -> Result<Self, CacheError> {

39 let url = config.validate().map_err(CacheError::config)?;

40 let pool_size = config.pool_size().max(1);

41

42 let (connections, node_info) = match config.mode() {

43 RedisMode::Cluster => {

44 let nodes = url

45 .split(',')

46 .map(str::trim)

47 .filter(|node| !node.is_empty())

48 .map(|node| connection_info(node, &config))

49 .collect::<Result<Vec<_>, _>>()?;

50

51 let node_info = nodes.first().cloned();

52 let client = redis::cluster::ClusterClient::builder(nodes)

53 .connection_timeout(config.connection_timeout())

54 .build()

55 .map_err(CacheError::from)?;

56

57 let mut connections = Vec::with_capacity(pool_size);

58 for _ in 0..pool_size {

59 let connection = client

60 .get_async_connection()

61 .await

62 .map_err(CacheError::from)?;

63

64 connections.push(RedisConnection::Cluster(connection));

65 }

66

67 (connections, node_info)

68 }

69 RedisMode::Standalone | RedisMode::Sentinel | RedisMode::ValkeyStandalone => {

70 let node_info = connection_info(url, &config)?;

71 let client = redis::Client::open(node_info.clone()).map_err(CacheError::from)?;

72 let manager_config = ConnectionManagerConfig::new()

73 .set_connection_timeout(Some(config.connection_timeout()));

74 let mut connections = Vec::with_capacity(pool_size);

75 for _ in 0..pool_size {

76 let manager =

77 ConnectionManager::new_with_config(client.clone(), manager_config.clone())

78 .await

79 .map_err(CacheError::from)?;

80

81 connections.push(RedisConnection::Standalone(manager));

82 }

83

84 (connections, Some(node_info))

85 }

86 };

19 Ok(Self::from_conn(CacheConn::init(&config).await?))

20 }

8721

88 Ok(Self {

89 config,

90 connections,

91 cursor: Arc::new(AtomicUsize::new(0)),

92 node_info,

93 })

22 pub fn from_conn(conn: CacheConn) -> Self {

23 Self { conn }

9424 }

9525

9626 pub fn config(&self) -> &RedisConfig {

97 &self.config

27 self.conn.config()

9828 }

9929

10030 pub fn mode(&self) -> RedisMode {

101 self.config.mode()

31 self.conn.mode()

10232 }

10333

10434 pub fn is_cluster(&self) -> bool {

105 self.config.mode() == RedisMode::Cluster

35 self.conn.is_cluster()

10636 }

10737

10838 pub fn connection(&self) -> &RedisConnection {

109 &self.connections[0]

39 self.conn.connection()

11040 }

11141

11242 pub fn connection_clone(&self) -> RedisConnection {

113 let index = self.cursor.fetch_add(1, Ordering::Relaxed) % self.connections.len();

43 self.conn.connection_clone()

44 }

11445

115 self.connections[index].clone()

46 pub async fn ping(&self) -> Result<(), CacheError> {

47 self.conn.ping().await

11648 }

11749

11850 pub async fn publish(&self, channel: &str, payload: &str) -> Result<(), CacheError> {

@@ -216,125 +148,20 @@

216148

217149 /// 阻塞式右端出队（`BRPOP key timeout`）；超时返回 `None`。

218150 ///

219 /// 走独立连接而不是连接池：BRPOP 会独占 socket 整个超时窗口，池化连接被

220 /// 它占住会让同一进程里的其它 Redis 调用一起排长队。

151 /// 每次调用新建并释放一条独立连接。需要长期轮询的消费者应改用

152 /// [`AppRedis::blocking`]，复用同一条 socket。

221153 pub async fn brpop_bytes(

222154 &self,

223155 key: &str,

224156 timeout_secs: u64,

225157 ) -> Result<Option<Vec<u8>>, CacheError> {

226 let mut conn = self.blocking_consumer_connection(key).await?;

227 let result: Option<(String, Vec<u8>)> = redis::cmd("BRPOP")

228 .arg(key)

229 .arg(timeout_secs.max(1))

230 .query_async(&mut conn)

231 .await

232 .map_err(CacheError::from)?;

233 Ok(result.map(|(_, payload)| payload))

158 let mut conn = self.conn.blocking(key).await?;

159 conn.brpop_bytes(key, timeout_secs).await

234160 }

235161

236 /// 打开一条专供阻塞命令的连接。

237 ///

238 /// cluster 下 BRPOP 落在持有该 key slot 的 master 上，否则会收到 MOVED；

239 /// standalone/sentinel 直接用首个节点。

240 async fn blocking_consumer_connection(

241 &self,

242 key: &str,

243 ) -> Result<MultiplexedConnection, CacheError> {

244 let client = if self.is_cluster() {

245 let node_info = self

246 .node_info

247 .clone()

248 .ok_or_else(|| CacheError::config("cluster connection info is unavailable"))?;

249 let mut conn = self.connection_clone();

250 let RedisConnection::Cluster(cluster) = &mut conn else {

251 return Err(CacheError::config("cluster connection missing"));

252 };

253 let slot: u16 = redis::cmd("CLUSTER")

254 .arg("KEYSLOT")

255 .arg(key)

256 .query_async(cluster)

257 .await

258 .map_err(CacheError::from)?;

259 let slots: Value = redis::cmd("CLUSTER")

260 .arg("SLOTS")

261 .query_async(cluster)

262 .await

263 .map_err(CacheError::from)?;

264 let (host, port) = 'inline_cluster_master_for_slot: {

265 let slots = &slots;

266 let Value::Array(groups) = slots else {

267 break 'inline_cluster_master_for_slot None;

268 };

269 for group in groups {

270 let Value::Array(fields) = group else {

271 continue;

272 };

273 let (Some(Value::Int(start)), Some(Value::Int(end))) =

274 (fields.first(), fields.get(1))

275 else {

276 continue;

277 };

278 if !(*start..=*end).contains(&(slot as i64)) {

279 continue;

280 }

281 let Some(Value::Array(nodes)) = fields.get(2) else {

282 continue;

283 };

284 let (Some(Value::BulkString(host)), Some(Value::Int(port))) =

285 (nodes.first(), nodes.get(1))

286 else {

287 continue;

288 };

289 break 'inline_cluster_master_for_slot Some((

290 String::from_utf8_lossy(host).into_owned(),

291 *port as u16,

292 ));

293 }

294 None

295 }

296 .ok_or_else(|| CacheError::config(format!("no master owns slot {slot}")))?;

297 let addr = match node_info.addr() {

298 ConnectionAddr::Tcp(_, _) => ConnectionAddr::Tcp(host, port),

299 ConnectionAddr::TcpTls {

300 insecure,

301 tls_params,

302 ..

303 } => ConnectionAddr::TcpTls {

304 host,

305 port,

306 insecure: *insecure,

307 tls_params: tls_params.clone(),

308 },

309 ConnectionAddr::Unix(_) => {

310 return Err(CacheError::config(

311 "unix sockets are not supported for cluster connections",

312 ));

313 }

314 _ => {

315 return Err(CacheError::config(

316 "unsupported connection address type for cluster connections",

317 ));

318 }

319 };

320 redis::Client::open(node_info.set_addr(addr)).map_err(CacheError::from)?

321 } else {

322 let url = self

323 .config

324 .url()

325 .split(',')

326 .map(str::trim)

327 .find(|node| !node.is_empty())

328 .ok_or_else(|| CacheError::config("redis url contains no nodes"))?;

329 redis::Client::open(connection_info(url, &self.config)?).map_err(CacheError::from)?

330 };

331 let config = redis::AsyncConnectionConfig::new()

332 .set_response_timeout(None)

333 .set_connection_timeout(Some(self.config.connection_timeout()));

334 client

335 .get_multiplexed_async_connection_with_config(&config)

336 .await

337 .map_err(CacheError::from)

162 /// Opens a long-lived blocking connection for a queue consumer.

163 pub async fn blocking(&self, key: &str) -> Result<BlockingConnection, CacheError> {

164 self.conn.blocking(key).await

338165 }

339166

340167 /// `SET key 1 NX EX ttl`; returns false when the key already exists

@@ -411,57 +238,6 @@

411238 }

412239 self.incr_window(key, window_secs).await <= max

413240 }

414

415 /// Returns `(slot_start, slot_end, master_index)` triples where `master_index`

416 /// matches the position of the owning connection in [`cluster_master_connections`].

417 ///

418 /// Empty for standalone connections. The ordering mirrors the dedup logic used

419 /// when collecting master addresses, so callers can map a slot to the right

420 /// pre-established connection without reconnecting.

421 pub async fn cluster_slot_ranges(&self) -> Result<Vec<(i64, i64, usize)>, CacheError> {

422 let mut conn = self.connection_clone();

423 let RedisConnection::Cluster(cluster) = &mut conn else {

424 return Ok(Vec::new());

425 };

426

427 let slots: Value = redis::cmd("CLUSTER")

428 .arg("SLOTS")

429 .query_async(cluster)

430 .await

431 .map_err(CacheError::from)?;

432

433 Ok(cluster_slot_ranges(&slots))

434 }

435}

436

437/// Builds the connection info of the first configured node, used as a template

438/// (scheme, auth, tls settings) for connections to other cluster nodes.

439fn connection_info(urls: &str, config: &RedisConfig) -> Result<ConnectionInfo, CacheError> {

440 let first = urls

441 .split(',')

442 .map(str::trim)

443 .find(|node| !node.is_empty())

444 .ok_or_else(|| CacheError::config("redis url contains no nodes"))?;

445

446 let url = if config.mode() == RedisMode::Cluster

447 && !(first.starts_with("redis://") || first.starts_with("rediss://"))

448 {

449 format!("redis://{first}")

450 } else {

451 first.to_string()

452 };

453 let info = redis::Client::open(url)?.get_connection_info().clone();

454 let mut settings = info.redis_settings().clone();

455 if let Some(database) = config.database {

456 settings = settings.set_db(i64::from(database));

457 }

458 if let Some(username) = config.username.as_deref().filter(|value| !value.is_empty()) {

459 settings = settings.set_username(username);

460 }

461 if let Some(password) = config.password.as_deref().filter(|value| !value.is_empty()) {

462 settings = settings.set_password(password);

463 }

464 Ok(info.set_redis_settings(settings))

465241}

466242

467243macro_rules! dispatch {

@@ -627,46 +403,4 @@

627403pub async fn exists(redis: &AppRedis, key: &str) -> Result<bool, redis::RedisError> {

628404 let mut conn = redis.connection_clone();

629405 dispatch!(conn, |c| redis::cmd("EXISTS").arg(key).query_async(c).await)

630}

631

632/// Extracts `(start, end, master_index)` for every slot range.

633///

634/// `master_index` counts unique master addresses in first-seen order, matching

635/// [`AppRedis::cluster_master_connections`].

636fn cluster_slot_ranges(slots: &Value) -> Vec<(i64, i64, usize)> {

637 let Value::Array(groups) = slots else {

638 return Vec::new();

639 };

640

641 let mut seen: std::collections::HashMap<(String, u16), usize> =

642 std::collections::HashMap::new();

643 let mut ranges = Vec::new();

644

645 for group in groups {

646 let Value::Array(fields) = group else {

647 continue;

648 };

649

650 let (Some(Value::Int(start)), Some(Value::Int(end))) = (fields.first(), fields.get(1))

651 else {

652 continue;

653 };

654

655 let Some(Value::Array(nodes)) = fields.get(2) else {

656 continue;

657 };

658

659 let (Some(Value::BulkString(host)), Some(Value::Int(port))) = (nodes.first(), nodes.get(1))

660 else {

661 continue;

662 };

663

664 let addr = (String::from_utf8_lossy(host).to_string(), *port as u16);

665 let next_index = seen.len();

666 let index = *seen.entry(addr).or_insert(next_index);

667

668 ranges.push((*start, *end, index));

669 }

670

671 ranges

672406}

Modified lib/infra/cache/src/conn.rs +190 −23 View lib/infra/cache/src/conn.rs

@@ -1,41 +1,56 @@

11use std::sync::Arc;

22use std::sync::atomic::{AtomicUsize, Ordering};

33

4use config::redis::RedisConfig;

5use redis::ConnectionInfo;

6use redis::aio::{ConnectionManager, ConnectionManagerConfig};

4use config::redis::{RedisConfig, RedisMode};

5use redis::aio::{ConnectionManager, ConnectionManagerConfig, MultiplexedConnection};

76use redis::cluster_async::ClusterConnection;

7use redis::{ConnectionAddr, ConnectionInfo, Value};

88

99use crate::error::{CacheError, CacheResult};

1010

11/// The process-wide Redis connection layer.

12///

13/// One instance owns `pool_size` multiplexed connections and is cloned into every

14/// consumer ([`crate::AppRedis`], [`crate::Cache`], session, actor, subscriptions)

15/// so a process opens exactly one pool instead of one per consumer.

1116#[derive(Clone)]

1217pub struct CacheConn {

1318 connections: Arc<Vec<CacheConnection>>,

1419 cursor: Arc<AtomicUsize>,

20 config: RedisConfig,

21 /// Connection parameters (scheme, auth, tls) of the first configured node,

22 /// reused to open per-node connections for cluster-wide operations.

23 node_info: Option<ConnectionInfo>,

1524 db: i64,

1625 cluster: bool,

1726}

1827

1928#[derive(Clone)]

2029pub enum CacheConnection {

21 Standard(ConnectionManager),

30 Standalone(ConnectionManager),

2231 Cluster(ClusterConnection),

2332}

2433

34impl CacheConnection {

35 pub fn is_cluster(&self) -> bool {

36 matches!(self, CacheConnection::Cluster(_))

37 }

38}

39

2540impl CacheConn {

2641 pub async fn init(cfg: &RedisConfig) -> CacheResult<Self> {

2742 cfg.validate().map_err(CacheError::Config)?;

2843

29 let db = cfg.database();

3044 let cluster = cfg.is_cluster();

31 let seeds = cfg.seeds().map_err(CacheError::Config)?;

45 let seeds = cfg.node_urls().map_err(CacheError::Config)?;

3246 let pool_size = cfg.pool_size().max(1);

3347

34 let connections = if cluster {

48 let (connections, node_info) = if cluster {

3549 let nodes = seeds

3650 .iter()

37 .map(|(host, port)| connection_info(host, *port, cfg))

51 .map(|url| connection_info(url, cfg))

3852 .collect::<CacheResult<Vec<_>>>()?;

53 let node_info = nodes.first().cloned();

3954 let client = redis::cluster::ClusterClient::builder(nodes)

4055 .connection_timeout(cfg.create_timeout())

4156 .build()

@@ -49,14 +64,13 @@

4964 .map_err(CacheError::from)?;

5065 connections.push(CacheConnection::Cluster(connection));

5166 }

52 connections

67 (connections, node_info)

5368 } else {

54 let (host, port) = seeds

69 let url = seeds

5570 .first()

56 .cloned()

5771 .ok_or_else(|| CacheError::config("redis requires at least one node"))?;

58 let info = connection_info(&host, port, cfg)?;

59 let client = redis::Client::open(info).map_err(CacheError::from)?;

72 let node_info = connection_info(url, cfg)?;

73 let client = redis::Client::open(node_info.clone()).map_err(CacheError::from)?;

6074 let manager_config =

6175 ConnectionManagerConfig::new().set_connection_timeout(Some(cfg.create_timeout()));

6276

@@ -66,14 +80,21 @@

6680 ConnectionManager::new_with_config(client.clone(), manager_config.clone())

6781 .await

6882 .map_err(CacheError::from)?;

69 connections.push(CacheConnection::Standard(manager));

83 connections.push(CacheConnection::Standalone(manager));

7084 }

71 connections

85 (connections, Some(node_info))

7286 };

7387

88 let db = node_info

89 .as_ref()

90 .map(|info| info.redis_settings().db())

91 .unwrap_or_else(|| cfg.database());

92

7493 let conn = Self {

7594 connections: Arc::new(connections),

7695 cursor: Arc::new(AtomicUsize::new(0)),

96 config: cfg.clone(),

97 node_info,

7798 db,

7899 cluster,

79100 };

@@ -97,6 +118,92 @@

97118 .map_err(CacheError::from)

98119 }

99120

121 pub fn config(&self) -> &RedisConfig {

122 &self.config

123 }

124

125 pub fn mode(&self) -> RedisMode {

126 self.config.mode()

127 }

128

129 pub fn connection(&self) -> &CacheConnection {

130 &self.connections[0]

131 }

132

133 pub fn connection_clone(&self) -> CacheConnection {

134 let index = self.cursor.fetch_add(1, Ordering::Relaxed) % self.connections.len();

135 self.connections[index].clone()

136 }

137

138 /// Opens a connection dedicated to a blocking command.

139 ///

140 /// `BRPOP` holds its socket for the whole timeout window, so it must not take

141 /// a pooled connection. Under cluster the command has to reach the master

142 /// owning the key's slot, otherwise it receives `MOVED`; standalone uses the

143 /// configured node directly.

144 pub async fn blocking_connection(&self, key: &str) -> CacheResult<MultiplexedConnection> {

145 let client = if self.cluster {

146 let node_info = self

147 .node_info

148 .clone()

149 .ok_or_else(|| CacheError::config("cluster connection info is unavailable"))?;

150 let mut conn = self.connection_clone();

151 let CacheConnection::Cluster(cluster) = &mut conn else {

152 return Err(CacheError::config("cluster connection missing"));

153 };

154 let slot: u16 = redis::cmd("CLUSTER")

155 .arg("KEYSLOT")

156 .arg(key)

157 .query_async(cluster)

158 .await

159 .map_err(CacheError::from)?;

160 let slots: Value = redis::cmd("CLUSTER")

161 .arg("SLOTS")

162 .query_async(cluster)

163 .await

164 .map_err(CacheError::from)?;

165 let (host, port) = master_for_slot(&slots, slot)

166 .ok_or_else(|| CacheError::config(format!("no master owns slot {slot}")))?;

167 let addr = match node_info.addr() {

168 ConnectionAddr::Tcp(_, _) => ConnectionAddr::Tcp(host, port),

169 ConnectionAddr::TcpTls {

170 insecure,

171 tls_params,

172 ..

173 } => ConnectionAddr::TcpTls {

174 host,

175 port,

176 insecure: *insecure,

177 tls_params: tls_params.clone(),

178 },

179 ConnectionAddr::Unix(_) => {

180 return Err(CacheError::config(

181 "unix sockets are not supported for cluster connections",

182 ));

183 }

184 _ => {

185 return Err(CacheError::config(

186 "unsupported connection address type for cluster connections",

187 ));

188 }

189 };

190 redis::Client::open(node_info.set_addr(addr)).map_err(CacheError::from)?

191 } else {

192 let node_info = self

193 .node_info

194 .clone()

195 .ok_or_else(|| CacheError::config("redis connection info is unavailable"))?;

196 redis::Client::open(node_info).map_err(CacheError::from)?

197 };

198 let config = redis::AsyncConnectionConfig::new()

199 .set_response_timeout(None)

200 .set_connection_timeout(Some(self.config.connection_timeout()));

201 client

202 .get_multiplexed_async_connection_with_config(&config)

203 .await

204 .map_err(CacheError::from)

205 }

206

100207 pub fn database(&self) -> i64 {

101208 self.db

102209 }

@@ -112,21 +219,54 @@

112219 pub fn available(&self) -> usize {

113220 self.connections.len()

114221 }

222

223 /// Opens a long-lived blocking connection for a queue consumer.

224 pub async fn blocking(&self, key: &str) -> CacheResult<BlockingConnection> {

225 Ok(BlockingConnection {

226 connection: self.blocking_connection(key).await?,

227 })

228 }

229}

230

231/// A socket dedicated to blocking commands, held for the consumer's lifetime.

232///

233/// Reusing one connection avoids a fresh TCP/TLS handshake on every poll cycle.

234pub struct BlockingConnection {

235 connection: MultiplexedConnection,

236}

237

238impl BlockingConnection {

239 /// `BRPOP key timeout`; returns `None` when the timeout elapses.

240 pub async fn brpop_bytes(

241 &mut self,

242 key: &str,

243 timeout_secs: u64,

244 ) -> CacheResult<Option<Vec<u8>>> {

245 let result: Option<(String, Vec<u8>)> = redis::cmd("BRPOP")

246 .arg(key)

247 .arg(timeout_secs.max(1))

248 .query_async(&mut self.connection)

249 .await

250 .map_err(CacheError::from)?;

251 Ok(result.map(|(_, payload)| payload))

252 }

115253}

116254

117fn connection_info(host: &str, port: u16, cfg: &RedisConfig) -> CacheResult<ConnectionInfo> {

118 let host = if host.contains(':') && !host.starts_with('[') {

119 format!("[{host}]")

120 } else {

121 host.to_string()

122 };

123 let url = format!("redis://{host}:{port}");

255/// Builds a node's connection info from its configured URL.

256///

257/// The URL is authoritative for scheme, credentials and database; the explicit

258/// config fields override it when set. This is the single place that decides how

259/// a seed string becomes a connection, so both the keyed and the raw-key

260/// consumers see identical semantics.

261fn connection_info(url: &str, cfg: &RedisConfig) -> CacheResult<ConnectionInfo> {

124262 let info = redis::Client::open(url)

125263 .map_err(CacheError::from)?

126264 .get_connection_info()

127265 .clone();

128266 let mut settings = info.redis_settings().clone();

129 settings = settings.set_db(cfg.database());

267 if let Some(database) = cfg.database {

268 settings = settings.set_db(i64::from(database));

269 }

130270 if let Some(username) = cfg.username.as_deref().filter(|value| !value.is_empty()) {

131271 settings = settings.set_username(username);

132272 }

@@ -135,3 +275,30 @@

135275 }

136276 Ok(info.set_redis_settings(settings))

137277}

278

279fn master_for_slot(slots: &Value, slot: u16) -> Option<(String, u16)> {

280 let Value::Array(groups) = slots else {

281 return None;

282 };

283 for group in groups {

284 let Value::Array(fields) = group else {

285 continue;

286 };

287 let (Some(Value::Int(start)), Some(Value::Int(end))) = (fields.first(), fields.get(1))

288 else {

289 continue;

290 };

291 if !(*start..=*end).contains(&(slot as i64)) {

292 continue;

293 }

294 let Some(Value::Array(nodes)) = fields.get(2) else {

295 continue;

296 };

297 let (Some(Value::BulkString(host)), Some(Value::Int(port))) = (nodes.first(), nodes.get(1))

298 else {

299 continue;

300 };

301 return Some((String::from_utf8_lossy(host).into_owned(), *port as u16));

302 }

303 None

304}

Modified lib/infra/cache/src/exec.rs +3 −3 View lib/infra/cache/src/exec.rs

@@ -36,7 +36,7 @@

3636impl ConnectionLike for CacheConnection {

3737 fn req_packed_command<'a>(&'a mut self, cmd: &'a Cmd) -> RedisFuture<'a, Value> {

3838 match self {

39 Self::Standard(connection) => connection.req_packed_command(cmd),

39 Self::Standalone(connection) => connection.req_packed_command(cmd),

4040 Self::Cluster(connection) => connection.req_packed_command(cmd),

4141 }

4242 }

@@ -48,14 +48,14 @@

4848 count: usize,

4949 ) -> RedisFuture<'a, Vec<Value>> {

5050 match self {

51 Self::Standard(connection) => connection.req_packed_commands(pipeline, offset, count),

51 Self::Standalone(connection) => connection.req_packed_commands(pipeline, offset, count),

5252 Self::Cluster(connection) => connection.req_packed_commands(pipeline, offset, count),

5353 }

5454 }

5555

5656 fn get_db(&self) -> i64 {

5757 match self {

58 Self::Standard(connection) => connection.get_db(),

58 Self::Standalone(connection) => connection.get_db(),

5959 Self::Cluster(connection) => connection.get_db(),

6060 }

6161 }

Modified lib/infra/cache/src/kv.rs +6 −2 View lib/infra/cache/src/kv.rs

@@ -16,11 +16,15 @@

1616impl Cache {

1717 pub async fn connect(cfg: &RedisConfig) -> CacheResult<Self> {

1818 let conn = CacheConn::init(cfg).await?;

19 Ok(Self {

19 Ok(Self::from_conn(conn, cfg))

20 }

21

22 pub fn from_conn(conn: CacheConn, cfg: &RedisConfig) -> Self {

23 Self {

2024 conn,

2125 prefix: cfg.key_prefix().to_string(),

2226 default_ttl: cfg.default_ttl(),

23 })

27 }

2428 }

2529

2630 pub fn conn(&self) -> &CacheConn {

Modified lib/infra/cache/src/lib.rs +1 −1 View lib/infra/cache/src/lib.rs

@@ -15,7 +15,7 @@

1515mod kv;

1616

1717pub use client::{AppRedis, RedisConnection};

18pub use conn::{CacheConn, CacheConnection};

18pub use conn::{BlockingConnection, CacheConn, CacheConnection};

1919pub use error::{CacheError, CacheResult};

2020pub use keys::{CacheKey, SEPARATOR, join, join_tagged, same_slot, slot_tag};

2121pub use kv::Cache;

Modified lib/infra/config/src/redis.rs +19 −61 View lib/infra/config/src/redis.rs

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

77pub enum RedisMode {

88 #[default]

99 Standalone,

10 Sentinel,

1110 Cluster,

1211 ValkeyStandalone,

1312}

@@ -16,8 +15,6 @@

1615 pub fn parse(value: &str) -> Option<Self> {

1716 if value.eq_ignore_ascii_case("standalone") {

1817 Some(Self::Standalone)

19 } else if value.eq_ignore_ascii_case("sentinel") {

20 Some(Self::Sentinel)

2118 } else if value.eq_ignore_ascii_case("cluster") {

2219 Some(Self::Cluster)

2320 } else if value.eq_ignore_ascii_case("valkey_standalone")

@@ -81,7 +78,13 @@

8178 self.default_ttl_secs.map(Duration::from_secs)

8279 }

8380

84 pub fn seeds(&self) -> Result<Vec<(String, u16)>, String> {

81 /// Full connection URLs for every configured node, in configuration order.

82 ///

83 /// Scheme, embedded credentials, path and IPv6 brackets are preserved so that

84 /// every consumer derives the same connection parameters from the same file.

85 /// Nodes without a scheme are treated as `host[:port]` and default to

86 /// `redis://`.

87 pub fn node_urls(&self) -> Result<Vec<String>, String> {

8588 let url = self

8689 .url

8790 .as_deref()

@@ -117,28 +120,14 @@

117120 .ok_or_else(|| "redis.url must be set".to_string())?;

118121 match self.mode() {

119122 RedisMode::Standalone | RedisMode::ValkeyStandalone => {

120 if !(url.starts_with("redis://") || url.starts_with("rediss://")) {

123 if !has_scheme(url.trim()) || url.contains(',') {

121124 return Err(format!(

122 "redis.url must start with redis:// or rediss://, got {url:?}"

125 "redis.url must be a single redis:// or rediss:// node, got {url:?}"

123126 ));

124127 }

125128 }

126129 RedisMode::Cluster => {

127 if !(url.starts_with("redis://")

128 || url.starts_with("rediss://")

129 || url.contains(","))

130 {

131 return Err(format!(

132 "redis cluster url must start with redis:// or rediss://, or list multiple hosts comma-separated, got {url:?}"

133 ));

134 }

135 }

136 RedisMode::Sentinel => {

137 if !(url.starts_with("redis://") || url.starts_with("rediss://")) {

138 return Err(format!(

139 "redis sentinel url must start with redis:// or rediss://, got {url:?}"

140 ));

141 }

130 self.node_urls()?;

142131 }

143132 }

144133 Ok(url)

@@ -149,49 +138,18 @@

149138 const KEY: &'static str = "redis";

150139}

151140

152fn parse_node(node: &str) -> Result<(String, u16), String> {

141fn has_scheme(node: &str) -> bool {

142 node.starts_with("redis://") || node.starts_with("rediss://")

143}

144

145fn parse_node(node: &str) -> Result<String, String> {

153146 let trimmed = node.trim();

154147 if trimmed.is_empty() {

155148 return Err("redis.url contains an empty entry".to_string());

156149 }

157 let without_scheme = trimmed

158 .strip_prefix("redis://")

159 .or_else(|| trimmed.strip_prefix("rediss://"))

160 .unwrap_or(trimmed);

161 let authority = without_scheme

162 .split(['/', '?'])

163 .next()

164 .unwrap_or(without_scheme);

165 let authority = match authority.rsplit_once('@') {

166 Some((_, host)) => host,

167 None => authority,

168 };

169

170 if let Some(rest) = authority.strip_prefix('[') {

171 let (host, tail) = rest

172 .split_once(']')

173 .ok_or_else(|| format!("redis node {node:?} has an unterminated IPv6 address"))?;

174 let port = match tail.strip_prefix(':') {

175 Some(port) => parse_port(port, node)?,

176 None => 6379,

177 };

178 return Ok((host.to_string(), port));

179 }

180

181 let colons = authority.matches(':').count();

182 if colons == 0 {

183 return Ok((authority.to_string(), 6379));

184 }

185 if colons == 1 {

186 let (host, port) = authority

187 .rsplit_once(':')

188 .ok_or_else(|| format!("redis node {node:?} is invalid"))?;

189 return Ok((host.to_string(), parse_port(port, node)?));

150 if has_scheme(trimmed) {

151 Ok(trimmed.to_string())

152 } else {

153 Ok(format!("redis://{trimmed}"))

190154 }

191 Ok((authority.to_string(), 6379))

192}

193

194fn parse_port(port: &str, node: &str) -> Result<u16, String> {

195 port.parse::<u16>()

196 .map_err(|_| format!("redis node {node:?} has an invalid port {port:?}"))

197155}

Modified lib/infra/internal/src/nonce.rs +5 −1 View lib/infra/internal/src/nonce.rs

@@ -9,11 +9,15 @@

99}

1010

1111impl NonceStore {

12 pub fn from_conn(conn: CacheConn) -> Self {

13 Self { conn }

14 }

15

1216 pub async fn connect(config: &RedisConfig, _prefix: &str) -> Result<Self, InternalError> {

1317 let conn = CacheConn::init(config)

1418 .await

1519 .map_err(|error| InternalError::Config(format!("redis: {error}")))?;

16 Ok(Self { conn })

20 Ok(Self::from_conn(conn))

1721 }

1822

1923 /// `SET key 1 NX EX ttl`：第一次出现返回 true，重放返回 false。

Modified lib/service/push/src/worker.rs +69 −41 View lib/service/push/src/worker.rs

@@ -56,58 +56,86 @@

5656

5757 pub async fn run(self) {

5858 let topic = crate::protocol::QUEUE_TOPIC.to_string();

59 loop {

60 let payload = tokio::select! {

61 _ = self.cancel.cancelled() => break,

62 result = self.state.redis.brpop_bytes(&topic, CONSUME_TIMEOUT_SECS) => {

63 match result {

64 Ok(Some(payload)) => payload,

65 Ok(None) => continue,

66 Err(error) => {

67 tracing::warn!(%error, "push queue consume failed");

68 tokio::time::sleep(Duration::from_secs(2)).await;

69 continue;

59 'consume: loop {

60 let Some(mut consumer) = self.connect(&topic).await else {

61 return;

62 };

63 loop {

64 let outcome = tokio::select! {

65 _ = self.cancel.cancelled() => break 'consume,

66 result = consumer.brpop_bytes(&topic, CONSUME_TIMEOUT_SECS) => result,

67 };

68 let payload = match outcome {

69 Ok(Some(payload)) => payload,

70 Ok(None) => continue,

71 Err(error) => {

72 tracing::warn!(%error, "push queue consume failed");

73 if !self.backoff().await {

74 break 'consume;

7075 }

76 continue 'consume;

7177 }

72 }

73 };

74

75 let job = match PushJob::decode(&payload) {

76 Ok(job) => job,

77 Err(error) => {

78 tracing::warn!(%error, "dropping malformed push job");

79 crate::protocol::enqueue_dlq(&self.state.redis, &payload).await;

80 continue;

81 }

82 };

78 };

79 let job = match PushJob::decode(&payload) {

80 Ok(job) => job,

81 Err(error) => {

82 tracing::warn!(%error, "dropping malformed push job");

83 crate::protocol::enqueue_dlq(&self.state.redis, &payload).await;

84 continue;

85 }

86 };

8387

84 let span = tracing::info_span!(

85 "push.delivery",

86 user = %job.user_id,

87 category = job.category.as_str(),

88 request_id = tracing::field::Empty

89 );

90 if let Some(request_id) = job.request_id.as_deref() {

91 span.record("request_id", request_id);

92 }

93 oltp::propagation::set_parent_from_traceparent(&span, job.trace.as_deref());

94 if let Err(error) = api::middleware::with_request_id(

95 job.request_id.as_deref(),

96 self.handle(&job).instrument(span),

97 )

98 .await

99 {

100 tracing::warn!(

101 %error,

88 let span = tracing::info_span!(

89 "push.delivery",

10290 user = %job.user_id,

10391 category = job.category.as_str(),

104 "push job failed"

92 request_id = tracing::field::Empty

10593 );

94 if let Some(request_id) = job.request_id.as_deref() {

95 span.record("request_id", request_id);

96 }

97 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

103 {

104 tracing::warn!(

105 %error,

106 user = %job.user_id,

107 category = job.category.as_str(),

108 "push job failed"

109 );

110 }

106111 }

107112 }

108113 tracing::debug!("push worker stopped");

109114 }

110115

116 /// Opens the queue consumer, retrying until it succeeds or shutdown is requested.

117 async fn connect(&self, topic: &str) -> Option<cache::BlockingConnection> {

118 loop {

119 match self.state.redis.blocking(topic).await {

120 Ok(consumer) => return Some(consumer),

121 Err(error) => {

122 tracing::warn!(%error, "push queue consumer connect failed");

123 if !self.backoff().await {

124 return None;

125 }

126 }

127 }

128 }

129 }

130

131 /// Waits before a reconnect; false when shutdown was requested first.

132 async fn backoff(&self) -> bool {

133 tokio::select! {

134 _ = self.cancel.cancelled() => false,

135 _ = tokio::time::sleep(Duration::from_secs(2)) => true,

136 }

137 }

138

111139 /// 一条任务的完整流程：闸门 → 组装 → 逐订阅投递。

112140 async fn handle(&self, job: &PushJob) -> Result<(), PushError> {

113141 if self.skip_by_preferences(job).await? {

Modified lib/share/events/src/gateway/subscriptions.rs +5 −0 View lib/share/events/src/gateway/subscriptions.rs

@@ -33,6 +33,11 @@

3333 Self { connection }

3434 }

3535

36 /// Reuses an already established pool instead of opening another one.

37 pub fn from_connection(connection: RedisConnection) -> Self {

38 Self { connection }

39 }

40

3641 pub async fn connect(config: &RedisConfig) -> Result<Self, ActorError> {

3742 Ok(Self::new(connect_redis_single(config).await?))

3843 }