use crate::{ProxyError, Result};
use serde::{Deserialize, Serialize};
use std::path::Path;
use std::time::Duration;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
#[serde(rename_all = "lowercase")]
pub enum PoolingMode {
#[default]
Session,
Transaction,
Statement,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
#[serde(rename_all = "lowercase")]
pub enum PreparedStatementMode {
#[default]
Disable,
Track,
Named,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PoolModeConfig {
#[serde(default)]
pub mode: PoolingMode,
#[serde(default = "default_pool_mode_max_size")]
pub max_pool_size: u32,
#[serde(default = "default_pool_mode_min_idle")]
pub min_idle: u32,
#[serde(default = "default_pool_mode_idle_timeout")]
pub idle_timeout_secs: u64,
#[serde(default = "default_pool_mode_max_lifetime")]
pub max_lifetime_secs: u64,
#[serde(default = "default_pool_mode_acquire_timeout")]
pub acquire_timeout_secs: u64,
#[serde(default = "default_reset_query")]
pub reset_query: String,
#[serde(default)]
pub prepared_statement_mode: PreparedStatementMode,
#[serde(default)]
pub skip_clean_reset: bool,
}
fn default_pool_mode_max_size() -> u32 {
100
}
fn default_pool_mode_min_idle() -> u32 {
10
}
fn default_pool_mode_idle_timeout() -> u64 {
600
}
fn default_pool_mode_max_lifetime() -> u64 {
3600
}
fn default_pool_mode_acquire_timeout() -> u64 {
5
}
fn default_reset_query() -> String {
"DISCARD ALL".to_string()
}
impl Default for PoolModeConfig {
fn default() -> Self {
Self {
mode: PoolingMode::default(),
max_pool_size: default_pool_mode_max_size(),
min_idle: default_pool_mode_min_idle(),
idle_timeout_secs: default_pool_mode_idle_timeout(),
max_lifetime_secs: default_pool_mode_max_lifetime(),
acquire_timeout_secs: default_pool_mode_acquire_timeout(),
reset_query: default_reset_query(),
prepared_statement_mode: PreparedStatementMode::default(),
skip_clean_reset: false,
}
}
}
impl PoolModeConfig {
pub fn session_mode() -> Self {
Self {
mode: PoolingMode::Session,
prepared_statement_mode: PreparedStatementMode::Named,
..Default::default()
}
}
pub fn transaction_mode() -> Self {
Self {
mode: PoolingMode::Transaction,
prepared_statement_mode: PreparedStatementMode::Track,
..Default::default()
}
}
pub fn statement_mode() -> Self {
Self {
mode: PoolingMode::Statement,
prepared_statement_mode: PreparedStatementMode::Disable,
..Default::default()
}
}
pub fn idle_timeout(&self) -> Duration {
Duration::from_secs(self.idle_timeout_secs)
}
pub fn max_lifetime(&self) -> Duration {
Duration::from_secs(self.max_lifetime_secs)
}
pub fn acquire_timeout(&self) -> Duration {
Duration::from_secs(self.acquire_timeout_secs)
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ProxyConfig {
pub listen_address: String,
pub admin_address: String,
#[serde(default)]
pub admin_token: Option<String>,
#[serde(default)]
pub admin_allow_insecure: bool,
pub tr_enabled: bool,
pub tr_mode: TrMode,
pub pool: PoolConfig,
#[serde(default)]
pub pool_mode: PoolModeConfig,
pub load_balancer: LoadBalancerConfig,
pub health: HealthConfig,
pub nodes: Vec<NodeConfig>,
pub tls: Option<TlsConfig>,
#[serde(default = "default_write_timeout_secs")]
pub write_timeout_secs: u64,
#[serde(default)]
pub plugins: PluginToml,
#[serde(default)]
pub hba: Vec<HbaRule>,
#[serde(default)]
pub auth: AuthConfig,
#[serde(default)]
pub mcp: McpConfig,
#[serde(default)]
pub agent_contracts: Vec<crate::agent_contract::AgentContract>,
#[serde(default)]
pub http_gateway: HttpGatewayConfig,
#[serde(default)]
pub mirror: MirrorConfig,
#[serde(default)]
pub edge: crate::edge::EdgeConfig,
#[serde(default)]
pub branch: BranchConfig,
#[serde(default)]
pub routing_hints: RoutingHintsConfig,
#[serde(default)]
pub rate_limit: RateLimitToml,
#[serde(default)]
pub circuit_breaker: CircuitBreakerToml,
#[serde(default)]
pub analytics: AnalyticsToml,
#[serde(default)]
pub lag_routing: LagRoutingToml,
#[serde(default)]
pub cache: CacheToml,
#[serde(default)]
pub query_rewrite: QueryRewriteToml,
#[serde(default)]
pub multi_tenancy: MultiTenancyToml,
#[serde(default)]
pub schema_routing: SchemaRoutingToml,
#[serde(default)]
pub graphql_gateway: GraphqlGatewayConfig,
#[serde(default = "default_true")]
pub optimize_unnamed_parse: bool,
#[serde(default = "default_drain_timeout_secs")]
pub shutdown_drain_timeout_secs: u64,
}
fn default_drain_timeout_secs() -> u64 {
60
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BranchConfig {
#[serde(default)]
pub enabled: bool,
#[serde(default = "default_localhost")]
pub backend_host: String,
#[serde(default = "default_pg_port")]
pub backend_port: u16,
#[serde(default = "default_pg_user")]
pub admin_user: String,
pub admin_password: Option<String>,
#[serde(default = "default_admin_db")]
pub admin_database: String,
#[serde(default = "default_admin_db")]
pub base_database: String,
}
impl Default for BranchConfig {
fn default() -> Self {
Self {
enabled: false,
backend_host: default_localhost(),
backend_port: default_pg_port(),
admin_user: default_pg_user(),
admin_password: None,
admin_database: default_admin_db(),
base_database: default_admin_db(),
}
}
}
fn default_admin_db() -> String {
"postgres".to_string()
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MirrorConfig {
#[serde(default)]
pub enabled: bool,
#[serde(default = "default_sample_rate")]
pub sample_rate: f64,
#[serde(default = "default_true_bool")]
pub writes_only: bool,
#[serde(default = "default_mirror_queue")]
pub queue_size: usize,
#[serde(default = "default_localhost")]
pub backend_host: String,
#[serde(default = "default_pg_port")]
pub backend_port: u16,
#[serde(default = "default_pg_user")]
pub backend_user: String,
pub backend_password: Option<String>,
pub backend_database: Option<String>,
#[serde(default = "default_localhost")]
pub source_host: String,
#[serde(default = "default_pg_port")]
pub source_port: u16,
#[serde(default = "default_pg_user")]
pub source_user: String,
pub source_password: Option<String>,
pub source_database: Option<String>,
}
impl Default for MirrorConfig {
fn default() -> Self {
Self {
enabled: false,
sample_rate: 1.0,
writes_only: true,
queue_size: 10_000,
backend_host: default_localhost(),
backend_port: default_pg_port(),
backend_user: default_pg_user(),
backend_password: None,
backend_database: None,
source_host: default_localhost(),
source_port: default_pg_port(),
source_user: default_pg_user(),
source_password: None,
source_database: None,
}
}
}
fn default_sample_rate() -> f64 {
1.0
}
fn default_mirror_queue() -> usize {
10_000
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HttpGatewayConfig {
#[serde(default)]
pub enabled: bool,
#[serde(default = "default_http_gw_listen")]
pub listen_address: String,
#[serde(default = "default_localhost")]
pub backend_host: String,
#[serde(default = "default_pg_port")]
pub backend_port: u16,
#[serde(default = "default_pg_user")]
pub backend_user: String,
pub backend_password: Option<String>,
pub backend_database: Option<String>,
#[serde(default)]
pub auth_token: Option<String>,
}
impl Default for HttpGatewayConfig {
fn default() -> Self {
Self {
enabled: false,
listen_address: default_http_gw_listen(),
backend_host: default_localhost(),
backend_port: default_pg_port(),
backend_user: default_pg_user(),
backend_password: None,
backend_database: None,
auth_token: None,
}
}
}
fn default_http_gw_listen() -> String {
"127.0.0.1:9093".to_string()
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct McpConfig {
#[serde(default)]
pub enabled: bool,
#[serde(default = "default_mcp_listen")]
pub listen_address: String,
#[serde(default = "default_localhost")]
pub backend_host: String,
#[serde(default = "default_pg_port")]
pub backend_port: u16,
#[serde(default = "default_pg_user")]
pub backend_user: String,
pub backend_password: Option<String>,
pub backend_database: Option<String>,
#[serde(default = "default_true_bool")]
pub read_only: bool,
#[serde(default)]
pub contract: Option<String>,
#[serde(default)]
pub auth_token: Option<String>,
}
impl Default for McpConfig {
fn default() -> Self {
Self {
enabled: false,
listen_address: default_mcp_listen(),
backend_host: default_localhost(),
backend_port: default_pg_port(),
backend_user: default_pg_user(),
backend_password: None,
backend_database: None,
read_only: true,
contract: None,
auth_token: None,
}
}
}
fn default_mcp_listen() -> String {
"127.0.0.1:9092".to_string()
}
fn default_localhost() -> String {
"127.0.0.1".to_string()
}
fn default_pg_port() -> u16 {
5432
}
fn default_pg_user() -> String {
"postgres".to_string()
}
fn default_true_bool() -> bool {
true
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct AuthConfig {
#[serde(default)]
pub mode: AuthMode,
#[serde(default)]
pub auth_file: Option<String>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
#[serde(rename_all = "lowercase")]
pub enum AuthMode {
#[default]
Passthrough,
Scram,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HbaRule {
pub action: HbaAction,
#[serde(default = "hba_all")]
pub user: String,
#[serde(default = "hba_all")]
pub database: String,
#[serde(default = "hba_all")]
pub address: String,
}
fn hba_all() -> String {
"all".to_string()
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum HbaAction {
Allow,
Reject,
}
fn default_write_timeout_secs() -> u64 {
30 }
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[serde(default)]
pub struct GqlTableToml {
pub name: String,
pub columns: Vec<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default)]
pub struct GraphqlGatewayConfig {
pub enabled: bool,
pub listen_address: String,
pub backend_host: String,
pub backend_port: u16,
pub backend_user: String,
pub backend_password: Option<String>,
pub backend_database: Option<String>,
pub auth_token: Option<String>,
pub tables: Vec<GqlTableToml>,
}
impl Default for GraphqlGatewayConfig {
fn default() -> Self {
Self {
enabled: false,
listen_address: "0.0.0.0:9091".to_string(),
backend_host: "127.0.0.1".to_string(),
backend_port: 5432,
backend_user: "postgres".to_string(),
backend_password: None,
backend_database: None,
auth_token: None,
tables: Vec::new(),
}
}
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[serde(default)]
pub struct SchemaRoutingToml {
pub enabled: bool,
pub analytics_node: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default)]
pub struct MultiTenancyToml {
pub enabled: bool,
pub identify_by: String,
pub tenant_column: String,
pub tenant_tables: Vec<String>,
pub tenants: Vec<String>,
}
impl Default for MultiTenancyToml {
fn default() -> Self {
Self {
enabled: false,
identify_by: "application_name".to_string(),
tenant_column: "tenant_id".to_string(),
tenant_tables: Vec::new(),
tenants: Vec::new(),
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
#[serde(default)]
pub struct RewriteRuleToml {
pub match_table: Option<String>,
pub match_regex: Option<String>,
pub replace_table_with: Option<String>,
pub append_where: Option<String>,
pub add_limit: Option<u32>,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[serde(default)]
pub struct QueryRewriteToml {
pub enabled: bool,
pub rules: Vec<RewriteRuleToml>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default)]
pub struct CacheToml {
pub enabled: bool,
pub ttl_secs: u64,
pub max_result_bytes: usize,
}
impl Default for CacheToml {
fn default() -> Self {
Self {
enabled: false,
ttl_secs: 300,
max_result_bytes: 1024 * 1024,
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default)]
pub struct LagRoutingToml {
pub enabled: bool,
pub ryw_window_ms: u64,
pub max_lag_bytes: u64,
}
impl Default for LagRoutingToml {
fn default() -> Self {
Self {
enabled: false,
ryw_window_ms: 500,
max_lag_bytes: 0,
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default)]
pub struct AnalyticsToml {
pub enabled: bool,
pub slow_query_ms: u64,
pub max_fingerprints: u32,
}
impl Default for AnalyticsToml {
fn default() -> Self {
Self {
enabled: false,
slow_query_ms: 1000,
max_fingerprints: 10000,
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default)]
pub struct CircuitBreakerToml {
pub enabled: bool,
pub failure_threshold: u32,
pub open_secs: u64,
pub success_threshold: u32,
}
impl Default for CircuitBreakerToml {
fn default() -> Self {
Self {
enabled: false,
failure_threshold: 5,
open_secs: 10,
success_threshold: 3,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
#[serde(rename_all = "snake_case")]
pub enum RateLimitKeyBy {
#[default]
User,
ClientIp,
Database,
Global,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default)]
pub struct RateLimitToml {
pub enabled: bool,
pub default_qps: u32,
pub default_burst: u32,
pub max_concurrent: u32,
pub key_by: RateLimitKeyBy,
}
impl Default for RateLimitToml {
fn default() -> Self {
Self {
enabled: false,
default_qps: 1000,
default_burst: 2000,
max_concurrent: 0,
key_by: RateLimitKeyBy::User,
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default)]
pub struct RoutingHintsConfig {
pub enabled: bool,
pub strip_hints: bool,
}
impl Default for RoutingHintsConfig {
fn default() -> Self {
Self {
enabled: false,
strip_hints: true,
}
}
}
impl Default for ProxyConfig {
fn default() -> Self {
Self {
listen_address: "0.0.0.0:5432".to_string(),
admin_address: "127.0.0.1:9090".to_string(),
admin_token: None,
admin_allow_insecure: false,
tr_enabled: true,
tr_mode: TrMode::Session,
pool: PoolConfig::default(),
pool_mode: PoolModeConfig::default(),
load_balancer: LoadBalancerConfig::default(),
health: HealthConfig::default(),
nodes: Vec::new(),
tls: None,
write_timeout_secs: default_write_timeout_secs(),
plugins: PluginToml::default(),
hba: Vec::new(),
auth: AuthConfig::default(),
mcp: McpConfig::default(),
agent_contracts: Vec::new(),
http_gateway: HttpGatewayConfig::default(),
mirror: MirrorConfig::default(),
edge: crate::edge::EdgeConfig::default(),
branch: BranchConfig::default(),
routing_hints: RoutingHintsConfig::default(),
rate_limit: RateLimitToml::default(),
circuit_breaker: CircuitBreakerToml::default(),
analytics: AnalyticsToml::default(),
lag_routing: LagRoutingToml::default(),
cache: CacheToml::default(),
query_rewrite: QueryRewriteToml::default(),
multi_tenancy: MultiTenancyToml::default(),
schema_routing: SchemaRoutingToml::default(),
graphql_gateway: GraphqlGatewayConfig::default(),
optimize_unnamed_parse: true,
shutdown_drain_timeout_secs: default_drain_timeout_secs(),
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PluginToml {
#[serde(default)]
pub enabled: bool,
#[serde(default = "default_plugin_dir")]
pub plugin_dir: String,
#[serde(default)]
pub hot_reload: bool,
#[serde(default = "default_plugin_memory_mb")]
pub memory_limit_mb: usize,
#[serde(default = "default_plugin_timeout_ms")]
pub timeout_ms: u64,
#[serde(default = "default_plugin_max")]
pub max_plugins: usize,
#[serde(default = "default_true")]
pub fuel_metering: bool,
#[serde(default = "default_plugin_fuel")]
pub fuel_limit: u64,
#[serde(default)]
pub trust_root: Option<String>,
}
fn default_plugin_dir() -> String {
"/etc/heliosproxy/plugins".to_string()
}
fn default_plugin_memory_mb() -> usize {
64
}
fn default_plugin_timeout_ms() -> u64 {
100
}
fn default_plugin_max() -> usize {
20
}
fn default_true() -> bool {
true
}
fn default_plugin_fuel() -> u64 {
1_000_000
}
impl Default for PluginToml {
fn default() -> Self {
Self {
enabled: false,
plugin_dir: default_plugin_dir(),
hot_reload: false,
memory_limit_mb: default_plugin_memory_mb(),
timeout_ms: default_plugin_timeout_ms(),
max_plugins: default_plugin_max(),
fuel_metering: true,
fuel_limit: default_plugin_fuel(),
trust_root: None,
}
}
}
fn substitute_env(text: &str) -> Result<String> {
let mut out = String::with_capacity(text.len());
for line in text.split_inclusive('\n') {
if line.trim_start().starts_with('#') {
out.push_str(line);
} else {
substitute_line(line, &mut out)?;
}
}
Ok(out)
}
fn substitute_line(line: &str, out: &mut String) -> Result<()> {
let mut rest = line;
while let Some(idx) = rest.find("${") {
out.push_str(&rest[..idx]);
let body = &rest[idx + 2..]; match parse_placeholder(body)? {
Some((value, consumed)) => {
out.push_str(&value);
rest = &body[consumed..];
}
None => {
out.push_str("${");
rest = body;
}
}
}
out.push_str(rest);
Ok(())
}
fn parse_placeholder(body: &str) -> Result<Option<(String, usize)>> {
let bytes = body.as_bytes();
let mut n = 0;
while n < bytes.len() {
let b = bytes[n];
let valid = if n == 0 {
b.is_ascii_alphabetic() || b == b'_'
} else {
b.is_ascii_alphanumeric() || b == b'_'
};
if valid {
n += 1;
} else {
break;
}
}
if n == 0 {
return Ok(None); }
let name = &body[..n];
let after = &body[n..];
if after.starts_with('}') {
match std::env::var(name) {
Ok(v) => Ok(Some((v, n + 1))),
Err(_) => Err(ProxyError::Config(format!(
"config env-var substitution: `${{{name}}}` references environment \
variable `{name}`, which is not set (and no `:-default` fallback \
was given)"
))),
}
} else if let Some(after_op) = after.strip_prefix(":-") {
match after_op.find('}') {
Some(end) => {
let default = &after_op[..end];
let value = std::env::var(name).unwrap_or_else(|_| default.to_string());
Ok(Some((value, n + 2 + end + 1)))
}
None => Ok(None), }
} else {
Ok(None) }
}
const KNOWN_TOP_LEVEL_KEYS: &[&str] = &[
"listen_address",
"admin_address",
"admin_token",
"admin_allow_insecure",
"tr_enabled",
"tr_mode",
"pool",
"pool_mode",
"load_balancer",
"health",
"nodes",
"tls",
"write_timeout_secs",
"plugins",
"hba",
"auth",
"mcp",
"agent_contracts",
"http_gateway",
"mirror",
"edge",
"branch",
"routing_hints",
"rate_limit",
"circuit_breaker",
"analytics",
"lag_routing",
"cache",
"query_rewrite",
"multi_tenancy",
"schema_routing",
"graphql_gateway",
"optimize_unnamed_parse",
"shutdown_drain_timeout_secs",
];
fn unknown_top_level_keys(text: &str) -> Vec<String> {
let Ok(value) = toml::from_str::<toml::Value>(text) else {
return Vec::new();
};
let Some(table) = value.as_table() else {
return Vec::new();
};
let mut unknown: Vec<String> = table
.keys()
.filter(|k| !KNOWN_TOP_LEVEL_KEYS.contains(&k.as_str()))
.cloned()
.collect();
unknown.sort();
unknown
}
impl ProxyConfig {
pub fn write_timeout(&self) -> Duration {
Duration::from_secs(self.write_timeout_secs)
}
pub fn from_file(path: &str) -> Result<Self> {
let path = Path::new(path);
if !path.exists() {
return Err(ProxyError::Config(format!(
"Configuration file not found: {}",
path.display()
)));
}
let raw = std::fs::read_to_string(path)
.map_err(|e| ProxyError::Config(format!("Failed to read config: {}", e)))?;
let contents = substitute_env(&raw)?;
let config: Self = toml::from_str(&contents)
.map_err(|e| ProxyError::Config(format!("Failed to parse config: {}", e)))?;
for key in unknown_top_level_keys(&contents) {
tracing::warn!(
"unknown config section/key '{}' ignored (not part of ProxyConfig)",
key
);
}
config.validate()?;
Ok(config)
}
pub fn add_node(&mut self, host_port: &str, role: &str) -> Result<()> {
let parts: Vec<&str> = host_port.rsplitn(2, ':').collect();
if parts.len() != 2 {
return Err(ProxyError::Config(format!(
"Invalid host:port format: {}",
host_port
)));
}
let port: u16 = parts[0]
.parse()
.map_err(|_| ProxyError::Config(format!("Invalid port: {}", parts[0])))?;
let host = parts[1].to_string();
let role = match role {
"primary" => NodeRole::Primary,
"standby" => NodeRole::Standby,
"replica" => NodeRole::ReadReplica,
_ => return Err(ProxyError::Config(format!("Unknown role: {}", role))),
};
self.nodes.push(NodeConfig {
host,
port,
http_port: default_http_port(),
role,
weight: 100,
enabled: true,
name: None,
});
Ok(())
}
pub fn validate(&self) -> Result<()> {
if self.nodes.is_empty() {
return Err(ProxyError::Config(
"No backend nodes configured".to_string(),
));
}
let has_primary = self.nodes.iter().any(|n| n.role == NodeRole::Primary);
if !has_primary {
return Err(ProxyError::Config("No primary node configured".to_string()));
}
if self.pool.max_connections < self.pool.min_connections {
return Err(ProxyError::Config(
"max_connections must be >= min_connections".to_string(),
));
}
if self.health.check_interval_secs == 0 {
return Err(ProxyError::Config(
"health.check_interval_secs must be >= 1".to_string(),
));
}
if self.admin_token.is_none() && !self.admin_allow_insecure {
if let Ok(sa) = self.admin_address.parse::<std::net::SocketAddr>() {
if !sa.ip().is_loopback() {
return Err(ProxyError::Config(format!(
"admin_address '{}' is not loopback but admin_token is unset — the admin \
API runs privileged operations and must not be exposed anonymously. Set \
admin_token, bind admin_address to 127.0.0.1, or set \
admin_allow_insecure = true to override.",
self.admin_address
)));
}
}
}
if self.edge.enabled {
if !cfg!(feature = "edge-proxy") {
return Err(ProxyError::Config(
"edge.enabled = true but this binary was built without the 'edge-proxy' \
feature — rebuild with `--features edge-proxy` or remove/disable the \
[edge] section."
.to_string(),
));
}
if self.edge.subscribe_gc_secs == 0 {
return Err(ProxyError::Config(
"edge.subscribe_gc_secs must be >= 1".to_string(),
));
}
if self.edge.liveness_window_secs == 0 {
return Err(ProxyError::Config(
"edge.liveness_window_secs must be >= 1".to_string(),
));
}
if self.edge.default_ttl_secs == 0 {
return Err(ProxyError::Config(
"edge.default_ttl_secs must be >= 1 when edge is enabled".to_string(),
));
}
if self.edge.role == crate::edge::EdgeRole::Edge {
if self.edge.home_url.trim().is_empty() {
return Err(ProxyError::Config(
"edge.role = 'edge' requires edge.home_url — the home proxy's admin \
base URL (e.g. \"https://home-proxy:9090\") the edge subscribes to \
for cache invalidations."
.to_string(),
));
}
if !self.edge.auth_token.is_empty()
&& !self.edge.allow_insecure_home_url
&& !self
.edge
.home_url
.trim()
.to_ascii_lowercase()
.starts_with("https://")
{
return Err(ProxyError::Config(format!(
"edge.home_url '{}' is not https:// but edge.auth_token is set — the \
token is the home's admin bearer and must not cross the network in \
cleartext. Front the home admin port with a TLS terminator and use \
https://, or set edge.allow_insecure_home_url = true for private \
links (VPN/WireGuard/service mesh).",
self.edge.home_url.trim()
)));
}
if cfg!(feature = "query-cache") && self.cache.enabled {
return Err(ProxyError::Config(
"edge.role = 'edge' cannot be combined with [cache] enabled = true — \
the query-result cache does not receive edge invalidations and would \
serve stale rows past the edge coherence bound. Disable [cache] on \
edge-role proxies; the edge cache serves cacheable SELECTs there."
.to_string(),
));
}
if self.nodes.is_empty() {
return Err(ProxyError::Config(
"edge.role = 'edge' requires at least one [[nodes]] entry pointing \
at the home proxy's PG-wire listener — cache misses and writes \
forward there."
.to_string(),
));
}
}
}
Ok(())
}
pub fn primary_node(&self) -> Option<&NodeConfig> {
self.nodes
.iter()
.find(|n| n.role == NodeRole::Primary && n.enabled)
}
pub fn standby_nodes(&self) -> Vec<&NodeConfig> {
self.nodes
.iter()
.filter(|n| n.role == NodeRole::Standby && n.enabled)
.collect()
}
pub fn enabled_nodes(&self) -> Vec<&NodeConfig> {
self.nodes.iter().filter(|n| n.enabled).collect()
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
#[derive(Default)]
pub enum TrMode {
None,
#[default]
Session,
Select,
Transaction,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PoolConfig {
pub min_connections: usize,
pub max_connections: usize,
pub idle_timeout_secs: u64,
pub max_lifetime_secs: u64,
pub acquire_timeout_secs: u64,
pub test_on_acquire: bool,
}
impl Default for PoolConfig {
fn default() -> Self {
Self {
min_connections: 2,
max_connections: 100,
idle_timeout_secs: 300,
max_lifetime_secs: 1800,
acquire_timeout_secs: 30,
test_on_acquire: true,
}
}
}
impl PoolConfig {
pub fn idle_timeout(&self) -> Duration {
Duration::from_secs(self.idle_timeout_secs)
}
pub fn max_lifetime(&self) -> Duration {
Duration::from_secs(self.max_lifetime_secs)
}
pub fn acquire_timeout(&self) -> Duration {
Duration::from_secs(self.acquire_timeout_secs)
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LoadBalancerConfig {
pub read_strategy: Strategy,
pub read_write_split: bool,
pub latency_threshold_ms: u64,
}
impl Default for LoadBalancerConfig {
fn default() -> Self {
Self {
read_strategy: Strategy::RoundRobin,
read_write_split: true,
latency_threshold_ms: 100,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Strategy {
RoundRobin,
WeightedRoundRobin,
LeastConnections,
LatencyBased,
Random,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HealthConfig {
pub check_interval_secs: u64,
pub check_timeout_secs: u64,
pub failure_threshold: u32,
pub success_threshold: u32,
pub check_query: String,
}
impl Default for HealthConfig {
fn default() -> Self {
Self {
check_interval_secs: 5,
check_timeout_secs: 3,
failure_threshold: 3,
success_threshold: 2,
check_query: "SELECT 1".to_string(),
}
}
}
impl HealthConfig {
pub fn check_interval(&self) -> Duration {
Duration::from_secs(self.check_interval_secs)
}
pub fn check_timeout(&self) -> Duration {
Duration::from_secs(self.check_timeout_secs)
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NodeConfig {
pub host: String,
pub port: u16,
#[serde(default = "default_http_port")]
pub http_port: u16,
pub role: NodeRole,
pub weight: u32,
pub enabled: bool,
pub name: Option<String>,
}
fn default_http_port() -> u16 {
8080
}
impl NodeConfig {
pub fn address(&self) -> String {
format!("{}:{}", self.host, self.port)
}
pub fn display_name(&self) -> &str {
self.name.as_deref().unwrap_or(&self.host)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum NodeRole {
Primary,
Standby,
#[serde(rename = "replica")]
ReadReplica,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TlsConfig {
pub enabled: bool,
pub cert_path: String,
pub key_path: String,
pub ca_path: Option<String>,
pub require_client_cert: bool,
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_default_config() {
let config = ProxyConfig::default();
assert_eq!(config.listen_address, "0.0.0.0:5432");
assert!(config.tr_enabled);
}
#[test]
fn test_add_node() {
let mut config = ProxyConfig::default();
config.add_node("localhost:5432", "primary").unwrap();
config.add_node("localhost:5433", "standby").unwrap();
assert_eq!(config.nodes.len(), 2);
assert!(config.primary_node().is_some());
assert_eq!(config.standby_nodes().len(), 1);
}
#[test]
fn test_substitute_env_set_value_wins() {
std::env::set_var("HELIOS_SUBST_TEST_SET", "hello");
assert_eq!(
substitute_env("x = \"${HELIOS_SUBST_TEST_SET:-fallback}\"").unwrap(),
"x = \"hello\""
);
assert_eq!(
substitute_env("x = \"${HELIOS_SUBST_TEST_SET}\"").unwrap(),
"x = \"hello\""
);
std::env::remove_var("HELIOS_SUBST_TEST_SET");
}
#[test]
fn test_substitute_env_default_fallback() {
std::env::remove_var("HELIOS_SUBST_TEST_UNSET_A");
assert_eq!(
substitute_env("s = \"${HELIOS_SUBST_TEST_UNSET_A:-abc}\"").unwrap(),
"s = \"abc\""
);
}
#[test]
fn test_substitute_env_empty_default() {
std::env::remove_var("HELIOS_SUBST_TEST_UNSET_B");
assert_eq!(
substitute_env("s = \"${HELIOS_SUBST_TEST_UNSET_B:-}\"").unwrap(),
"s = \"\""
);
}
#[test]
fn test_substitute_env_missing_no_default_errors() {
std::env::remove_var("HELIOS_SUBST_TEST_UNSET_C");
let err = substitute_env("s = \"${HELIOS_SUBST_TEST_UNSET_C}\"").unwrap_err();
let msg = err.to_string();
assert!(
msg.contains("HELIOS_SUBST_TEST_UNSET_C"),
"error must name the missing variable, got: {msg}"
);
}
#[test]
fn test_substitute_env_skips_full_line_comments() {
std::env::remove_var("HELIOS_SUBST_TEST_UNSET_D");
let input = " # default_password = \"${HELIOS_SUBST_TEST_UNSET_D}\"\nx = 1\n";
assert_eq!(substitute_env(input).unwrap(), input);
}
#[test]
fn test_substitute_env_multiple_on_one_line() {
std::env::remove_var("HELIOS_SUBST_TEST_UNSET_E");
std::env::remove_var("HELIOS_SUBST_TEST_UNSET_F");
assert_eq!(
substitute_env(
"addr = \"${HELIOS_SUBST_TEST_UNSET_E:-host}:${HELIOS_SUBST_TEST_UNSET_F:-5432}\""
)
.unwrap(),
"addr = \"host:5432\""
);
}
#[test]
fn test_substitute_env_unquoted_numeric_position() {
std::env::remove_var("HELIOS_SUBST_TEST_UNSET_G");
let out = substitute_env("max_connections = ${HELIOS_SUBST_TEST_UNSET_G:-50}").unwrap();
assert_eq!(out, "max_connections = 50");
#[derive(serde::Deserialize)]
struct P {
max_connections: u32,
}
let p: P = toml::from_str(&out).unwrap();
assert_eq!(p.max_connections, 50);
}
#[test]
fn test_substitute_env_leaves_malformed_literal() {
assert_eq!(substitute_env("cost = $5.00\n").unwrap(), "cost = $5.00\n");
std::env::remove_var("HELIOS_SUBST_TEST_UNSET_H");
assert_eq!(
substitute_env("x = \"${HELIOS_SUBST_TEST_UNSET_H:-oops\"").unwrap(),
"x = \"${HELIOS_SUBST_TEST_UNSET_H:-oops\""
);
}
#[test]
fn test_unknown_top_level_keys_detection() {
let text = "listen_address = \"x\"\n\
[pool]\nmin_connections = 1\n\
[ha]\nenabled = true\n\
[logging]\nlevel = \"info\"\n";
assert_eq!(
unknown_top_level_keys(text),
vec!["ha".to_string(), "logging".to_string()]
);
}
#[test]
fn test_unknown_top_level_keys_nested_are_out_of_scope() {
let text = "[cache]\nenabled = true\n[cache.l1]\nsize = 500\n";
assert!(unknown_top_level_keys(text).is_empty());
}
#[test]
fn test_known_top_level_keys_cover_struct_fields() {
let value = toml::Value::try_from(ProxyConfig::default()).unwrap();
let table = value.as_table().unwrap();
for k in table.keys() {
assert!(
KNOWN_TOP_LEVEL_KEYS.contains(&k.as_str()),
"field '{k}' is present in a serialised default ProxyConfig but \
missing from KNOWN_TOP_LEVEL_KEYS"
);
}
}
#[test]
fn test_all_shipped_configs_parse() {
let manifest = env!("CARGO_MANIFEST_DIR");
let config_dir = format!("{manifest}/config");
let regress_dir = format!("{manifest}/scripts/regress");
let mut config_checked = 0usize;
let entries = std::fs::read_dir(&config_dir)
.unwrap_or_else(|e| panic!("config dir {config_dir} unreadable: {e}"));
for entry in entries {
let path = entry.unwrap().path();
if path.extension().and_then(|e| e.to_str()) != Some("toml") {
continue;
}
let path_str = path
.to_str()
.unwrap_or_else(|| panic!("non-UTF-8 config path {}", path.display()));
let loaded = ProxyConfig::from_file(path_str);
assert!(
loaded.is_ok(),
"shipped config {} failed to load via from_file() \
(substitute + parse + validate): {}",
path.display(),
loaded.err().unwrap()
);
config_checked += 1;
}
assert!(
config_checked >= 3,
"expected to load at least the 3 config/*.toml files, checked {config_checked}"
);
if let Ok(entries) = std::fs::read_dir(®ress_dir) {
for entry in entries {
let path = entry.unwrap().path();
if path.extension().and_then(|e| e.to_str()) != Some("toml") {
continue;
}
let raw = std::fs::read_to_string(&path).unwrap();
let substituted = substitute_env(&raw).unwrap_or_else(|e| {
panic!("env substitution failed for {}: {e}", path.display())
});
let parsed = toml::from_str::<ProxyConfig>(&substituted);
assert!(
parsed.is_ok(),
"regress config {} failed to deserialize: {}",
path.display(),
parsed.err().unwrap()
);
}
}
}
#[test]
fn test_validate_no_nodes() {
let config = ProxyConfig::default();
assert!(config.validate().is_err());
}
#[test]
fn test_validate_no_primary() {
let mut config = ProxyConfig::default();
config.add_node("localhost:5432", "standby").unwrap();
assert!(config.validate().is_err());
}
#[test]
fn test_validate_success() {
let mut config = ProxyConfig::default();
config.add_node("localhost:5432", "primary").unwrap();
assert!(config.validate().is_ok());
}
#[test]
fn test_validate_refuses_anonymous_nonloopback_admin() {
let base = || {
let mut c = ProxyConfig::default();
c.add_node("localhost:5432", "primary").unwrap();
c
};
let mut c = base();
c.admin_address = "127.0.0.1:9090".to_string();
assert!(c.validate().is_ok());
let mut c = base();
c.admin_address = "0.0.0.0:9090".to_string();
assert!(
c.validate().is_err(),
"anonymous 0.0.0.0 admin must be refused"
);
let mut c = base();
c.admin_address = "0.0.0.0:9090".to_string();
c.admin_token = Some("secret".to_string());
assert!(c.validate().is_ok());
let mut c = base();
c.admin_address = "0.0.0.0:9090".to_string();
c.admin_allow_insecure = true;
assert!(c.validate().is_ok());
}
#[test]
fn test_validate_rejects_zero_health_interval() {
let mut config = ProxyConfig::default();
config.add_node("localhost:5432", "primary").unwrap();
config.health.check_interval_secs = 0;
assert!(config.validate().is_err());
config.health.check_interval_secs = 1;
assert!(config.validate().is_ok());
}
#[test]
fn test_validate_edge_disabled_section_is_inert() {
let mut config = ProxyConfig::default();
config.add_node("localhost:5432", "primary").unwrap();
assert!(!config.edge.enabled);
assert!(config.validate().is_ok());
}
#[test]
fn test_validate_edge_enabled_requires_feature() {
let mut config = ProxyConfig::default();
config.add_node("localhost:5432", "primary").unwrap();
config.edge.enabled = true;
if cfg!(feature = "edge-proxy") {
assert!(config.validate().is_ok());
} else {
assert!(
config.validate().is_err(),
"edge.enabled on a build without the edge-proxy feature must be refused"
);
}
}
#[cfg(feature = "edge-proxy")]
#[test]
fn test_validate_edge_rejects_zero_intervals() {
let base = || {
let mut c = ProxyConfig::default();
c.add_node("localhost:5432", "primary").unwrap();
c.edge.enabled = true;
c
};
let mut c = base();
c.edge.subscribe_gc_secs = 0;
assert!(c.validate().is_err());
let mut c = base();
c.edge.liveness_window_secs = 0;
assert!(c.validate().is_err());
let mut c = base();
c.edge.enabled = false;
c.edge.subscribe_gc_secs = 0;
assert!(c.validate().is_ok());
}
#[cfg(feature = "edge-proxy")]
#[test]
fn test_validate_edge_role_requires_home_url() {
let base = || {
let mut c = ProxyConfig::default();
c.add_node("localhost:5432", "primary").unwrap();
c.edge.enabled = true;
c.edge.role = crate::edge::EdgeRole::Edge;
c
};
let c = base();
let err = c.validate().unwrap_err().to_string();
assert!(err.contains("home_url"), "unexpected error: {}", err);
let mut c = base();
c.edge.home_url = "http://home-proxy:9090".to_string();
assert!(c.validate().is_ok());
}
#[cfg(feature = "edge-proxy")]
#[test]
fn test_validate_edge_zero_ttl_refused_when_enabled() {
let mut c = ProxyConfig::default();
c.add_node("localhost:5432", "primary").unwrap();
c.edge.enabled = true;
c.edge.default_ttl_secs = 0;
let err = c.validate().unwrap_err().to_string();
assert!(
err.contains("default_ttl_secs"),
"unexpected error: {}",
err
);
c.edge.enabled = false;
assert!(c.validate().is_ok());
}
#[cfg(feature = "edge-proxy")]
#[test]
fn test_validate_edge_token_requires_https_home_url() {
let base = || {
let mut c = ProxyConfig::default();
c.add_node("localhost:5432", "primary").unwrap();
c.edge.enabled = true;
c.edge.role = crate::edge::EdgeRole::Edge;
c.edge.home_url = "http://home-proxy:9090".to_string();
c
};
assert!(base().validate().is_ok());
let mut c = base();
c.edge.auth_token = "secret".to_string();
let err = c.validate().unwrap_err().to_string();
assert!(err.contains("https"), "unexpected error: {}", err);
assert!(err.contains("allow_insecure_home_url"), "{}", err);
let mut c = base();
c.edge.auth_token = "secret".to_string();
c.edge.allow_insecure_home_url = true;
assert!(c.validate().is_ok());
let mut c = base();
c.edge.auth_token = "secret".to_string();
c.edge.home_url = "https://home-proxy:9090".to_string();
assert!(c.validate().is_ok());
}
#[cfg(all(feature = "edge-proxy", feature = "query-cache"))]
#[test]
fn test_validate_edge_role_rejects_query_cache_combo() {
let mut c = ProxyConfig::default();
c.add_node("localhost:5432", "primary").unwrap();
c.edge.enabled = true;
c.edge.role = crate::edge::EdgeRole::Edge;
c.edge.home_url = "https://home-proxy:9090".to_string();
c.cache.enabled = true;
let err = c.validate().unwrap_err().to_string();
assert!(err.contains("[cache]"), "unexpected error: {}", err);
c.edge.role = crate::edge::EdgeRole::Home;
c.edge.home_url.clear();
assert!(c.validate().is_ok());
c.edge.role = crate::edge::EdgeRole::Edge;
c.edge.home_url = "https://home-proxy:9090".to_string();
c.cache.enabled = false;
assert!(c.validate().is_ok());
}
#[test]
fn test_pool_config_durations() {
let config = PoolConfig::default();
assert_eq!(config.idle_timeout(), Duration::from_secs(300));
assert_eq!(config.max_lifetime(), Duration::from_secs(1800));
}
#[test]
fn test_pool_mode_default() {
let config = PoolModeConfig::default();
assert_eq!(config.mode, PoolingMode::Session);
assert_eq!(config.max_pool_size, 100);
assert_eq!(config.min_idle, 10);
assert_eq!(config.reset_query, "DISCARD ALL");
}
#[test]
fn test_pool_mode_session() {
let config = PoolModeConfig::session_mode();
assert_eq!(config.mode, PoolingMode::Session);
assert_eq!(config.prepared_statement_mode, PreparedStatementMode::Named);
}
#[test]
fn test_pool_mode_transaction() {
let config = PoolModeConfig::transaction_mode();
assert_eq!(config.mode, PoolingMode::Transaction);
assert_eq!(config.prepared_statement_mode, PreparedStatementMode::Track);
}
#[test]
fn test_pool_mode_statement() {
let config = PoolModeConfig::statement_mode();
assert_eq!(config.mode, PoolingMode::Statement);
assert_eq!(
config.prepared_statement_mode,
PreparedStatementMode::Disable
);
}
#[test]
fn test_pool_mode_durations() {
let config = PoolModeConfig::default();
assert_eq!(config.idle_timeout(), Duration::from_secs(600));
assert_eq!(config.max_lifetime(), Duration::from_secs(3600));
assert_eq!(config.acquire_timeout(), Duration::from_secs(5));
}
#[test]
fn test_proxy_config_has_pool_mode() {
let config = ProxyConfig::default();
assert_eq!(config.pool_mode.mode, PoolingMode::Session);
}
#[test]
fn test_plugin_toml_default_is_disabled() {
let config = ProxyConfig::default();
assert!(!config.plugins.enabled);
assert_eq!(config.plugins.plugin_dir, "/etc/heliosproxy/plugins");
assert_eq!(config.plugins.memory_limit_mb, 64);
assert_eq!(config.plugins.timeout_ms, 100);
}
#[test]
fn test_proxy_config_toml_without_plugins_section_still_parses() {
let toml_text = r#"
listen_address = "0.0.0.0:5432"
admin_address = "0.0.0.0:9090"
tr_enabled = true
tr_mode = "session"
nodes = []
[pool]
min_connections = 2
max_connections = 10
idle_timeout_secs = 300
max_lifetime_secs = 1800
acquire_timeout_secs = 30
test_on_acquire = true
[load_balancer]
read_strategy = "round_robin"
read_write_split = true
latency_threshold_ms = 100
[health]
check_interval_secs = 5
check_timeout_secs = 3
failure_threshold = 3
success_threshold = 2
check_query = "SELECT 1"
"#;
let config: ProxyConfig = toml::from_str(toml_text).expect("parse");
assert!(!config.plugins.enabled);
}
#[test]
fn test_plugin_toml_overrides_parse() {
let toml_text = r#"
listen_address = "0.0.0.0:5432"
admin_address = "0.0.0.0:9090"
tr_enabled = true
tr_mode = "session"
nodes = []
[pool]
min_connections = 2
max_connections = 10
idle_timeout_secs = 300
max_lifetime_secs = 1800
acquire_timeout_secs = 30
test_on_acquire = true
[load_balancer]
read_strategy = "round_robin"
read_write_split = true
latency_threshold_ms = 100
[health]
check_interval_secs = 5
check_timeout_secs = 3
failure_threshold = 3
success_threshold = 2
check_query = "SELECT 1"
[plugins]
enabled = true
plugin_dir = "/tmp/helios-plugins"
hot_reload = true
memory_limit_mb = 128
timeout_ms = 250
"#;
let config: ProxyConfig = toml::from_str(toml_text).expect("parse");
assert!(config.plugins.enabled);
assert_eq!(config.plugins.plugin_dir, "/tmp/helios-plugins");
assert!(config.plugins.hot_reload);
assert_eq!(config.plugins.memory_limit_mb, 128);
assert_eq!(config.plugins.timeout_ms, 250);
assert_eq!(config.plugins.max_plugins, 20);
assert!(config.plugins.fuel_metering);
}
}