From 1c8ccee5bb5fb2409c381ff67577e5acd7f8a463 Mon Sep 17 00:00:00 2001 From: CodexM5 Date: Sat, 26 Sep 2026 09:00:28 +0800 Subject: [PATCH 01/14] config: project single-cluster Rust VIP inputs --- rust/crates/control-config/src/lib.rs | 2 +- rust/crates/control-config/src/model.rs | 100 +++++++++++++++++++++ rust/crates/control-config/tests/source.rs | 62 +++++++++++++ 3 files changed, 163 insertions(+), 1 deletion(-) diff --git a/rust/crates/control-config/src/lib.rs b/rust/crates/control-config/src/lib.rs index 00d25d817..6e64da766 100644 --- a/rust/crates/control-config/src/lib.rs +++ b/rust/crates/control-config/src/lib.rs @@ -27,7 +27,7 @@ pub use model::{ RoutingBalancePolicy, RoutingConfig, RoutingConnectionFactorConfig, RoutingFactorConfig, RoutingNamespace, RoutingRule, RoutingSelectionPolicy, ServingConfig, ServingKeepalive, ServingListener, ServingNamespace, ServingTlsConfig, SourceRevision, TopologyConfig, - TopologyRuntimeIdentity, + TopologyRuntimeIdentity, VipConfig, }; pub use module::{ ConfigModule, ConfigModuleHandle, ConfigModuleOptions, ConfigMutationError, diff --git a/rust/crates/control-config/src/model.rs b/rust/crates/control-config/src/model.rs index 9649fc1b1..c01c5169a 100644 --- a/rust/crates/control-config/src/model.rs +++ b/rust/crates/control-config/src/model.rs @@ -100,6 +100,45 @@ pub struct ConfigPersistence { pub cluster_tls: ClientTlsConfig, } +/// Inputs for the dedicated VIP election and network owner. +/// +/// The decision to start is pinned at process startup, matching Go's VIP +/// manager. A missing address/interface pair or a non-singleton backend set +/// disables VIP; neither case grants ownership of a partially configured IP. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct VipConfig { + /// Address as configured, including its optional CIDR prefix. + pub address: Arc, + /// Parsed host IP used for the dedicated election key and GARP. + pub ip: IpAddr, + /// Prefix length to compare and bind on the Linux interface. + pub prefix_len: u8, + /// Linux interface that holds the address. + pub interface: Arc, + /// The sole backend cluster's PD endpoints, which own the VIP election. + pub pd_addrs: Arc<[Arc]>, + /// TLS material for that cluster's PD connection. + pub cluster_tls: ClientTlsConfig, + /// Immediate GARP packets in each burst. + pub garp_burst_count: u64, + /// One-second follow-up GARP bursts after takeover. + pub garp_refresh_count: u64, +} + +impl VipConfig { + /// Returns the Go-compatible election prefix for this exact VIP host IP. + #[must_use] + pub fn election_name(&self) -> String { + format!("/tiproxy/vip/{}/owner", self.ip) + } + + /// Returns the canonical address and prefix for Linux address operations. + #[must_use] + pub fn cidr(&self) -> String { + format!("{}/{}", self.ip, self.prefix_len) + } +} + /// Health observer policy consumed by CP-TOPO. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub struct HealthCheckConfig { @@ -596,6 +635,46 @@ impl EffectiveConfig { }) } + /// Projects an enabled VIP only for a single backend cluster. The + /// configured address and interface must both be present. Go elects via + /// that cluster's PD endpoints, which may differ from `proxy.pd-addrs`. + /// + /// # Errors + /// + /// Returns an invalid VIP address, backend normalization error or an + /// unrepresentable GARP count. Config validation already rejects negative + /// counts and gives a zero burst count the Go-compatible value of one. + pub fn vip(&self) -> Result, ConfigError> { + if self.ha.virtual_ip.is_empty() || self.ha.interface.is_empty() { + return Ok(None); + } + let clusters = self.normalized_backend_clusters()?; + let [cluster] = clusters.as_slice() else { + return Ok(None); + }; + let (ip, prefix_len) = parse_vip_address(&self.ha.virtual_ip)?; + let garp_burst_count = + u64::try_from(self.ha.garp_burst_count).map_err(|_| ConfigError::InvalidField { + field: "ha.garp-burst-count", + class: "out_of_range", + })?; + let garp_refresh_count = + u64::try_from(self.ha.garp_refresh_count).map_err(|_| ConfigError::InvalidField { + field: "ha.garp-refresh-count", + class: "out_of_range", + })?; + Ok(Some(VipConfig { + address: Arc::from(self.ha.virtual_ip.as_str()), + ip, + prefix_len, + interface: Arc::from(self.ha.interface.as_str()), + pd_addrs: Arc::clone(&cluster.pd_addrs), + cluster_tls: client_tls(&self.security.cluster_tls), + garp_burst_count, + garp_refresh_count, + })) + } + /// Returns the complete SQL-serving projection formerly built by Go. /// /// # Errors @@ -2424,6 +2503,27 @@ fn normalize_ip_prefix(value: &str) -> Result { } } +fn parse_vip_address(value: &str) -> Result<(IpAddr, u8), ConfigError> { + let failure = || ConfigError::InvalidField { + field: "ha.virtual-ip", + class: "invalid_ip_or_cidr", + }; + let (ip, prefix_len) = if let Some((ip, prefix)) = value.split_once('/') { + ( + ip.parse::().map_err(|_| failure())?, + prefix.parse::().map_err(|_| failure())?, + ) + } else { + let ip = value.parse::().map_err(|_| failure())?; + let prefix_len = if ip.is_ipv4() { 32 } else { 128 }; + (ip, prefix_len) + }; + if prefix_len > if ip.is_ipv4() { 32 } else { 128 } { + return Err(failure()); + } + Ok((ip, prefix_len)) +} + fn path_arc(value: &str) -> Option> { (!value.is_empty()).then(|| Arc::from(Path::new(value))) } diff --git a/rust/crates/control-config/tests/source.rs b/rust/crates/control-config/tests/source.rs index ec98836d7..6e80891f5 100644 --- a/rust/crates/control-config/tests/source.rs +++ b/rust/crates/control-config/tests/source.rs @@ -670,6 +670,68 @@ pd-addrs = "routing-pd:2379" assert_eq!(persistence.pd_addrs[0].as_ref(), "owner-pd:2379"); } +#[test] +fn vip_requires_complete_ha_pair_and_one_backend_cluster() { + let load = |source: &str| { + ConfigNamespaceStore::from_toml(source.as_bytes(), None, current_dir()) + .unwrap_or_else(|error| unreachable!("VIP config: {error}")) + }; + let basic = r#" +[proxy] +pd-addrs = "pd:2379" +[ha] +virtual-ip = "192.0.2.5/24" +interface = "eth0" +garp-burst-count = 0 +garp-refresh-count = 2 +"#; + let store = load(basic); + let vip = store + .current() + .effective() + .vip() + .unwrap_or_else(|error| unreachable!("VIP projection: {error}")) + .unwrap_or_else(|| unreachable!("complete single-cluster VIP is enabled")); + assert_eq!(vip.address.as_ref(), "192.0.2.5/24"); + assert_eq!(vip.interface.as_ref(), "eth0"); + assert_eq!(vip.election_name(), "/tiproxy/vip/192.0.2.5/owner"); + assert_eq!(vip.cidr(), "192.0.2.5/24"); + assert_eq!(vip.garp_burst_count, 1); + assert_eq!(vip.garp_refresh_count, 2); + + let incomplete = load(&basic.replace("interface = \"eth0\"", "interface = \"\"")); + assert!(matches!(incomplete.current().effective().vip(), Ok(None))); + let no_pd = load(&basic.replace("pd-addrs = \"pd:2379\"", "pd-addrs = \"\"")); + assert!(matches!(no_pd.current().effective().vip(), Ok(None))); + let explicit_cluster = load(&basic.replace( + "[ha]", + "[[proxy.backend-clusters]]\nname = \"only\"\npd-addrs = \"cluster-pd:2379\"\n[ha]", + )); + let cluster_vip = explicit_cluster + .current() + .effective() + .vip() + .unwrap_or_else(|error| unreachable!("VIP projection: {error}")) + .unwrap_or_else(|| unreachable!("single explicit backend cluster enables VIP")); + assert_eq!(cluster_vip.pd_addrs[0].as_ref(), "cluster-pd:2379"); + let two_clusters = load( + &basic.replace( + "[ha]", + "[[proxy.backend-clusters]]\nname = \"a\"\npd-addrs = \"pd-a:2379\"\n[[proxy.backend-clusters]]\nname = \"b\"\npd-addrs = \"pd-b:2379\"\n[ha]", + ), + ); + assert!(matches!(two_clusters.current().effective().vip(), Ok(None))); + + let invalid_address = load(&basic.replace("192.0.2.5/24", "192.0.2.5/33")); + assert!(matches!( + invalid_address.current().effective().vip(), + Err(ConfigError::InvalidField { + field: "ha.virtual-ip", + class: "invalid_ip_or_cidr" + }) + )); +} + #[test] fn config_module_rejects_a_factory_that_disagrees_with_initial_transport() { let initial = EtcdClientConfig::new(["initial-pd:2379".to_owned()], None) From 8b9481b643ae93fd62dda9232edc51f607dcb058 Mon Sep 17 00:00:00 2001 From: CodexM5 Date: Sat, 26 Sep 2026 09:41:29 +0800 Subject: [PATCH 02/14] Add standalone Rust VIP election and network owner --- rust/crates/tiproxy-rs/Cargo.toml | 2 +- rust/crates/tiproxy-rs/src/main.rs | 138 ++++++- rust/crates/tiproxy-rs/src/vip_network.rs | 268 +++++++++++++ rust/crates/tiproxy-rs/src/vip_owner.rs | 443 ++++++++++++++++++++++ 4 files changed, 840 insertions(+), 11 deletions(-) create mode 100644 rust/crates/tiproxy-rs/src/vip_network.rs create mode 100644 rust/crates/tiproxy-rs/src/vip_owner.rs diff --git a/rust/crates/tiproxy-rs/Cargo.toml b/rust/crates/tiproxy-rs/Cargo.toml index 8ea36e4b9..3995a29d9 100644 --- a/rust/crates/tiproxy-rs/Cargo.toml +++ b/rust/crates/tiproxy-rs/Cargo.toml @@ -28,7 +28,7 @@ if-addrs.workspace = true libc.workspace = true rustix.workspace = true rustls.workspace = true -tokio = { workspace = true, features = ["signal"] } +tokio = { workspace = true, features = ["process", "signal"] } tokio-rustls.workspace = true [dev-dependencies] diff --git a/rust/crates/tiproxy-rs/src/main.rs b/rust/crates/tiproxy-rs/src/main.rs index bf958f812..cf7ce3942 100644 --- a/rust/crates/tiproxy-rs/src/main.rs +++ b/rust/crates/tiproxy-rs/src/main.rs @@ -23,6 +23,8 @@ mod native_meter; mod startup; mod tls_material; mod topology_composition; +mod vip_network; +mod vip_owner; use std::env; use std::fmt::Display; @@ -78,6 +80,8 @@ use topology_composition::{ ArtifactClusterFactory, CompositeCandidateValidator, MetricServerConnectionAcceptor, TopologyCandidateValidator, interface_advertise_candidates, metric_owner_bind_address, }; +use vip_network::{LinuxNetwork, NetworkOperation}; +use vip_owner::{VipHandle, VipModule}; const VERSION: &str = env!("TIPROXY_BUILD_VERSION"); const COMMIT: &str = env!("TIPROXY_BUILD_COMMIT"); @@ -698,6 +702,50 @@ async fn run(options: Options) -> Result<(), String> { .await); } }; + // Go still owns VIP in the two-process configuration. Only the pure-Rust + // standalone process may start the new election, so the two owners cannot + // silently compete for the same address during the migration period. + let vip_plan = if options.standalone { + let vip = match config_owner.handle.source().current().effective().vip() { + Ok(vip) => vip, + Err(error) => return Err(guard.rollback(format!("project VIP config: {error}")).await), + }; + if let Some(vip) = vip { + let roots = tls_material::open_tls_roots(&config_owner.tls_roots); + let tls = match etcd_tls(&vip.cluster_tls, &roots) { + Ok(tls) => tls, + Err(error) => { + return Err(guard.rollback(format!("prepare VIP TLS: {error}")).await); + } + }; + let endpoints = vip + .pd_addrs + .iter() + .map(ToString::to_string) + .collect::>(); + let client = EtcdClientConfig::new(endpoints, tls).and_then(|client| { + client.with_timeouts( + Duration::from_secs(2), + Duration::from_secs(1), + Duration::from_secs(1), + Duration::from_secs(1), + Duration::from_secs(30), + ) + }); + match client { + Ok(client) => Some((vip, client)), + Err(error) => { + return Err(guard + .rollback(format!("prepare VIP etcd client: {error}")) + .await); + } + } + } else { + None + } + } else { + None + }; let metric_bind_address = metric_owner_bind_address(&initial_topology, &metric_advertise_host); let (topology_module, mut topology_handle) = match TopologyModule::new( Arc::new(config_owner.handle.source().clone()), @@ -897,7 +945,7 @@ async fn run(options: Options) -> Result<(), String> { serving.clone(), route_plane_handle, route_config_source, - topology_handle, + topology_handle.clone(), Arc::clone(&runtime_stats), HealthGates { lb_readiness: lb_readiness.clone(), @@ -1065,11 +1113,63 @@ async fn run(options: Options) -> Result<(), String> { .await); } + let mut vip_handle = None; + if let Some((vip, etcd)) = vip_plan { + let network = match LinuxNetwork::new(&vip) { + Ok(network) => Arc::new(network), + Err(error) => { + return Err(guard + .rollback(format!( + "initialize VIP network: {} / {}", + error.operation, error.class + )) + .await); + } + }; + // Match Go Start: clear a stale address before the process becomes + // ready or the dedicated election can be won. + if let Err(error) = network.delete_ip().await { + return Err(guard + .rollback(format!( + "clear stale VIP: {} / {}", + error.operation, error.class + )) + .await); + } + let host = metric_advertise_host.as_ref(); + let member_id = if host.contains(':') { + format!("[{host}]:{}", initial_topology.sql_port) + } else { + format!("{host}:{}", initial_topology.sql_port) + }; + let process_id = runtime_handle.lifecycle().owner_id; + let (module, handle) = match VipModule::new( + &vip, + member_id, + process_id.as_ref(), + etcd, + network, + topology_handle.owner_history(), + ) { + Ok(parts) => parts, + Err(error) => return Err(guard.rollback(error).await), + }; + if let Err(error) = guard.spawn_module(module) { + return Err(guard.rollback(format!("start VIP module: {error}")).await); + } + vip_handle = Some(handle); + } + if let Err(error) = in_process.mark_ready() { meter_ready.send_replace(Some(false)); - return Err(guard - .rollback(format!("mark in-process control runtime ready: {error}")) - .await); + let vip_close = close_vip(&mut vip_handle).await.err(); + let failure = match vip_close { + Some(vip_close) => { + format!("mark in-process control runtime ready: {error}; {vip_close}") + } + None => format!("mark in-process control runtime ready: {error}"), + }; + return Err(guard.rollback(failure).await); } meter_ready.send_replace(Some(true)); if let Some(app) = &admin_app { @@ -1117,12 +1217,13 @@ async fn run(options: Options) -> Result<(), String> { None => std::future::pending().await, } }); - let (control_result, sampler_result, serving_result, module_result) = tokio::select! { + let (control_result, sampler_result, serving_result, module_result, vip_close_result) = tokio::select! { control = &mut control_runtime => { let control = match control { Ok(result) => result.map_err(|error| error.to_string()), Err(_) => Err("control runtime supervisor panicked".to_owned()), }; + let vip_close_result = close_vip(&mut vip_handle).await; if control.is_err() { in_process.fail("legacy_bridge", "runtime_failure"); } else { @@ -1141,13 +1242,14 @@ async fn run(options: Options) -> Result<(), String> { Ok(result) => result.map_err(|error| error.to_string()), Err(_) => Err("metering sampler panicked".to_owned()), }; - (control, sampler, serving_result, Ok(())) + (control, sampler, serving_result, Ok(()), vip_close_result) } sampler = &mut metering_sampler => { let sampler = match sampler { Ok(result) => result.map_err(|error| error.to_string()), Err(_) => Err("metering sampler panicked".to_owned()), }; + let vip_close_result = close_vip(&mut vip_handle).await; if sampler.is_err() { in_process.fail("metering_sampler", "runtime_failure"); } else { @@ -1170,12 +1272,13 @@ async fn run(options: Options) -> Result<(), String> { Ok(result) => result.map_err(|error| error.to_string()), Err(_) => Err("control runtime supervisor panicked".to_owned()), }; - (control, sampler, serving_result, Ok(())) + (control, sampler, serving_result, Ok(()), vip_close_result) } () = &mut termination => { // Close readiness as soon as SIGTERM wins the select, even if an // in-flight config apply briefly delays the applied-snapshot read. lb_readiness.mark_unhealthy(); + let vip_close_result = close_vip(&mut vip_handle).await; let lb_wait = serving.graceful_wait_before_shutdown().await; let event = async { tokio::select! { @@ -1242,13 +1345,14 @@ async fn run(options: Options) -> Result<(), String> { Err(_) => Err("control runtime supervisor panicked".to_owned()), }, }; - (control, sampler, serving_result, module_result) + (control, sampler, serving_result, module_result, vip_close_result) } admin = &mut admin_exit => { let failure = match admin { Ok(()) => "control admin server exited unexpectedly".to_owned(), Err(_) => "control admin server panicked".to_owned(), }; + let vip_close_result = close_vip(&mut vip_handle).await; in_process.fail("control_admin", "runtime_failure"); let serving_result = stop_drain_and_join_sessions( &in_process, @@ -1266,7 +1370,7 @@ async fn run(options: Options) -> Result<(), String> { Ok(result) => result.map_err(|error| error.to_string()), Err(_) => Err("control runtime supervisor panicked".to_owned()), }; - (control, sampler, serving_result, Err(failure)) + (control, sampler, serving_result, Err(failure), vip_close_result) } module = modules.join_next() => { let failure = match module { @@ -1276,6 +1380,7 @@ async fn run(options: Options) -> Result<(), String> { }, None => "control module executor became empty unexpectedly".to_owned(), }; + let vip_close_result = close_vip(&mut vip_handle).await; in_process.fail("control_module", "runtime_failure"); let serving_result = stop_drain_and_join_sessions( &in_process, @@ -1293,7 +1398,7 @@ async fn run(options: Options) -> Result<(), String> { Ok(result) => result.map_err(|error| error.to_string()), Err(_) => Err("control runtime supervisor panicked".to_owned()), }; - (control, sampler, serving_result, Err(failure)) + (control, sampler, serving_result, Err(failure), vip_close_result) } }; if let Some(observer) = routing_shadow { @@ -1334,6 +1439,7 @@ async fn run(options: Options) -> Result<(), String> { prefer_control_failure(control_result, sampler_result)?; serving_result?; module_result?; + vip_close_result?; module_executor_result?; finish_result?; Ok(()) @@ -1347,6 +1453,18 @@ fn prefer_control_failure( sampler } +async fn close_vip(handle: &mut Option) -> Result<(), String> { + let Some(handle) = handle.as_mut() else { + return Ok(()); + }; + handle.pre_close().await.map_err(|error| { + format!( + "remove VIP before SQL drain: {} / {}", + error.operation, error.class + ) + }) +} + enum OrderedWaitExit { Control(Result<(), String>), Sampler(Result<(), String>), diff --git a/rust/crates/tiproxy-rs/src/vip_network.rs b/rust/crates/tiproxy-rs/src/vip_network.rs new file mode 100644 index 000000000..57df39611 --- /dev/null +++ b/rust/crates/tiproxy-rs/src/vip_network.rs @@ -0,0 +1,268 @@ +// Copyright 2026 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Linux virtual-IP address operations behind a replaceable owner boundary. + +use std::future::Future; +use std::net::IpAddr; +use std::pin::Pin; +use std::sync::Arc; +use std::time::Duration; + +use control_config::VipConfig; +use tokio::process::Command; + +const COMMAND_TIMEOUT: Duration = Duration::from_secs(2); + +type NetworkFuture<'a, T> = Pin> + Send + 'a>>; + +/// The VIP actor uses the same interface for a real Linux host and a fake. +pub(super) trait NetworkOperation: Send + Sync { + fn has_ip(&self) -> NetworkFuture<'_, bool>; + fn add_ip(&self) -> NetworkFuture<'_, ()>; + fn delete_ip(&self) -> NetworkFuture<'_, ()>; + fn send_arp(&self) -> NetworkFuture<'_, ()>; +} + +/// Bounded class of one failed Linux VIP operation; no command output leaks. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) struct NetworkError { + pub operation: &'static str, + pub class: &'static str, +} + +/// IP address and interface configured for this process's VIP owner. +pub(super) struct LinuxNetwork { + ip: IpAddr, + prefix_len: u8, + interface: Arc, + burst_count: u64, +} + +impl LinuxNetwork { + pub fn new(config: &VipConfig) -> Result { + if !cfg!(target_os = "linux") { + return Err(NetworkError { + operation: "initialize", + class: "unsupported_os", + }); + } + Ok(Self { + ip: config.ip, + prefix_len: config.prefix_len, + interface: Arc::clone(&config.interface), + burst_count: config.garp_burst_count, + }) + } + + fn cidr(&self) -> String { + format!("{}/{}", self.ip, self.prefix_len) + } + + async fn ip_output( + &self, + args: &[&str], + operation: &'static str, + ) -> Result { + run_command("ip", args, operation).await + } + + async fn mutate(&self, verb: &'static str) -> Result<(), NetworkError> { + let cidr = self.cidr(); + let args = ["addr", verb, cidr.as_str(), "dev", self.interface.as_ref()]; + let output = self.ip_output(&args, verb).await?; + if output.success { + return Ok(()); + } + // Go retries netlink AddrAdd/AddrDel through sudo only for EPERM. + // The `ip` CLI emits this stable English text with LC_ALL=C. + if !output.stderr.contains("Operation not permitted") + && !output.stderr.contains("Permission denied") + { + return Err(NetworkError { + operation: verb, + class: "ip_failed", + }); + } + let sudo_args = [ + "-n", + "ip", + "addr", + verb, + cidr.as_str(), + "dev", + self.interface.as_ref(), + ]; + let retry = run_command("sudo", &sudo_args, verb).await?; + retry.success.then_some(()).ok_or(NetworkError { + operation: verb, + class: "sudo_ip_failed", + }) + } +} + +impl NetworkOperation for LinuxNetwork { + fn has_ip(&self) -> NetworkFuture<'_, bool> { + Box::pin(async move { + let args = ["-o", "addr", "show", "dev", self.interface.as_ref()]; + let output = self.ip_output(&args, "list").await?; + if !output.success { + return Err(NetworkError { + operation: "list", + class: "ip_failed", + }); + } + Ok(contains_exact_address( + &output.stdout, + self.ip, + self.prefix_len, + )) + }) + } + + fn add_ip(&self) -> NetworkFuture<'_, ()> { + Box::pin(async move { + if self.has_ip().await? { + return Ok(()); + } + self.mutate("add").await + }) + } + + fn delete_ip(&self) -> NetworkFuture<'_, ()> { + Box::pin(async move { + if !self.has_ip().await? { + return Ok(()); + } + self.mutate("del").await + }) + } + + fn send_arp(&self) -> NetworkFuture<'_, ()> { + Box::pin(async move { + if !self.ip.is_ipv4() { + return Err(NetworkError { + operation: "garp", + class: "not_ipv4", + }); + } + let ip = self.ip.to_string(); + let args = ["-c", "1", "-U", "-I", self.interface.as_ref(), ip.as_str()]; + for _ in 0..self.burst_count { + let direct = run_command("arping", &args, "garp").await; + if matches!(direct, Ok(Output { success: true, .. })) { + continue; + } + // Go tries both a library GARP and sudo arping for each packet. + // The CLI path is retained here; a direct arping success or + // noninteractive sudo success counts as one logical packet. + let sudo_args = [ + "-n", + "arping", + "-c", + "1", + "-U", + "-I", + self.interface.as_ref(), + ip.as_str(), + ]; + let sudo = run_command("sudo", &sudo_args, "garp").await?; + if !sudo.success { + return Err(NetworkError { + operation: "garp", + class: "arping_failed", + }); + } + } + Ok(()) + }) + } +} + +struct Output { + success: bool, + stdout: String, + stderr: String, +} + +async fn run_command( + program: &str, + args: &[&str], + operation: &'static str, +) -> Result { + let mut command = Command::new(program); + command.args(args).env("LC_ALL", "C").kill_on_drop(true); + let output = tokio::time::timeout(COMMAND_TIMEOUT, command.output()) + .await + .map_err(|_| NetworkError { + operation, + class: "timeout", + })? + .map_err(|error| NetworkError { + operation, + class: if error.kind() == std::io::ErrorKind::NotFound { + "command_missing" + } else { + "command_io" + }, + })?; + Ok(Output { + success: output.status.success(), + stdout: String::from_utf8_lossy(&output.stdout).into_owned(), + stderr: String::from_utf8_lossy(&output.stderr).into_owned(), + }) +} + +fn contains_exact_address(output: &str, ip: IpAddr, prefix_len: u8) -> bool { + output.lines().any(|line| { + let mut words = line.split_whitespace(); + while let Some(word) = words.next() { + if word != "inet" && word != "inet6" { + continue; + } + let Some((address, prefix)) = words.next().and_then(|value| value.split_once('/')) + else { + continue; + }; + if address.parse::().ok() == Some(ip) + && prefix.parse::().ok() == Some(prefix_len) + { + return true; + } + } + false + }) +} + +#[cfg(test)] +mod tests { + use super::contains_exact_address; + use std::net::IpAddr; + + #[test] + fn exact_ip_and_prefix_match_is_strict() { + let output = "2: eth0 inet 192.0.2.5/24 brd 192.0.2.255 scope global eth0\n\ + 2: eth0 inet6 2001:db8::5/64 scope global\n"; + let v4: IpAddr = "192.0.2.5" + .parse() + .unwrap_or_else(|error| unreachable!("test IP: {error}")); + let v6: IpAddr = "2001:db8::5" + .parse() + .unwrap_or_else(|error| unreachable!("test IP: {error}")); + assert!(contains_exact_address(output, v4, 24)); + assert!(!contains_exact_address(output, v4, 32)); + assert!(contains_exact_address(output, v6, 64)); + assert!(!contains_exact_address(output, v6, 128)); + } +} diff --git a/rust/crates/tiproxy-rs/src/vip_owner.rs b/rust/crates/tiproxy-rs/src/vip_owner.rs new file mode 100644 index 000000000..a6e30a75e --- /dev/null +++ b/rust/crates/tiproxy-rs/src/vip_owner.rs @@ -0,0 +1,443 @@ +// Copyright 2026 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Dedicated, process-local VIP election with an awaited pre-close boundary. + +use std::sync::Arc; +use std::time::Duration; + +use control_config::VipConfig; +use control_etcd::{ElectionConfig, ElectionSession, ElectionState}; +use control_external::EtcdClientConfig; +use control_plane::logging::{self, Level}; +use control_plane::{ControlModule, LifecyclePhase, ModuleContext, ModuleError, ModuleFuture}; +use control_topology::ElectionOwnerHistory; +use tokio::sync::watch; + +use crate::vip_network::{NetworkError, NetworkOperation}; + +const TTL_SECONDS: i64 = 3; +const RETRY_DELAY: Duration = Duration::from_millis(500); +const PRE_CLOSE_TIMEOUT: Duration = Duration::from_secs(15); + +pub(super) struct VipModule { + network: Arc, + etcd: EtcdClientConfig, + election: ElectionConfig, + election_name: String, + refresh_count: u64, + stop: watch::Receiver, + closed: watch::Sender>>, + history: Arc, +} + +pub(super) struct VipHandle { + network: Arc, + stop: watch::Sender, + closed: watch::Receiver>>, +} + +impl VipModule { + pub fn new( + config: &VipConfig, + member_id: String, + process_id: &str, + etcd: EtcdClientConfig, + network: Arc, + history: Arc, + ) -> Result<(Self, VipHandle), String> { + let election_name = config.election_name(); + let session_key = format!("/tiproxy/vip/{}/session/{process_id}", config.ip); + let election = + ElectionConfig::new(election_name.clone(), member_id, session_key, TTL_SECONDS) + .map_err(|error| format!("validate VIP election: {error}"))?; + let (stop_tx, stop) = watch::channel(false); + let (closed, closed_rx) = watch::channel(None); + Ok(( + Self { + network: Arc::clone(&network), + etcd, + election, + election_name, + refresh_count: config.garp_refresh_count, + stop, + closed, + history, + }, + VipHandle { + network, + stop: stop_tx, + closed: closed_rx, + }, + )) + } + + async fn serve( + mut self, + mut lifecycle: watch::Receiver, + owner: control_plane::OwnerToken, + ) -> Result<(), ModuleError> { + // Go removes a stale local VIP before entering the election. Failing + // here is safer than campaigning with an address this process cannot + // remove on retirement. + self.network.delete_ip().await.map_err(network_failure)?; + // The module is spawned during guarded startup. A winner must not + // advertise this node before SQL serving has crossed the ready gate. + while lifecycle.borrow().phase != LifecyclePhase::Ready { + if *self.stop.borrow() || shutting_down(&lifecycle) { + return Ok(()); + } + tokio::select! { + _ = self.stop.changed() => return Ok(()), + changed = lifecycle.changed() => { + if changed.is_err() { + return Ok(()); + } + } + } + } + loop { + if *self.stop.borrow() || shutting_down(&lifecycle) { + return Ok(()); + } + let campaign = tokio::select! { + biased; + _ = self.stop.changed() => return Ok(()), + _ = lifecycle.changed() => { + if shutting_down(&lifecycle) { + return Ok(()); + } + continue; + }, + result = ElectionSession::campaign(owner.clone(), self.etcd.clone(), self.election.clone()) => result, + }; + let mut session = match campaign { + Ok(session) => session, + Err(error) => { + logging::emit(Level::Warn, &format!("VIP campaign retry: {error}")); + tokio::select! { + _ = self.stop.changed() => return Ok(()), + _ = lifecycle.changed() => continue, + () = tokio::time::sleep(RETRY_DELAY) => continue, + } + } + }; + let _owned = OwnedElection::new(Arc::clone(&self.history), self.election_name.clone()); + // If pre-close races a Linux command, dropping its future kills the + // child; deletion below covers a command that already completed. + let add = tokio::select! { + biased; + _ = self.stop.changed() => None, + _ = lifecycle.changed() => None, + result = self.network.add_ip() => Some(result), + }; + if let Some(Err(error)) = add { + self.network.delete_ip().await.map_err(network_failure)?; + let _ = session.shutdown().await; + return Err(network_failure(error)); + } + if add.is_some() { + let first_arp = tokio::select! { + biased; + _ = self.stop.changed() => None, + _ = lifecycle.changed() => None, + result = self.network.send_arp() => Some(result), + }; + if let Some(Err(error)) = first_arp { + logging::emit( + Level::Warn, + &format!( + "VIP first GARP burst failed: {} / {}", + error.operation, error.class + ), + ); + } + logging::emit(Level::Info, "VIP election won and address bound"); + if first_arp.is_some() { + Box::pin(self.maintain(&mut session, &mut lifecycle)).await; + } + } + // This awaited delete is the retirement boundary. Never resign the + // lease while our local interface still has the VIP. + self.network.delete_ip().await.map_err(network_failure)?; + let _ = session.shutdown().await; + logging::emit(Level::Info, "VIP retired and address removed"); + if *self.stop.borrow() || shutting_down(&lifecycle) { + return Ok(()); + } + tokio::select! { + _ = self.stop.changed() => return Ok(()), + _ = lifecycle.changed() => {}, + () = tokio::time::sleep(RETRY_DELAY) => {} + } + } + } + + async fn maintain( + &mut self, + session: &mut ElectionSession, + lifecycle: &mut watch::Receiver, + ) { + let mut heartbeat = tokio::time::interval(Duration::from_secs(1)); + heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + let mut refresh = tokio::time::interval(Duration::from_secs(1)); + refresh.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + refresh.tick().await; // the first burst was sent synchronously above + let mut remaining = self.refresh_count; + while !*self.stop.borrow() && !shutting_down(lifecycle) { + enum Event { + Stop, + Heartbeat, + Watch, + Refresh, + } + let event = tokio::select! { + biased; + _ = self.stop.changed() => Event::Stop, + _ = lifecycle.changed() => Event::Stop, + _ = heartbeat.tick() => Event::Heartbeat, + result = session.watch_once() => { + if result.is_err() { Event::Stop } else { Event::Watch } + }, + _ = refresh.tick(), if remaining > 0 => Event::Refresh, + }; + match event { + Event::Stop => break, + Event::Heartbeat => { + if session.keep_alive().await.is_err() { + // Uncertain ownership cannot leave an address bound + // until a remote lease may expire and a peer takes it. + break; + } + } + Event::Watch => {} + Event::Refresh => { + remaining -= 1; + let send = tokio::select! { + biased; + _ = self.stop.changed() => break, + _ = lifecycle.changed() => break, + result = self.network.send_arp() => result, + }; + if let Err(error) = send { + logging::emit( + Level::Warn, + &format!( + "VIP GARP refresh stopped: {} / {}", + error.operation, error.class + ), + ); + remaining = 0; + } + } + } + if matches!( + session.snapshot().state, + ElectionState::Uncertain | ElectionState::Retired | ElectionState::Stopped + ) { + break; + } + } + } +} + +impl ControlModule for VipModule { + fn name(&self) -> &'static str { + "vip_owner" + } + + fn run(self: Box, context: ModuleContext) -> ModuleFuture { + Box::pin(async move { + let network = Arc::clone(&self.network); + let closed = self.closed.clone(); + let result = Box::pin(self.serve(context.lifecycle(), context.owner().clone())).await; + // A final delete covers an interrupted add and gives pre-close a + // direct result even if the election loop failed before it won. + let cleanup = network.delete_ip().await; + closed.send_replace(Some(cleanup)); + match (result, cleanup) { + (Err(error), _) => Err(error), + (Ok(()), Err(error)) => Err(network_failure(error)), + (Ok(()), Ok(())) => Ok(()), + } + }) + } +} + +impl VipHandle { + /// Removes the local VIP and waits for the actor to retire before the SQL + /// listener enters drain. A timed-out actor is never reported as closed: + /// it may still have an in-flight add after a direct deletion. + pub async fn pre_close(&mut self) -> Result<(), NetworkError> { + self.stop.send_replace(true); + let wait = async { + loop { + if let Some(result) = *self.closed.borrow() { + return result; + } + if self.closed.changed().await.is_err() { + return self.network.delete_ip().await; + } + } + }; + if let Ok(result) = tokio::time::timeout(PRE_CLOSE_TIMEOUT, wait).await { + match result { + Ok(()) => Ok(()), + Err(_) => self.network.delete_ip().await, + } + } else { + self.network.delete_ip().await?; + Err(NetworkError { + operation: "pre_close", + class: "actor_timeout", + }) + } + } +} + +struct OwnedElection { + history: Arc, + key: String, +} + +impl OwnedElection { + fn new(history: Arc, key: String) -> Self { + history.won(&key); + Self { history, key } + } +} + +impl Drop for OwnedElection { + fn drop(&mut self) { + self.history.retired(&self.key); + } +} + +fn shutting_down(lifecycle: &watch::Receiver) -> bool { + matches!( + lifecycle.borrow().phase, + LifecyclePhase::Quiescing + | LifecyclePhase::Draining + | LifecyclePhase::Stopping + | LifecyclePhase::Stopped + | LifecyclePhase::Failed + ) +} + +fn network_failure(error: NetworkError) -> ModuleError { + logging::emit( + Level::Error, + &format!( + "VIP network operation failed: {} / {}", + error.operation, error.class + ), + ); + ModuleError { + module: "vip_owner", + error_class: "network_failed", + } +} + +#[cfg(test)] +mod tests { + use std::future::Future; + use std::pin::Pin; + use std::sync::Arc; + use std::sync::atomic::{AtomicUsize, Ordering}; + + use tokio::sync::watch; + + use super::VipHandle; + use crate::vip_network::{NetworkError, NetworkOperation}; + + type NetworkFuture<'a, T> = Pin> + Send + 'a>>; + + #[derive(Default)] + struct FakeNetwork { + deletes: AtomicUsize, + } + + impl NetworkOperation for FakeNetwork { + fn has_ip(&self) -> NetworkFuture<'_, bool> { + Box::pin(async { Ok(false) }) + } + + fn add_ip(&self) -> NetworkFuture<'_, ()> { + Box::pin(async { Ok(()) }) + } + + fn delete_ip(&self) -> NetworkFuture<'_, ()> { + Box::pin(async { + self.deletes.fetch_add(1, Ordering::SeqCst); + Ok(()) + }) + } + + fn send_arp(&self) -> NetworkFuture<'_, ()> { + Box::pin(async { Ok(()) }) + } + } + + #[tokio::test] + async fn pre_close_waits_for_actor_retirement_before_reporting_success() { + let network = Arc::new(FakeNetwork::default()); + let (stop, mut stop_rx) = watch::channel(false); + let (closed, closed_rx) = watch::channel(None); + let mut handle = VipHandle { + network: network.clone(), + stop, + closed: closed_rx, + }; + let close = tokio::spawn(async move { handle.pre_close().await }); + assert!(stop_rx.changed().await.is_ok()); + assert!(*stop_rx.borrow()); + assert!(!close.is_finished(), "SQL drain must await VIP retirement"); + closed.send_replace(Some(Ok(()))); + assert!(matches!(close.await, Ok(Ok(())))); + assert_eq!(network.deletes.load(Ordering::SeqCst), 0); + } + + #[tokio::test] + async fn pre_close_deletes_after_actor_exits_without_an_ack() { + let network = Arc::new(FakeNetwork::default()); + let (stop, _stop_rx) = watch::channel(false); + let (closed, closed_rx) = watch::channel(None); + drop(closed); + let mut handle = VipHandle { + network: network.clone(), + stop, + closed: closed_rx, + }; + assert!(handle.pre_close().await.is_ok()); + assert_eq!(network.deletes.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn pre_close_retries_delete_after_actor_cleanup_failure() { + let network = Arc::new(FakeNetwork::default()); + let (stop, _stop_rx) = watch::channel(false); + let (closed, closed_rx) = watch::channel(Some(Err(NetworkError { + operation: "del", + class: "ip_failed", + }))); + let mut handle = VipHandle { + network: network.clone(), + stop, + closed: closed_rx, + }; + assert!(handle.pre_close().await.is_ok()); + assert_eq!(network.deletes.load(Ordering::SeqCst), 1); + drop(closed); + } +} From 63e736000dd55c71afffb02573d161639069000d Mon Sep 17 00:00:00 2001 From: CodexM5 Date: Sat, 26 Sep 2026 09:50:47 +0800 Subject: [PATCH 03/14] Keep VIP lease alive through network commands --- rust/crates/tiproxy-rs/src/vip_owner.rs | 148 +++++++++++++++++------- 1 file changed, 104 insertions(+), 44 deletions(-) diff --git a/rust/crates/tiproxy-rs/src/vip_owner.rs b/rust/crates/tiproxy-rs/src/vip_owner.rs index a6e30a75e..1fc737b7d 100644 --- a/rust/crates/tiproxy-rs/src/vip_owner.rs +++ b/rust/crates/tiproxy-rs/src/vip_owner.rs @@ -134,43 +134,27 @@ impl VipModule { } }; let _owned = OwnedElection::new(Arc::clone(&self.history), self.election_name.clone()); - // If pre-close races a Linux command, dropping its future kills the - // child; deletion below covers a command that already completed. - let add = tokio::select! { - biased; - _ = self.stop.changed() => None, - _ = lifecycle.changed() => None, - result = self.network.add_ip() => Some(result), - }; - if let Some(Err(error)) = add { - self.network.delete_ip().await.map_err(network_failure)?; + // Linux commands can outlast the 3s lease. Keep it alive while + // binding; dropping an in-flight command kills its child, and the + // retirement delete below covers a command that already finished. + let add = self.bind_while_owned(&mut session, &mut lifecycle).await; + if let Err(error) = add { + self.delete_while_owned(&mut session) + .await + .map_err(network_failure)?; let _ = session.shutdown().await; return Err(network_failure(error)); } - if add.is_some() { - let first_arp = tokio::select! { - biased; - _ = self.stop.changed() => None, - _ = lifecycle.changed() => None, - result = self.network.send_arp() => Some(result), - }; - if let Some(Err(error)) = first_arp { - logging::emit( - Level::Warn, - &format!( - "VIP first GARP burst failed: {} / {}", - error.operation, error.class - ), - ); - } + if matches!(add, Ok(true)) { logging::emit(Level::Info, "VIP election won and address bound"); - if first_arp.is_some() { - Box::pin(self.maintain(&mut session, &mut lifecycle)).await; - } + Box::pin(self.maintain(&mut session, &mut lifecycle)).await; } // This awaited delete is the retirement boundary. Never resign the - // lease while our local interface still has the VIP. - self.network.delete_ip().await.map_err(network_failure)?; + // lease while our local interface still has the VIP. Keepalive + // continues because a slow `ip`/`sudo` command can outlast TTL. + self.delete_while_owned(&mut session) + .await + .map_err(network_failure)?; let _ = session.shutdown().await; logging::emit(Level::Info, "VIP retired and address removed"); if *self.stop.borrow() || shutting_down(&lifecycle) { @@ -184,6 +168,60 @@ impl VipModule { } } + async fn bind_while_owned( + &mut self, + session: &mut ElectionSession, + lifecycle: &mut watch::Receiver, + ) -> Result { + let network = Arc::clone(&self.network); + let mut add = Box::pin(network.add_ip()); + let mut heartbeat = tokio::time::interval(Duration::from_secs(1)); + heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + tokio::select! { + biased; + _ = self.stop.changed() => return Ok(false), + changed = lifecycle.changed() => { + if changed.is_err() || shutting_down(lifecycle) { + return Ok(false); + } + }, + _ = heartbeat.tick() => { + if session.keep_alive().await.is_err() + || session.snapshot().state != ElectionState::Leader + { + return Ok(false); + } + }, + result = &mut add => return result.map(|()| true), + } + } + } + + async fn delete_while_owned(&self, session: &mut ElectionSession) -> Result<(), NetworkError> { + let network = Arc::clone(&self.network); + let mut delete = Box::pin(network.delete_ip()); + let mut heartbeat = tokio::time::interval_at( + tokio::time::Instant::now() + Duration::from_secs(1), + Duration::from_secs(1), + ); + heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + let mut lease_alive = session.snapshot().state == ElectionState::Leader; + loop { + tokio::select! { + biased; + result = &mut delete => return result, + _ = heartbeat.tick(), if lease_alive => { + lease_alive = session.keep_alive().await.is_ok() + && session.snapshot().state == ElectionState::Leader; + if !lease_alive { + logging::emit(Level::Warn, "VIP lease uncertain during address removal"); + } + }, + } + } + } + async fn maintain( &mut self, session: &mut ElectionSession, @@ -191,9 +229,10 @@ impl VipModule { ) { let mut heartbeat = tokio::time::interval(Duration::from_secs(1)); heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); - let mut refresh = tokio::time::interval(Duration::from_secs(1)); - refresh.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); - refresh.tick().await; // the first burst was sent synchronously above + let network = Arc::clone(&self.network); + let mut arp = Some(network.send_arp()); + let mut first_arp = true; + let mut refresh: Option = None; let mut remaining = self.refresh_count; while !*self.stop.borrow() && !shutting_down(lifecycle) { enum Event { @@ -201,16 +240,30 @@ impl VipModule { Heartbeat, Watch, Refresh, + Arp(Result<(), NetworkError>), } let event = tokio::select! { biased; _ = self.stop.changed() => Event::Stop, _ = lifecycle.changed() => Event::Stop, _ = heartbeat.tick() => Event::Heartbeat, + result = async { + if let Some(arp) = arp.as_mut() { + arp.await + } else { + std::future::pending().await + } + }, if arp.is_some() => Event::Arp(result), result = session.watch_once() => { if result.is_err() { Event::Stop } else { Event::Watch } }, - _ = refresh.tick(), if remaining > 0 => Event::Refresh, + () = async { + if let Some(refresh) = refresh.as_mut() { + refresh.tick().await; + } else { + std::future::pending::<()>().await; + } + }, if refresh.is_some() && arp.is_none() && remaining > 0 => Event::Refresh, }; match event { Event::Stop => break, @@ -224,21 +277,28 @@ impl VipModule { Event::Watch => {} Event::Refresh => { remaining -= 1; - let send = tokio::select! { - biased; - _ = self.stop.changed() => break, - _ = lifecycle.changed() => break, - result = self.network.send_arp() => result, - }; - if let Err(error) = send { + arp = Some(network.send_arp()); + } + Event::Arp(result) => { + arp = None; + if let Err(error) = result { logging::emit( Level::Warn, &format!( - "VIP GARP refresh stopped: {} / {}", + "VIP GARP burst failed: {} / {}", error.operation, error.class ), ); - remaining = 0; + if !first_arp { + remaining = 0; + } + } + if first_arp { + first_arp = false; + refresh = Some(tokio::time::interval_at( + tokio::time::Instant::now() + Duration::from_secs(1), + Duration::from_secs(1), + )); } } } From 1ab2ad5fb928ab9cd0c93ce4aa7695e8168862e6 Mon Sep 17 00:00:00 2001 From: CodexM5 Date: Sat, 26 Sep 2026 10:08:19 +0800 Subject: [PATCH 04/14] Add privileged two-node Linux VIP acceptance harness --- .github/workflows/rust-vip-linux.yml | 68 ++++ tests/dataplane/integration/vip-standalone.sh | 318 ++++++++++++++++++ 2 files changed, 386 insertions(+) create mode 100644 .github/workflows/rust-vip-linux.yml create mode 100644 tests/dataplane/integration/vip-standalone.sh diff --git a/.github/workflows/rust-vip-linux.yml b/.github/workflows/rust-vip-linux.yml new file mode 100644 index 000000000..fce6b331b --- /dev/null +++ b/.github/workflows/rust-vip-linux.yml @@ -0,0 +1,68 @@ +name: Rust VIP Linux acceptance + +on: + pull_request: + paths: + - .github/workflows/rust-vip-linux.yml + - rust/crates/control-config/** + - rust/crates/tiproxy-rs/** + - tests/dataplane/integration/vip-standalone.sh + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: rust-vip-linux-${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + +jobs: + two-node-vip: + name: Two Linux standalone VIP owners + runs-on: ubuntu-24.04 + timeout-minutes: 45 + env: + DATAPLANE_ARTIFACT_ROOT: ${{ github.workspace }}/tests/dataplane/integration/artifacts + steps: + - name: Check out source + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + - name: Select repository Go toolchain + uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0 + with: + go-version-file: go.mod + cache: true + - name: Install checksum-pinned TiUP + run: | + source tests/dataplane/integration/versions.env + archive="$RUNNER_TEMP/tiup-v${TIUP_VERSION}-linux-amd64.tar.gz" + install_root="$RUNNER_TEMP/tiup-v${TIUP_VERSION}" + curl --fail --location --retry 3 --proto '=https' --tlsv1.2 \ + --output "$archive" \ + "https://github.com/pingcap/tiup/releases/download/v${TIUP_VERSION}/tiup-v${TIUP_VERSION}-linux-amd64.tar.gz" + echo "$TIUP_LINUX_AMD64_SHA256 $archive" | sha256sum -c - + mkdir -p "$install_root" + tar -xzf "$archive" -C "$install_root" + "$install_root/bin/tiup" --version + echo "$install_root/bin" >>"$GITHUB_PATH" + echo "TIUP_HOME=$install_root" >>"$GITHUB_ENV" + - name: Preinstall frozen TiUP components + run: bash tests/dataplane/integration/warm-tiup-components.sh + - name: Install Linux network and Rust build prerequisites + run: | + sudo apt-get update + sudo apt-get install --yes iproute2 arping protobuf-compiler + command -v mysql + - name: Build standalone Rust executable + run: make rust-build + - name: Prove VIP ownership and SQL continuity + run: bash tests/dataplane/integration/vip-standalone.sh + - name: Upload node and election evidence + if: always() + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: rust-vip-linux-${{ github.run_id }} + path: tests/dataplane/integration/artifacts/vip-*/diagnostics/** + if-no-files-found: warn + retention-days: 14 diff --git a/tests/dataplane/integration/vip-standalone.sh b/tests/dataplane/integration/vip-standalone.sh new file mode 100644 index 000000000..a9f86d28a --- /dev/null +++ b/tests/dataplane/integration/vip-standalone.sh @@ -0,0 +1,318 @@ +#!/usr/bin/env bash +# Copyright 2026 PingCAP, Inc. +# SPDX-License-Identifier: Apache-2.0 + +# Linux-only, privileged two-node VIP smoke. Each Rust process owns its own +# network namespace, so both may bind the same SQL port and only the elected +# process may bind the shared VIP on its eth0. The bridge gives both processes +# a real PD/TiDB and gives the host a route to query the elected VIP. +set -euo pipefail + +if [[ $(uname -s) != Linux ]]; then + echo 'VIP acceptance requires Linux network namespaces' >&2 + exit 2 +fi +for command in ip arping tiup mysql curl; do + command -v "$command" >/dev/null || { echo "missing $command" >&2; exit 2; } +done + +script_dir=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) +repo_root=$(cd "$script_dir/../../.." && pwd) +source "$script_dir/versions.env" +rust_binary="$repo_root/rust/target/debug/tiproxy-rs" +[[ -x $rust_binary ]] || { echo "missing $rust_binary" >&2; exit 2; } + +run_id=${GITHUB_RUN_ID:-local-$$} +run_dir=${DATAPLANE_ARTIFACT_ROOT:-$script_dir/artifacts}/vip-$run_id +mkdir -p "$run_dir/diagnostics" +run_dir=$(cd "$run_dir" && pwd) +tag=vip-$run_id +bridge=br-vip-test +ns_a=vip-a-$run_id +ns_b=vip-b-$run_id +link_id=${run_id: -6} +host_a=vha-$link_id +host_b=vhb-$link_id +host_ip=198.19.99.1 +vip_ip=198.19.99.100 +node_a_ip=198.19.99.11 +node_b_ip=198.19.99.12 +offset=13000 +pd_port=$((2379 + offset)) +tidb_port=$((4000 + offset)) +tiup_pid= +monitor_pid= + +owned_rust_pid() { + local node=$1 pid=$2 + [[ $pid =~ ^[0-9]+$ ]] || return 1 + sudo ps -ww -p "$pid" -o args= 2>/dev/null | + grep -Fq -- "$run_dir/$node.toml" +} + +cleanup() { + local status=$? + trap - EXIT + set +e + [[ -n $monitor_pid ]] && kill "$monitor_pid" 2>/dev/null + [[ -s $run_dir/diagnostics/overlap.log ]] && status=1 + for node in a b; do + pidfile=$run_dir/$node.pid + if [[ -f $pidfile ]]; then + pid=$(cat "$pidfile") + owned_rust_pid "$node" "$pid" && sudo kill -INT "$pid" 2>/dev/null + fi + done + sleep 2 + for node in a b; do + pidfile=$run_dir/$node.pid + if [[ -f $pidfile ]]; then + pid=$(cat "$pidfile") + owned_rust_pid "$node" "$pid" && sudo kill -KILL "$pid" 2>/dev/null + fi + done + if [[ -n $tiup_pid ]]; then + tiup clean "$tag" >>"$run_dir/diagnostics/cleanup.log" 2>&1 + kill -INT "$tiup_pid" 2>/dev/null + fi + sudo ip netns del "$ns_a" 2>/dev/null + sudo ip netns del "$ns_b" 2>/dev/null + sudo ip link del "$bridge" 2>/dev/null + echo "exit=$status" >>"$run_dir/diagnostics/result.txt" + exit "$status" +} +trap cleanup EXIT + +sudo ip link add "$bridge" type bridge +sudo ip addr add "$host_ip/24" dev "$bridge" +sudo ip link set "$bridge" up + +attach_node() { + local namespace=$1 host_link=$2 address=$3 peer=peer0 + sudo ip link add "$host_link" type veth peer name "$peer" + sudo ip link set "$peer" netns "$namespace" + sudo ip link set "$host_link" master "$bridge" + sudo ip link set "$host_link" up + sudo ip -n "$namespace" link set "$peer" name eth0 + sudo ip -n "$namespace" link set eth0 up + sudo ip -n "$namespace" addr add "$address/24" dev eth0 + sudo ip -n "$namespace" route add default via "$host_ip" +} + +sudo ip netns add "$ns_a" +sudo ip netns add "$ns_b" +sudo ip -n "$ns_a" link set lo up +sudo ip -n "$ns_b" link set lo up +attach_node "$ns_a" "$host_a" "$node_a_ip" +attach_node "$ns_b" "$host_b" "$node_b_ip" + +# The backend playground is deliberately on the host's bridge address so a +# namespace's 127.0.0.1 can never accidentally bypass the veth network. +tiup "playground:v${TIUP_VERSION}" "$TIDB_VERSION" --tag "$tag" \ + --without-monitor --host "$host_ip" --port-offset "$offset" \ + --pd 1 --kv 1 --db 1 --tiflash 0 --tiproxy 0 \ + >"$run_dir/diagnostics/tiup.log" 2>&1 & +tiup_pid=$! + +wait_mysql() { + local host=$1 port=$2 label=$3 + for _ in {1..120}; do + if mysql --protocol=TCP --connect-timeout=1 -h "$host" -P "$port" \ + -u root --batch --skip-column-names -e 'SELECT 1' 2>/dev/null | + grep -qx 1; then + echo "$(date +%s.%N) $label SELECT 1 PASS" >>"$run_dir/diagnostics/events.log" + return 0 + fi + kill -0 "$tiup_pid" 2>/dev/null || break + sleep 1 + done + echo "$label SQL query failed" >&2 + return 1 +} +wait_mysql "$host_ip" "$tidb_port" backend + +write_config() { + local node=$1 address=$2 + mkdir -p "$run_dir/$node-work" + cat >"$run_dir/$node.toml" <"$1"; shift; exec "$@"' \ + sh "$run_dir/$node.pid" "$rust_binary" --standalone \ + --config "$run_dir/$node.toml" --health-port 8080 \ + >"$run_dir/diagnostics/$node.log" 2>&1 & +} +has_vip() { + local namespace=$1 + sudo ip -n "$namespace" -o addr show dev eth0 2>/dev/null | + grep -Fq "inet $vip_ip/24 " +} + +monitor_overlap() { + local a_bound b_bound + while :; do + a_bound=0 + b_bound=0 + has_vip "$ns_a" && a_bound=1 + has_vip "$ns_b" && b_bound=1 + if ((a_bound + b_bound > 1)); then + echo "$(date +%s.%N) overlap" >>"$run_dir/diagnostics/overlap.log" + return 1 + fi + sleep 0.1 + done +} + +assert_no_overlap() { + if [[ -s $run_dir/diagnostics/overlap.log ]]; then + echo 'two VIP holders observed by background sampler' >&2 + return 1 + fi +} + +monitor_overlap & +monitor_pid=$! +start_node a "$ns_a" +start_node b "$ns_b" + +wait_ready() { + local node=$1 namespace=$2 + for _ in {1..60}; do + if sudo ip netns exec "$namespace" curl --fail --silent --max-time 1 \ + 'http://127.0.0.1:8080/health' >/dev/null; then + echo "$(date +%s.%N) $node ready" >>"$run_dir/diagnostics/events.log" + return 0 + fi + if [[ -f $run_dir/$node.pid ]]; then + sudo kill -0 "$(cat "$run_dir/$node.pid")" 2>/dev/null || break + fi + sleep 1 + done + echo "$node Rust process did not become ready" >&2 + return 1 +} +wait_ready a "$ns_a" +wait_ready b "$ns_b" + +snapshot() { + local label=$1 + sudo ip -j -n "$ns_a" addr show >"$run_dir/diagnostics/$label-a-addr.json" + sudo ip -j -n "$ns_b" addr show >"$run_dir/diagnostics/$label-b-addr.json" + local ctl=${TIUP_HOME:-$HOME/.tiup}/components/ctl/$TIDB_VERSION/etcdctl + if [[ -x $ctl ]]; then + ETCDCTL_API=3 "$ctl" --endpoints "http://$host_ip:$pd_port" \ + get "/tiproxy/vip/$vip_ip/owner" --prefix -w json \ + >"$run_dir/diagnostics/$label-election.json" 2>&1 || true + fi +} + +wait_single_owner() { + local label=$1 expected=${2:-either} a_bound b_bound + for _ in {1..45}; do + a_bound=0 + b_bound=0 + has_vip "$ns_a" && a_bound=1 + has_vip "$ns_b" && b_bound=1 + if ((a_bound + b_bound == 1)); then + if [[ $expected == either || ($expected == a && $a_bound == 1) || + ($expected == b && $b_bound == 1) ]]; then + snapshot "$label" + echo "$(date +%s.%N) $label owner=$([[ $a_bound == 1 ]] && echo a || echo b)" \ + >>"$run_dir/diagnostics/events.log" + [[ $a_bound == 1 ]] && echo a || echo b + return 0 + fi + fi + if ((a_bound + b_bound > 1)); then + snapshot "$label-overlap" + echo 'two VIP holders observed' >&2 + return 1 + fi + sleep 1 + done + snapshot "$label-timeout" + echo "single VIP owner not observed at $label" >&2 + return 1 +} + +owner=$(wait_single_owner initial) +assert_no_overlap +wait_mysql "$vip_ip" 6000 initial-vip +if [[ $owner == a ]]; then + former=a + new=b +else + former=b + new=a +fi +former_pid=$(cat "$run_dir/$former.pid") +sudo kill -INT "$former_pid" +wait_single_owner after-controlled-close "$new" >/dev/null +assert_no_overlap +wait_mysql "$vip_ip" 6000 controlled-failover +for _ in {1..30}; do + sudo kill -0 "$former_pid" 2>/dev/null || break + sleep 1 +done +if sudo kill -0 "$former_pid" 2>/dev/null; then + echo 'former VIP owner did not exit after SIGINT' >&2 + exit 1 +fi + +# A restarted former owner removes any stale local address before campaigning. +if [[ $former == a ]]; then + start_node a "$ns_a" + wait_ready a "$ns_a" +else + start_node b "$ns_b" + wait_ready b "$ns_b" +fi +wait_single_owner after-restart "$new" >/dev/null +assert_no_overlap +wait_mysql "$vip_ip" 6000 restart + +# Model a whole-node crash: terminate its process and remove its network link. +# Process-only SIGKILL leaves an IP on a live Linux interface and is separately +# documented as a remaining Go/Rust operational fencing gap. +sudo kill -KILL "$(cat "$run_dir/$new.pid")" +if [[ $new == a ]]; then + sudo ip link del "$host_a" +else + sudo ip link del "$host_b" +fi +wait_single_owner after-node-link-loss "$former" >/dev/null +assert_no_overlap +wait_mysql "$vip_ip" 6000 node-failover +kill "$monitor_pid" 2>/dev/null || true +wait "$monitor_pid" 2>/dev/null || true +monitor_pid= +assert_no_overlap +echo 'PASS: one VIP holder and SQL continuity through controlled close, restart and node/link loss' From 310ab17ba6931fab3e0b9fdb4810f149c9c82d8f Mon Sep 17 00:00:00 2001 From: CodexM5 Date: Sat, 26 Sep 2026 10:21:49 +0800 Subject: [PATCH 05/14] tests: capture VIP SQL failures and prove namespace backend reachability --- tests/dataplane/integration/vip-standalone.sh | 20 ++++++++++++++++--- 1 file changed, 17 insertions(+), 3 deletions(-) diff --git a/tests/dataplane/integration/vip-standalone.sh b/tests/dataplane/integration/vip-standalone.sh index a9f86d28a..4f6cde70a 100644 --- a/tests/dataplane/integration/vip-standalone.sh +++ b/tests/dataplane/integration/vip-standalone.sh @@ -116,13 +116,15 @@ tiup_pid=$! wait_mysql() { local host=$1 port=$2 label=$3 + local result for _ in {1..120}; do - if mysql --protocol=TCP --connect-timeout=1 -h "$host" -P "$port" \ - -u root --batch --skip-column-names -e 'SELECT 1' 2>/dev/null | - grep -qx 1; then + if result=$(mysql --protocol=TCP --connect-timeout=1 --ssl-mode=DISABLED \ + -h "$host" -P "$port" -u root --batch --skip-column-names \ + -e 'SELECT 1' 2>&1) && [[ $result == 1 ]]; then echo "$(date +%s.%N) $label SELECT 1 PASS" >>"$run_dir/diagnostics/events.log" return 0 fi + printf '%s\n' "$result" >"$run_dir/diagnostics/$label-mysql-last-error.txt" kill -0 "$tiup_pid" 2>/dev/null || break sleep 1 done @@ -131,6 +133,18 @@ wait_mysql() { } wait_mysql "$host_ip" "$tidb_port" backend +# The runner must prove each network namespace can reach the same backend +# before a failed VIP query is attributed to routing in the Rust processes. +for namespace in "$ns_a" "$ns_b"; do + if ! sudo ip netns exec "$namespace" mysql --protocol=TCP \ + --connect-timeout=2 --ssl-mode=DISABLED -h "$host_ip" -P "$tidb_port" \ + -u root --batch --skip-column-names -e 'SELECT 1' \ + >"$run_dir/diagnostics/$namespace-backend-preflight.txt" 2>&1; then + echo "$namespace cannot query TiDB from its network namespace" >&2 + exit 1 + fi +done + write_config() { local node=$1 address=$2 mkdir -p "$run_dir/$node-work" From 22a09bf32e4046d45f5915a1dcdc1b370827fc8a Mon Sep 17 00:00:00 2001 From: CodexM5 Date: Sat, 26 Sep 2026 10:32:07 +0800 Subject: [PATCH 06/14] tests: assign distinct MACs to Linux VIP test nodes --- tests/dataplane/integration/vip-standalone.sh | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/tests/dataplane/integration/vip-standalone.sh b/tests/dataplane/integration/vip-standalone.sh index 4f6cde70a..5a79df86d 100644 --- a/tests/dataplane/integration/vip-standalone.sh +++ b/tests/dataplane/integration/vip-standalone.sh @@ -88,12 +88,17 @@ sudo ip addr add "$host_ip/24" dev "$bridge" sudo ip link set "$bridge" up attach_node() { - local namespace=$1 host_link=$2 address=$3 peer=peer0 + local namespace=$1 host_link=$2 address=$3 mac_suffix=$4 peer=peer0 sudo ip link add "$host_link" type veth peer name "$peer" sudo ip link set "$peer" netns "$namespace" + # GitHub's runner assigned the same generated MAC to both peers in the + # first acceptance runs. A shared MAC lets the bridge learn the wrong port + # for the VIP even though exactly one namespace owns its IP address. + sudo ip link set "$host_link" address "02:00:00:99:10:$mac_suffix" sudo ip link set "$host_link" master "$bridge" sudo ip link set "$host_link" up sudo ip -n "$namespace" link set "$peer" name eth0 + sudo ip -n "$namespace" link set eth0 address "02:00:00:99:20:$mac_suffix" sudo ip -n "$namespace" link set eth0 up sudo ip -n "$namespace" addr add "$address/24" dev eth0 sudo ip -n "$namespace" route add default via "$host_ip" @@ -103,8 +108,8 @@ sudo ip netns add "$ns_a" sudo ip netns add "$ns_b" sudo ip -n "$ns_a" link set lo up sudo ip -n "$ns_b" link set lo up -attach_node "$ns_a" "$host_a" "$node_a_ip" -attach_node "$ns_b" "$host_b" "$node_b_ip" +attach_node "$ns_a" "$host_a" "$node_a_ip" 11 +attach_node "$ns_b" "$host_b" "$node_b_ip" 12 # The backend playground is deliberately on the host's bridge address so a # namespace's 127.0.0.1 can never accidentally bypass the veth network. From be0ee6d836f96324c8560683508c91f7713297c3 Mon Sep 17 00:00:00 2001 From: CodexM5 Date: Sat, 26 Sep 2026 10:42:15 +0800 Subject: [PATCH 07/14] tests: capture VIP route health and PD topology on SQL failure --- tests/dataplane/integration/vip-standalone.sh | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/tests/dataplane/integration/vip-standalone.sh b/tests/dataplane/integration/vip-standalone.sh index 5a79df86d..294605718 100644 --- a/tests/dataplane/integration/vip-standalone.sh +++ b/tests/dataplane/integration/vip-standalone.sh @@ -133,6 +133,9 @@ wait_mysql() { kill -0 "$tiup_pid" 2>/dev/null || break sleep 1 done + if declare -F snapshot >/dev/null; then + snapshot "$label-failed" + fi echo "$label SQL query failed" >&2 return 1 } @@ -226,7 +229,8 @@ wait_ready() { local node=$1 namespace=$2 for _ in {1..60}; do if sudo ip netns exec "$namespace" curl --fail --silent --max-time 1 \ - 'http://127.0.0.1:8080/health' >/dev/null; then + 'http://127.0.0.1:8080/health' \ + >"$run_dir/diagnostics/$node-health-ready.json"; then echo "$(date +%s.%N) $node ready" >>"$run_dir/diagnostics/events.log" return 0 fi @@ -245,11 +249,20 @@ snapshot() { local label=$1 sudo ip -j -n "$ns_a" addr show >"$run_dir/diagnostics/$label-a-addr.json" sudo ip -j -n "$ns_b" addr show >"$run_dir/diagnostics/$label-b-addr.json" + sudo ip netns exec "$ns_a" curl --silent --max-time 2 \ + 'http://127.0.0.1:8080/health' \ + >"$run_dir/diagnostics/$label-a-health.json" || true + sudo ip netns exec "$ns_b" curl --silent --max-time 2 \ + 'http://127.0.0.1:8080/health' \ + >"$run_dir/diagnostics/$label-b-health.json" || true local ctl=${TIUP_HOME:-$HOME/.tiup}/components/ctl/$TIDB_VERSION/etcdctl if [[ -x $ctl ]]; then ETCDCTL_API=3 "$ctl" --endpoints "http://$host_ip:$pd_port" \ get "/tiproxy/vip/$vip_ip/owner" --prefix -w json \ >"$run_dir/diagnostics/$label-election.json" 2>&1 || true + ETCDCTL_API=3 "$ctl" --endpoints "http://$host_ip:$pd_port" \ + get '/topology/tidb/' --prefix -w json \ + >"$run_dir/diagnostics/$label-tidb-topology.json" 2>&1 || true fi } From ae4b63c28ffbecc200f02c9eb9dcee2a3604e320 Mon Sep 17 00:00:00 2001 From: CodexM5 Date: Sat, 26 Sep 2026 10:53:28 +0800 Subject: [PATCH 08/14] tests: expose debug local route rejection class in Linux VIP run --- rust/crates/dataplane/src/route_local.rs | 4 ++++ tests/dataplane/integration/vip-standalone.sh | 3 ++- 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/rust/crates/dataplane/src/route_local.rs b/rust/crates/dataplane/src/route_local.rs index abc0034a1..75095b71d 100644 --- a/rust/crates/dataplane/src/route_local.rs +++ b/rust/crates/dataplane/src/route_local.rs @@ -166,6 +166,10 @@ impl LocalRouteChannel { fn terminal(&self, error: RouteError) -> RouteAssignment { let category = error.category(); + #[cfg(debug_assertions)] + if std::env::var_os("TIPROXY_ROUTE_DIAGNOSTIC").is_some() { + eprintln!("local route terminal category={category}"); + } let (code, detail) = match error { RouteError::NoBackend | RouteError::WrappedNoBackend => ( RouteCode::NoBackend, diff --git a/tests/dataplane/integration/vip-standalone.sh b/tests/dataplane/integration/vip-standalone.sh index 294605718..34657ce7b 100644 --- a/tests/dataplane/integration/vip-standalone.sh +++ b/tests/dataplane/integration/vip-standalone.sh @@ -188,7 +188,8 @@ start_node() { # would not be a reliable signal target for the Rust shutdown path. sudo ip netns exec "$namespace" sh -c \ 'echo "$$" >"$1"; shift; exec "$@"' \ - sh "$run_dir/$node.pid" "$rust_binary" --standalone \ + sh "$run_dir/$node.pid" env TIPROXY_ROUTE_DIAGNOSTIC=1 \ + "$rust_binary" --standalone \ --config "$run_dir/$node.toml" --health-port 8080 \ >"$run_dir/diagnostics/$node.log" 2>&1 & } From ec151c3f368e9913de6a530b84383a1083d6be11 Mon Sep 17 00:00:00 2001 From: CodexM5 Date: Sat, 26 Sep 2026 11:03:44 +0800 Subject: [PATCH 09/14] tests: classify debug route acquisition failures in VIP smoke --- rust/crates/dataplane/src/session_engine.rs | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/rust/crates/dataplane/src/session_engine.rs b/rust/crates/dataplane/src/session_engine.rs index ff883d60a..2ac456919 100644 --- a/rust/crates/dataplane/src/session_engine.rs +++ b/rust/crates/dataplane/src/session_engine.rs @@ -2027,6 +2027,27 @@ impl Engine { acquired } Err(error) => { + #[cfg(debug_assertions)] + if std::env::var_os("TIPROXY_ROUTE_DIAGNOSTIC").is_some() { + let class = match &error { + AcquireError::NoBackend { .. } => "no_backend", + AcquireError::Routing { .. } => "routing", + AcquireError::BudgetExhausted { last_failure: None } => { + "budget_without_dial_failure" + } + AcquireError::BudgetExhausted { + last_failure: Some(_), + } => "budget_after_dial_failure", + AcquireError::MalformedAssignment { .. } => "malformed_assignment", + AcquireError::ClusterUnsupported { .. } => "cluster_unsupported", + AcquireError::Channel(RouteChannelError::Rejected) => "channel_rejected", + AcquireError::Channel(RouteChannelError::ControlLost) => "control_lost", + }; + eprintln!( + "local route acquisition failed class={class} stats={:?}", + route_engine.stats() + ); + } self.metrics.try_record(Observation::GetBackend { duration: acquisition_started.elapsed(), succeeded: false, From 78b0a9415a7db9e3a299e8ec24c0b70e02e46134 Mon Sep 17 00:00:00 2001 From: CodexM5 Date: Sat, 26 Sep 2026 11:15:49 +0800 Subject: [PATCH 10/14] test: supply TiDB keyspace label in standalone VIP acceptance --- rust/crates/dataplane/src/route_local.rs | 4 -- rust/crates/dataplane/src/session_engine.rs | 21 ---------- tests/dataplane/integration/vip-standalone.sh | 39 ++++++++++++++++++- 3 files changed, 37 insertions(+), 27 deletions(-) diff --git a/rust/crates/dataplane/src/route_local.rs b/rust/crates/dataplane/src/route_local.rs index 75095b71d..abc0034a1 100644 --- a/rust/crates/dataplane/src/route_local.rs +++ b/rust/crates/dataplane/src/route_local.rs @@ -166,10 +166,6 @@ impl LocalRouteChannel { fn terminal(&self, error: RouteError) -> RouteAssignment { let category = error.category(); - #[cfg(debug_assertions)] - if std::env::var_os("TIPROXY_ROUTE_DIAGNOSTIC").is_some() { - eprintln!("local route terminal category={category}"); - } let (code, detail) = match error { RouteError::NoBackend | RouteError::WrappedNoBackend => ( RouteCode::NoBackend, diff --git a/rust/crates/dataplane/src/session_engine.rs b/rust/crates/dataplane/src/session_engine.rs index 2ac456919..ff883d60a 100644 --- a/rust/crates/dataplane/src/session_engine.rs +++ b/rust/crates/dataplane/src/session_engine.rs @@ -2027,27 +2027,6 @@ impl Engine { acquired } Err(error) => { - #[cfg(debug_assertions)] - if std::env::var_os("TIPROXY_ROUTE_DIAGNOSTIC").is_some() { - let class = match &error { - AcquireError::NoBackend { .. } => "no_backend", - AcquireError::Routing { .. } => "routing", - AcquireError::BudgetExhausted { last_failure: None } => { - "budget_without_dial_failure" - } - AcquireError::BudgetExhausted { - last_failure: Some(_), - } => "budget_after_dial_failure", - AcquireError::MalformedAssignment { .. } => "malformed_assignment", - AcquireError::ClusterUnsupported { .. } => "cluster_unsupported", - AcquireError::Channel(RouteChannelError::Rejected) => "channel_rejected", - AcquireError::Channel(RouteChannelError::ControlLost) => "control_lost", - }; - eprintln!( - "local route acquisition failed class={class} stats={:?}", - route_engine.stats() - ); - } self.metrics.try_record(Observation::GetBackend { duration: acquisition_started.elapsed(), succeeded: false, diff --git a/tests/dataplane/integration/vip-standalone.sh b/tests/dataplane/integration/vip-standalone.sh index 34657ce7b..1601dd551 100644 --- a/tests/dataplane/integration/vip-standalone.sh +++ b/tests/dataplane/integration/vip-standalone.sh @@ -113,9 +113,14 @@ attach_node "$ns_b" "$host_b" "$node_b_ip" 12 # The backend playground is deliberately on the host's bridge address so a # namespace's 127.0.0.1 can never accidentally bypass the veth network. +cat >"$run_dir/tidb.toml" <<'CONFIG' +[labels] +keyspace = "vip-ks" +CONFIG tiup "playground:v${TIUP_VERSION}" "$TIDB_VERSION" --tag "$tag" \ --without-monitor --host "$host_ip" --port-offset "$offset" \ - --pd 1 --kv 1 --db 1 --tiflash 0 --tiproxy 0 \ + --pd 1 --kv 1 --db 1 --db.config "$run_dir/tidb.toml" \ + --tiflash 0 --tiproxy 0 \ >"$run_dir/diagnostics/tiup.log" 2>&1 & tiup_pid=$! @@ -153,6 +158,36 @@ for namespace in "$ns_a" "$ns_b"; do fi done +# Standalone metering requires a nonempty backend keyspace before forwarding +# SQL. TiDB publishes this label in its classic /topology/tidb info record. +ctl=${TIUP_HOME:-$HOME/.tiup}/components/ctl/$TIDB_VERSION/etcdctl +[[ -x $ctl ]] || { echo "missing $ctl" >&2; exit 2; } +topology="$run_dir/diagnostics/backend-topology.json" +registered=0 +for _ in {1..30}; do + ETCDCTL_API=3 "$ctl" --endpoints "http://$host_ip:$pd_port" \ + get '/topology/tidb/' --prefix -w json >"$topology" + if python3 - "$topology" <<'PY' +import base64, json, sys +with open(sys.argv[1], encoding="utf-8") as source: + entries = json.load(source).get("kvs", []) +for entry in entries: + key = base64.b64decode(entry["key"]) + if not key.endswith(b"/info"): + continue + info = json.loads(base64.b64decode(entry["value"])) + if info.get("labels", {}).get("keyspace") == "vip-ks": + sys.exit(0) +sys.exit(1) +PY + then + registered=1 + break + fi + sleep 1 +done +[[ $registered == 1 ]] || { echo 'TiDB keyspace label missing from PD topology' >&2; exit 1; } + write_config() { local node=$1 address=$2 mkdir -p "$run_dir/$node-work" @@ -188,7 +223,7 @@ start_node() { # would not be a reliable signal target for the Rust shutdown path. sudo ip netns exec "$namespace" sh -c \ 'echo "$$" >"$1"; shift; exec "$@"' \ - sh "$run_dir/$node.pid" env TIPROXY_ROUTE_DIAGNOSTIC=1 \ + sh "$run_dir/$node.pid" \ "$rust_binary" --standalone \ --config "$run_dir/$node.toml" --health-port 8080 \ >"$run_dir/diagnostics/$node.log" 2>&1 & From c7b983a858dadb97084ba388fcc9dd1a4583b4e5 Mon Sep 17 00:00:00 2001 From: CodexM5 Date: Sat, 26 Sep 2026 11:24:27 +0800 Subject: [PATCH 11/14] test: capture passive ARP evidence during VIP handoff --- .github/workflows/rust-vip-linux.yml | 2 +- tests/dataplane/integration/vip-standalone.sh | 18 +++++++++++++++++- 2 files changed, 18 insertions(+), 2 deletions(-) diff --git a/.github/workflows/rust-vip-linux.yml b/.github/workflows/rust-vip-linux.yml index fce6b331b..7460014bd 100644 --- a/.github/workflows/rust-vip-linux.yml +++ b/.github/workflows/rust-vip-linux.yml @@ -52,7 +52,7 @@ jobs: - name: Install Linux network and Rust build prerequisites run: | sudo apt-get update - sudo apt-get install --yes iproute2 arping protobuf-compiler + sudo apt-get install --yes iproute2 arping tcpdump protobuf-compiler command -v mysql - name: Build standalone Rust executable run: make rust-build diff --git a/tests/dataplane/integration/vip-standalone.sh b/tests/dataplane/integration/vip-standalone.sh index 1601dd551..a3b26beac 100644 --- a/tests/dataplane/integration/vip-standalone.sh +++ b/tests/dataplane/integration/vip-standalone.sh @@ -12,7 +12,7 @@ if [[ $(uname -s) != Linux ]]; then echo 'VIP acceptance requires Linux network namespaces' >&2 exit 2 fi -for command in ip arping tiup mysql curl; do +for command in ip arping tcpdump tiup mysql curl; do command -v "$command" >/dev/null || { echo "missing $command" >&2; exit 2; } done @@ -42,6 +42,7 @@ pd_port=$((2379 + offset)) tidb_port=$((4000 + offset)) tiup_pid= monitor_pid= +capture_pid= owned_rust_pid() { local node=$1 pid=$2 @@ -55,6 +56,7 @@ cleanup() { trap - EXIT set +e [[ -n $monitor_pid ]] && kill "$monitor_pid" 2>/dev/null + [[ -n $capture_pid ]] && kill "$capture_pid" 2>/dev/null [[ -s $run_dir/diagnostics/overlap.log ]] && status=1 for node in a b; do pidfile=$run_dir/$node.pid @@ -188,6 +190,17 @@ PY done [[ $registered == 1 ]] || { echo 'TiDB keyspace label missing from PD topology' >&2; exit 1; } +# Keep passive on-wire ARP evidence separate from SQL continuity. In +# particular, some arping versions report failure for a sent unsolicited +# packet because no reply is expected; an exit code alone cannot prove GARP. +arping -V >"$run_dir/diagnostics/arping-version.txt" 2>&1 || true +sudo timeout 40s tcpdump -l -nn -tt -i "$bridge" arp \ + >"$run_dir/diagnostics/arp-packets.txt" \ + 2>"$run_dir/diagnostics/tcpdump-stderr.txt" & +capture_pid=$! +sleep 1 +kill -0 "$capture_pid" + write_config() { local node=$1 address=$2 mkdir -p "$run_dir/$node-work" @@ -382,5 +395,8 @@ wait_mysql "$vip_ip" 6000 node-failover kill "$monitor_pid" 2>/dev/null || true wait "$monitor_pid" 2>/dev/null || true monitor_pid= +kill "$capture_pid" 2>/dev/null || true +wait "$capture_pid" 2>/dev/null || true +capture_pid= assert_no_overlap echo 'PASS: one VIP holder and SQL continuity through controlled close, restart and node/link loss' From 0264bf4b75165cf79ec0aad8c1318fc85e1eda68 Mon Sep 17 00:00:00 2001 From: CodexM5 Date: Sat, 26 Sep 2026 11:32:08 +0800 Subject: [PATCH 12/14] fix: send VIP GARP with both Linux arping CLI dialects --- rust/crates/tiproxy-rs/src/vip_network.rs | 57 +++++++++++++++-------- 1 file changed, 38 insertions(+), 19 deletions(-) diff --git a/rust/crates/tiproxy-rs/src/vip_network.rs b/rust/crates/tiproxy-rs/src/vip_network.rs index 57df39611..0e67b30ff 100644 --- a/rust/crates/tiproxy-rs/src/vip_network.rs +++ b/rust/crates/tiproxy-rs/src/vip_network.rs @@ -158,27 +158,46 @@ impl NetworkOperation for LinuxNetwork { }); } let ip = self.ip.to_string(); - let args = ["-c", "1", "-U", "-I", self.interface.as_ref(), ip.as_str()]; for _ in 0..self.burst_count { - let direct = run_command("arping", &args, "garp").await; - if matches!(direct, Ok(Output { success: true, .. })) { - continue; + let mut sent = false; + // iputils uses -I for interface; Thomas Habets' arping uses + // -i. Both implement -U as an unsolicited ARP announcement. + // Try the normal and noninteractive sudo paths for each. + for interface_flag in ["-I", "-i"] { + let args = [ + "-c", + "1", + "-U", + interface_flag, + self.interface.as_ref(), + ip.as_str(), + ]; + if run_command("arping", &args, "garp") + .await + .is_ok_and(|output| output.success) + { + sent = true; + break; + } + let sudo_args = [ + "-n", + "arping", + "-c", + "1", + "-U", + interface_flag, + self.interface.as_ref(), + ip.as_str(), + ]; + if run_command("sudo", &sudo_args, "garp") + .await + .is_ok_and(|output| output.success) + { + sent = true; + break; + } } - // Go tries both a library GARP and sudo arping for each packet. - // The CLI path is retained here; a direct arping success or - // noninteractive sudo success counts as one logical packet. - let sudo_args = [ - "-n", - "arping", - "-c", - "1", - "-U", - "-I", - self.interface.as_ref(), - ip.as_str(), - ]; - let sudo = run_command("sudo", &sudo_args, "garp").await?; - if !sudo.success { + if !sent { return Err(NetworkError { operation: "garp", class: "arping_failed", From 5ad76d6bdd48382012130f4a48c80d0b44af6c27 Mon Sep 17 00:00:00 2001 From: CodexM5 Date: Sat, 26 Sep 2026 11:39:04 +0800 Subject: [PATCH 13/14] fix: announce VIP source address with Habets arping --- rust/crates/tiproxy-rs/src/vip_network.rs | 32 +++++++------------ tests/dataplane/integration/vip-standalone.sh | 5 +++ 2 files changed, 16 insertions(+), 21 deletions(-) diff --git a/rust/crates/tiproxy-rs/src/vip_network.rs b/rust/crates/tiproxy-rs/src/vip_network.rs index 0e67b30ff..e4cb490ab 100644 --- a/rust/crates/tiproxy-rs/src/vip_network.rs +++ b/rust/crates/tiproxy-rs/src/vip_network.rs @@ -160,18 +160,16 @@ impl NetworkOperation for LinuxNetwork { let ip = self.ip.to_string(); for _ in 0..self.burst_count { let mut sent = false; - // iputils uses -I for interface; Thomas Habets' arping uses - // -i. Both implement -U as an unsolicited ARP announcement. - // Try the normal and noninteractive sudo paths for each. + // iputils uses -I for interface and chooses the VIP as the + // source of -U announcements. Thomas Habets' arping uses -i + // and needs -S VIP as well; otherwise it announces the node + // address rather than the VIP. Try direct and sudo for each. for interface_flag in ["-I", "-i"] { - let args = [ - "-c", - "1", - "-U", - interface_flag, - self.interface.as_ref(), - ip.as_str(), - ]; + let mut args = vec!["-c", "1", "-U", interface_flag, self.interface.as_ref()]; + if interface_flag == "-i" { + args.extend(["-S", ip.as_str()]); + } + args.push(ip.as_str()); if run_command("arping", &args, "garp") .await .is_ok_and(|output| output.success) @@ -179,16 +177,8 @@ impl NetworkOperation for LinuxNetwork { sent = true; break; } - let sudo_args = [ - "-n", - "arping", - "-c", - "1", - "-U", - interface_flag, - self.interface.as_ref(), - ip.as_str(), - ]; + let mut sudo_args = vec!["-n", "arping"]; + sudo_args.extend(args); if run_command("sudo", &sudo_args, "garp") .await .is_ok_and(|output| output.success) diff --git a/tests/dataplane/integration/vip-standalone.sh b/tests/dataplane/integration/vip-standalone.sh index a3b26beac..e8571d958 100644 --- a/tests/dataplane/integration/vip-standalone.sh +++ b/tests/dataplane/integration/vip-standalone.sh @@ -399,4 +399,9 @@ kill "$capture_pid" 2>/dev/null || true wait "$capture_pid" 2>/dev/null || true capture_pid= assert_no_overlap +if ! grep -F "tell $vip_ip" "$run_dir/diagnostics/arp-packets.txt" | + grep -Fq "who-has $vip_ip"; then + echo 'no gratuitous ARP announcement for the VIP observed on the bridge' >&2 + exit 1 +fi echo 'PASS: one VIP holder and SQL continuity through controlled close, restart and node/link loss' From da5904d5c6d6065cf100ee8b2d99a81898c8f0c7 Mon Sep 17 00:00:00 2001 From: CodexM5 Date: Sat, 26 Sep 2026 11:46:16 +0800 Subject: [PATCH 14/14] test: prove VIP-source GARP on each ownership transition --- tests/dataplane/integration/vip-standalone.sh | 40 ++++++++++++++++--- 1 file changed, 34 insertions(+), 6 deletions(-) diff --git a/tests/dataplane/integration/vip-standalone.sh b/tests/dataplane/integration/vip-standalone.sh index e8571d958..7e1763369 100644 --- a/tests/dataplane/integration/vip-standalone.sh +++ b/tests/dataplane/integration/vip-standalone.sh @@ -194,7 +194,7 @@ done # particular, some arping versions report failure for a sent unsolicited # packet because no reply is expected; an exit code alone cannot prove GARP. arping -V >"$run_dir/diagnostics/arping-version.txt" 2>&1 || true -sudo timeout 40s tcpdump -l -nn -tt -i "$bridge" arp \ +sudo timeout 40s tcpdump -e -l -nn -tt -i "$bridge" arp \ >"$run_dir/diagnostics/arp-packets.txt" \ 2>"$run_dir/diagnostics/tcpdump-stderr.txt" & capture_pid=$! @@ -269,6 +269,34 @@ assert_no_overlap() { fi } +garp_count() { + local node=$1 mac + if [[ $node == a ]]; then + mac=02:00:00:99:20:11 + else + mac=02:00:00:99:20:12 + fi + awk -v vip="$vip_ip" -v mac="$mac" ' + index($0, mac) && index($0, "who-has " vip) && index($0, "tell " vip) { count++ } + END { print count + 0 } + ' "$run_dir/diagnostics/arp-packets.txt" +} + +wait_garp() { + local label=$1 node=$2 previous=$3 count + for _ in {1..30}; do + count=$(garp_count "$node") + if ((count > previous)); then + echo "$(date +%s.%N) $label $node GARP observed" \ + >>"$run_dir/diagnostics/events.log" + return 0 + fi + sleep 0.2 + done + echo "$label: no VIP-source GARP from $node observed on the bridge" >&2 + return 1 +} + monitor_overlap & monitor_pid=$! start_node a "$ns_a" @@ -346,6 +374,7 @@ wait_single_owner() { owner=$(wait_single_owner initial) assert_no_overlap +wait_garp initial "$owner" 0 wait_mysql "$vip_ip" 6000 initial-vip if [[ $owner == a ]]; then former=a @@ -355,9 +384,11 @@ else new=a fi former_pid=$(cat "$run_dir/$former.pid") +before_garp=$(garp_count "$new") sudo kill -INT "$former_pid" wait_single_owner after-controlled-close "$new" >/dev/null assert_no_overlap +wait_garp after-controlled-close "$new" "$before_garp" wait_mysql "$vip_ip" 6000 controlled-failover for _ in {1..30}; do sudo kill -0 "$former_pid" 2>/dev/null || break @@ -383,6 +414,7 @@ wait_mysql "$vip_ip" 6000 restart # Model a whole-node crash: terminate its process and remove its network link. # Process-only SIGKILL leaves an IP on a live Linux interface and is separately # documented as a remaining Go/Rust operational fencing gap. +before_garp=$(garp_count "$former") sudo kill -KILL "$(cat "$run_dir/$new.pid")" if [[ $new == a ]]; then sudo ip link del "$host_a" @@ -391,6 +423,7 @@ else fi wait_single_owner after-node-link-loss "$former" >/dev/null assert_no_overlap +wait_garp after-node-link-loss "$former" "$before_garp" wait_mysql "$vip_ip" 6000 node-failover kill "$monitor_pid" 2>/dev/null || true wait "$monitor_pid" 2>/dev/null || true @@ -399,9 +432,4 @@ kill "$capture_pid" 2>/dev/null || true wait "$capture_pid" 2>/dev/null || true capture_pid= assert_no_overlap -if ! grep -F "tell $vip_ip" "$run_dir/diagnostics/arp-packets.txt" | - grep -Fq "who-has $vip_ip"; then - echo 'no gratuitous ARP announcement for the VIP observed on the bridge' >&2 - exit 1 -fi echo 'PASS: one VIP holder and SQL continuity through controlled close, restart and node/link loss'