fix(redis): make blocking reads and sync queue work with cluster mode

The deployment runs a 3-master valkey cluster, but two code paths assumed a
single endpoint:

- AppRedis::connection_clone opened multiplexed connections without an
  explicit connection config, so blocking commands such as BRPOP were judged
  by the default response timeout and surfaced spurious timeout errors.
  Disable the response timeout for those connections and reuse the
  configured connection timeout.
- SyncQueue round-robined BRPOP over one connection per consumer, which
  cannot work in cluster mode: a blocking pop must target the master that
  owns the key's slot or it receives MOVED. Open one connection per master
  per consumer, resolve the slot with CLUSTER KEYSLOT and pick the owning
  connection. Standalone behaviour is unchanged.

AppRedis also exposes cluster_slot_ranges() so callers can map a slot to the
master index that matches cluster_master_connections() ordering, and the
single-node URL used by the consumer connections is now taken from the first
entry of the comma-separated cluster list, which Client::open cannot parse.

ZhenYi committed 1 month ago

parent 080f579 commit 3b76bf0 Browse files

File tree 2

redis.rs Modified
queue.rs Modified
No files match.

lib/export/src/redis.rs +69 −1 View

@@ -266,8 +266,13 @@

266266 info = info.set_addr(addr);

267267

268268 let client = redis::Client::open(info).map_err(ExportError::from)?;

269 // BRPOP 之类的阻塞命令会持有连接直到超时；必须关闭响应超时，

270 // 否则默认超时会让阻塞读被误判为 timeout 错误。

271 let conn_config = redis::AsyncConnectionConfig::new()

272 .set_response_timeout(None)

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

269274 let connection = client

270 .get_multiplexed_async_connection()

275 .get_multiplexed_async_connection_with_config(&conn_config)

271276 .await

272277 .map_err(ExportError::from)?;

273278

@@ -276,6 +281,27 @@

276281

277282 Ok(connections)

278283 }

284

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

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

287 ///

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

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

290 /// pre-established connection without reconnecting.

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

292 let mut conn = self.connection_clone();

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

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

295 };

296

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

298 .arg("SLOTS")

299 .query_async(cluster)

300 .await

301 .map_err(ExportError::from)?;

302

303 Ok(cluster_slot_ranges(&slots))

304 }

279305}

280306

281307/// Extracts the host/port of every master from a `CLUSTER SLOTS` response.

@@ -411,3 +437,45 @@

411437 .await

412438 })

413439}

440

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

442///

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

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

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

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

447 return Vec::new();

448 };

449

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

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

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

453

454 for group in groups {

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

456 continue;

457 };

458

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

460 else {

461 continue;

462 };

463

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

465 continue;

466 };

467

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

469 else {

470 continue;

471 };

472

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

474 let next_index = seen.len();

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

476

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

478 }

479

480 ranges

481}

lib/git/src/sync/queue.rs +100 −1 View

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

1818pub struct SyncQueue {

1919 redis: AppRedis,

2020 consumer_conns: Vec<MultiplexedConnection>,

21 /// cluster 模式：每个消费者一组到所有 master 的连接（按 slot 选择）

22 cluster_consumer_groups: Vec<Vec<MultiplexedConnection>>,

2123}

2224

2325impl SyncQueue {

@@ -26,12 +28,48 @@

2628 Self {

2729 redis,

2830 consumer_conns: Vec::new(),

31 cluster_consumer_groups: Vec::new(),

2932 }

3033 }

3134

3235 pub async fn with_consumer_conns(redis: AppRedis, count: usize) -> SyncResult<Self> {

3336 let count = count.max(1);

38 // cluster 模式：BRPOP 是对单个 key 的操作，必须打到该 key slot 的 master，

39 // 否则会收到 MOVED。这里按哨兵轮询的方式给每个消费者预建到所有 master 的连接，

40 // 消费时再按 slot 选择对应连接。standalone 模式维持单连接。

3441 let mut consumer_conns = Vec::with_capacity(count);

42 if redis.is_cluster() {

43 let masters = redis.cluster_master_connections().await.map_err(|e| {

44 SyncError::Internal(format!("open cluster master connections: {e}"))

45 })?;

46 if masters.is_empty() {

47 return Err(SyncError::Internal(

48 "redis cluster reported no master nodes".to_string(),

49 ));

50 }

51 // 每个消费者一组 master 连接；取模轮询避免所有消费者压在同一节点

52 let mut groups = Vec::with_capacity(count);

53 for i in 0..count {

54 let mut group = Vec::with_capacity(masters.len());

55 for offset in 0..masters.len() {

56 let idx = (i + offset) % masters.len();

57 group.push(masters[idx].clone());

58 }

59 groups.push(group);

60 }

61 tracing::debug!(

62 count,

63 masters = masters.len(),

64 "sync queue cluster consumer connections opened"

65 );

66 return Ok(Self {

67 redis,

68 consumer_conns: Vec::new(),

69 cluster_consumer_groups: groups,

70 });

71 }

3573 for _ in 0..count {

3674 let conn = open_multiplexed(&redis).await?;

3775 consumer_conns.push(conn);

@@ -40,6 +78,7 @@

4078 Ok(Self {

4179 redis,

4280 consumer_conns,

81 cluster_consumer_groups: Vec::new(),

4382 })

4483 }

4584

@@ -64,6 +103,16 @@

64103 dispatch!(conn, |c| {

65104 redis::cmd("RPOP").arg(key).query_async(c).await

66105 })?

106 } else if !self.cluster_consumer_groups.is_empty() {

107 // cluster：BRPOP 必须落在持有该 key slot 的 master 上，否则会 MOVED。

108 // 先算 KEYSLOT，再到负责该 slot 的 master 上消费。

109 let conn = self.cluster_master_for_key(key).await?;

110 let mut conn = conn;

111 redis::cmd("BRPOP")

112 .arg(key)

113 .arg(timeout_secs)

114 .query_async(&mut conn)

115 .await?

67116 } else if let Some(conn) = self.consumer_conns.get(slot).cloned() {

68117 let mut conn = conn;

69118 redis::cmd("BRPOP")

@@ -125,6 +174,47 @@

125174 Ok(n)

126175 }

127176

177 /// 计算 key 所属 slot，并返回负责该 slot 的 master 连接（cluster 模式）。

178 async fn cluster_master_for_key(&self, key: &str) -> SyncResult<MultiplexedConnection> {

179 let group = self

180 .cluster_consumer_groups

181 .first()

182 .ok_or_else(|| SyncError::Internal("no cluster consumer connections".to_string()))?;

183

184 let mut conn = self.redis.connection_clone();

185 let key_slot: u16 = {

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

187 return Err(SyncError::Internal(

188 "cluster master lookup on non-cluster connection".to_string(),

189 ));

190 };

191 redis::cmd("CLUSTER")

192 .arg("KEYSLOT")

193 .arg(key)

194 .query_async(cluster)

195 .await

196 .map_err(|e| SyncError::Internal(format!("CLUSTER KEYSLOT failed: {e}")))?

197 };

198

199 let ranges = self

200 .redis

201 .cluster_slot_ranges()

202 .await

203 .map_err(|e| SyncError::Internal(format!("cluster slot ranges: {e}")))?;

204

205 let index = ranges

206 .iter()

207 .find(|(start, end, _)| (*start..=*end).contains(&(key_slot as i64)))

208 .map(|(_, _, idx)| *idx)

209 .ok_or_else(|| SyncError::Internal(format!("no master owns slot {key_slot}")))?;

210

211 group

212 .get(index)

213 .or_else(|| group.first())

214 .cloned()

215 .ok_or_else(|| SyncError::Internal("cluster consumer connection missing".to_string()))

216 }

217

128218 pub async fn delete(&self, key: &str) -> SyncResult<()> {

129219 let mut conn = self.redis.connection_clone();

130220 let _: i64 = dispatch!(conn, |c| {

@@ -135,7 +225,16 @@

135225}

136226

137227async fn open_multiplexed(redis: &AppRedis) -> SyncResult<MultiplexedConnection> {

138 let url = redis.config().url().to_string();

228 // cluster 模式配置是逗号分隔的节点列表；BRPOP 消费者连单个节点即可

229 // （`redis::Client::open` 无法解析多节点串）。

230 let url = redis

231 .config()

232 .url()

233 .split(',')

234 .map(str::trim)

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

236 .ok_or_else(|| SyncError::Internal("redis url contains no nodes".to_string()))?

237 .to_string();

139238 let client = redis::Client::open(url.as_str())

140239 .map_err(|e| SyncError::Internal(format!("open redis client for consumer: {e}")))?;

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