ecat_data_sqlx/
routing.rs

1// Copyright (c) 2026 erik <erik@erik.xyz> — https://erik.xyz
2//! 读写分离路由的配置与构建器：一份主库配置 + 零或多份副本配置 →
3//! [`RdbmsRouting`]。
4
5use std::sync::Arc;
6use std::time::Duration;
7
8use ecat_data::{RdbmsClient, RdbmsRouting};
9use serde::Deserialize;
10
11use crate::{SqlxClient, SqlxConfig};
12
13/// `fallback_to_primary` 的默认值。
14fn default_true() -> bool {
15    true
16}
17
18/// 读写分离配置。**未知键直接报错**（拼错 `replicas` 会静默退化成单主库）。
19#[derive(Debug, Clone, Deserialize)]
20#[serde(deny_unknown_fields)]
21pub struct SqlxRoutingConfig {
22    pub primary: SqlxConfig,
23    #[serde(default)]
24    pub replicas: Vec<SqlxConfig>,
25    /// 副本实测延迟超过它即跳过；`None` = 不探测。
26    #[serde(default)]
27    pub max_replica_lag_secs: Option<u64>,
28    /// 副本全不可用时是否降级读主，默认 `true`。
29    #[serde(default = "default_true")]
30    pub fallback_to_primary: bool,
31}
32
33/// 按配置逐份建连并组装路由。任何一份连不上都直接返回该错误（不静默丢弃）。
34pub async fn build_routing(cfg: &SqlxRoutingConfig) -> Result<RdbmsRouting, sqlx::Error> {
35    let primary = Arc::new(SqlxClient::from_config(cfg.primary.clone()).await?);
36    let mut replicas: Vec<Arc<dyn RdbmsClient>> = Vec::with_capacity(cfg.replicas.len());
37    for replica in &cfg.replicas {
38        replicas.push(Arc::new(SqlxClient::from_config(replica.clone()).await?));
39    }
40    let mut routing =
41        RdbmsRouting::new(primary, replicas).fallback_to_primary(cfg.fallback_to_primary);
42    if let Some(secs) = cfg.max_replica_lag_secs {
43        routing = routing.max_replica_lag(Duration::from_secs(secs));
44    }
45    Ok(routing)
46}
47
48#[cfg(test)]
49mod tests;
ecat_data_sqlx/routing.rs

ecat_data_sqlx/
routing.rs
