fix(dataplane): address five critical issues in Rust dataplane

- #1051: IPv4-mapped IPv6 addresses now checked against v4 denylist in
  SSRF filter; ULA check extended to cover full fc00::/7 range
- #1052: Router is now rebuilt on incremental upsert_route/delete_route
  via a shared rebuild_router helper, not just on full apply_config
- #1053: LB cache invalidation in upsert_cluster/delete_cluster changed
  from try_write() (silently skipped on contention) to async write().await
- #1054: PoolGuard now stores Arc<Semaphore> and implements Drop to
  return the permit automatically, preventing leaks on panic
- #1055: Backend selection replaced SystemTime nanos with per-Route
  AtomicU64 counter for deterministic weighted round-robin

Author: piwi3910

dataplane/novaedge-dataplane/src/config.rs

@@ -272,23 +272,18 @@ impl RuntimeConfig {
     ///
     /// Also invalidates the cached LB instance for this cluster so that
     /// endpoint or algorithm changes take effect on the next request.
-    pub fn upsert_cluster(&self, cluster: ClusterState) {
+    pub async fn upsert_cluster(&self, cluster: ClusterState) {
         let name = cluster.name.clone();
         self.clusters.insert(name.clone(), cluster);
-        // Invalidate the LB cache entry for this cluster (best-effort; uses try_write
-        // to avoid blocking the caller if the lock is held by an async task).
-        if let Ok(mut cache) = self.lb_cache.try_write() {
-            cache.remove(&name);
-        }
+        // Invalidate the LB cache entry for this cluster.
+        self.lb_cache.write().await.remove(&name);
         self.bump();
     }
 
     /// Remove a cluster by name.
-    pub fn delete_cluster(&self, name: &str) {
+    pub async fn delete_cluster(&self, name: &str) {
         self.clusters.remove(name);
-        if let Ok(mut cache) = self.lb_cache.try_write() {
-            cache.remove(name);
-        }
+        self.lb_cache.write().await.remove(name);
         self.bump();
     }
 
@@ -659,34 +654,34 @@ mod tests {
         assert_eq!(snap.routes["route-1"].backend_refs[0].0, "backend-1");
     }
 
-    #[test]
-    fn test_upsert_and_get_cluster() {
+    #[tokio::test]
+    async fn test_upsert_and_get_cluster() {
         let cfg = RuntimeConfig::new();
-        cfg.upsert_cluster(test_cluster("cluster-1"));
+        cfg.upsert_cluster(test_cluster("cluster-1")).await;
         let snap = cfg.snapshot();
         assert_eq!(snap.clusters.len(), 1);
         assert_eq!(snap.clusters["cluster-1"].endpoints.len(), 1);
     }
 
-    #[test]
-    fn test_get_cluster() {
+    #[tokio::test]
+    async fn test_get_cluster() {
         let cfg = RuntimeConfig::new();
         assert!(cfg.get_cluster("missing").is_none());
-        cfg.upsert_cluster(test_cluster("cluster-1"));
+        cfg.upsert_cluster(test_cluster("cluster-1")).await;
         let c = cfg.get_cluster("cluster-1").unwrap();
         assert_eq!(c.endpoints.len(), 1);
     }
 
-    #[test]
-    fn test_delete_removes_entry() {
+    #[tokio::test]
+    async fn test_delete_removes_entry() {
         let cfg = RuntimeConfig::new();
         cfg.upsert_gateway(test_gateway("gw-1", 8080));
         cfg.upsert_route(test_route("route-1", "b"));
-        cfg.upsert_cluster(test_cluster("cluster-1"));
+        cfg.upsert_cluster(test_cluster("cluster-1")).await;
 
         cfg.delete_gateway("gw-1");
         cfg.delete_route("route-1");
-        cfg.delete_cluster("cluster-1");
+        cfg.delete_cluster("cluster-1").await;
 
         let snap = cfg.snapshot();
         assert!(snap.gateways.is_empty());

dataplane/novaedge-dataplane/src/listener.rs

@@ -615,54 +615,56 @@ mod tests {
             hedge_delay_ms: 0,
             hedge_max_requests: 0,
         });
-        config.upsert_cluster(ClusterState {
-            name: "tcp-cluster".into(),
-            endpoints: vec![EndpointState {
-                address: backend_addr.ip().to_string(),
-                port: backend_addr.port() as u32,
-                weight: 1,
-                healthy: true,
-                zone: None,
-                priority: 0,
-                draining: false,
-                drain_start: None,
-            }],
-            lb_algorithm: "round-robin".into(),
-            health_check_path: String::new(),
-            session_affinity_type: String::new(),
-            session_affinity_cookie: String::new(),
-            session_affinity_header: String::new(),
-            outlier_consecutive_5xx: 0,
-            outlier_ejection_duration_ms: 0,
-            outlier_max_ejection_pct: 0,
-            outlier_interval_ms: 0,
-            outlier_sr_min_hosts: 0,
-            outlier_sr_min_requests: 0,
-            outlier_sr_stdev_factor: 0.0,
-            slow_start_window_ms: 0,
-            slow_start_aggression: 1.0,
-            protocol: String::new(),
-            upstream_proxy_protocol_enabled: false,
-            upstream_proxy_protocol_version: 0,
-            tls_enabled: false,
-            tls_insecure_skip_verify: false,
-            tls_ca_pem: Vec::new(),
-            tls_server_name: String::new(),
-            cb_failure_threshold: 0,
-            cb_success_threshold: 0,
-            cb_open_duration_ms: 0,
-            cb_half_open_max_requests: 0,
-            pool_max_connections: 0,
-            pool_max_idle: 0,
-            pool_idle_timeout_ms: 0,
-            pool_connect_timeout_ms: 0,
-            panic_threshold_percent: 0,
-            max_requests_per_connection: 0,
-            request_queue_depth: 0,
-            request_queue_timeout_ms: 0,
-            subset_size: 0,
-            remote_endpoints: Vec::new(),
-        });
+        config
+            .upsert_cluster(ClusterState {
+                name: "tcp-cluster".into(),
+                endpoints: vec![EndpointState {
+                    address: backend_addr.ip().to_string(),
+                    port: backend_addr.port() as u32,
+                    weight: 1,
+                    healthy: true,
+                    zone: None,
+                    priority: 0,
+                    draining: false,
+                    drain_start: None,
+                }],
+                lb_algorithm: "round-robin".into(),
+                health_check_path: String::new(),
+                session_affinity_type: String::new(),
+                session_affinity_cookie: String::new(),
+                session_affinity_header: String::new(),
+                outlier_consecutive_5xx: 0,
+                outlier_ejection_duration_ms: 0,
+                outlier_max_ejection_pct: 0,
+                outlier_interval_ms: 0,
+                outlier_sr_min_hosts: 0,
+                outlier_sr_min_requests: 0,
+                outlier_sr_stdev_factor: 0.0,
+                slow_start_window_ms: 0,
+                slow_start_aggression: 1.0,
+                protocol: String::new(),
+                upstream_proxy_protocol_enabled: false,
+                upstream_proxy_protocol_version: 0,
+                tls_enabled: false,
+                tls_insecure_skip_verify: false,
+                tls_ca_pem: Vec::new(),
+                tls_server_name: String::new(),
+                cb_failure_threshold: 0,
+                cb_success_threshold: 0,
+                cb_open_duration_ms: 0,
+                cb_half_open_max_requests: 0,
+                pool_max_connections: 0,
+                pool_max_idle: 0,
+                pool_idle_timeout_ms: 0,
+                pool_connect_timeout_ms: 0,
+                panic_threshold_percent: 0,
+                max_requests_per_connection: 0,
+                request_queue_depth: 0,
+                request_queue_timeout_ms: 0,
+                subset_size: 0,
+                remote_endpoints: Vec::new(),
+            })
+            .await;
 
         let handler = make_handler(config.clone());
         let mgr = ListenerManager::new(config.clone(), handler);

dataplane/novaedge-dataplane/src/middleware/auth/forward.rs

@@ -155,8 +155,12 @@ fn is_denied_ip(ip: &std::net::IpAddr) -> bool {
                 || (v4.octets()[0] == 169 && v4.octets()[1] == 254) // 169.254.0.0/16 (link-local / cloud metadata)
         }
         std::net::IpAddr::V6(v6) => {
+            // Detect IPv4-mapped IPv6 (::ffff:x.x.x.x) and check the inner v4 address.
+            if let Some(v4) = v6.to_ipv4_mapped() {
+                return is_denied_ip(&std::net::IpAddr::V4(v4));
+            }
             v6.is_loopback()                              // ::1
-                || v6.octets()[0] == 0xfd                 // fd00::/8 (ULA)
+                || (v6.octets()[0] & 0xfe) == 0xfc        // fc00::/7 (ULA)
                 || v6.octets()[0] == 0xfe && (v6.octets()[1] & 0xc0) == 0x80 // fe80::/10 (link-local)
         }
     }
@@ -264,6 +268,17 @@ mod tests {
         assert!(is_denied_ip(&IpAddr::V6(Ipv6Addr::new(
             0xfd00, 0, 0, 0, 0, 0, 0, 1
         ))));
+        // fc00::/8 (ULA — the other half of fc00::/7)
+        assert!(is_denied_ip(&IpAddr::V6(Ipv6Addr::new(
+            0xfc00, 0, 0, 0, 0, 0, 0, 1
+        ))));
+        // IPv4-mapped IPv6 addresses must be checked against the v4 denylist
+        assert!(is_denied_ip(&IpAddr::V6(Ipv6Addr::new(
+            0, 0, 0, 0, 0, 0xffff, 0x7f00, 0x0001 // ::ffff:127.0.0.1
+        ))));
+        assert!(is_denied_ip(&IpAddr::V6(Ipv6Addr::new(
+            0, 0, 0, 0, 0, 0xffff, 0xa9fe, 0xa9fe // ::ffff:169.254.169.254
+        ))));
 
         // Allowed ranges
         assert!(!is_denied_ip(&IpAddr::V4(Ipv4Addr::new(8, 8, 8, 8))));

dataplane/novaedge-dataplane/src/proxy/handler.rs

@@ -351,6 +351,7 @@ impl ProxyHandler {
                         rewrite_path: None,
                         add_headers: HashMap::new(),
                         middleware_refs: Vec::new(),
+                        rr_counter: std::sync::Arc::new(std::sync::atomic::AtomicU64::new(0)),
                     }
                 } else {
                     debug!("No route matched for {method} {host}{path}");
@@ -682,7 +683,7 @@ impl ProxyHandler {
             let backend_addr = SocketAddr::new(selected.addr, selected.port);
 
             // Pool acquire.
-            let _pool_guard = match self.connection_pool.acquire(backend_addr).await {
+            let mut pool_guard = match self.connection_pool.acquire(backend_addr).await {
                 Ok(guard) => guard,
                 Err(e) => {
                     warn!(
@@ -841,6 +842,7 @@ impl ProxyHandler {
                             cb.record_failure();
                             outlier_detector.record_failure(backend_addr);
                             balancer.report(backend_idx, upstream_start.elapsed(), false);
+                            pool_guard.mark_released();
                             self.connection_pool.release(backend_addr).await;
                             if attempt < max_attempts - 1
                                 && is_retryable_condition("reset", &retry_on)
@@ -857,6 +859,7 @@ impl ProxyHandler {
                         }
                     };
 
+                    pool_guard.mark_released();
                     self.connection_pool.release(backend_addr).await;
 
                     final_status = status;
@@ -881,6 +884,7 @@ impl ProxyHandler {
                     cb.record_failure();
                     outlier_detector.record_failure(backend_addr);
                     balancer.report(backend_idx, upstream_latency, false);
+                    pool_guard.mark_released();
                     self.connection_pool.release(backend_addr).await;
                     self.connection_pool
                         .record_connection_failure(backend_addr)
@@ -1116,6 +1120,7 @@ impl ProxyHandler {
                         rewrite_path: None,
                         add_headers: HashMap::new(),
                         middleware_refs: Vec::new(),
+                        rr_counter: std::sync::Arc::new(std::sync::atomic::AtomicU64::new(0)),
                     }
                 } else {
                     return Err(hyper::Response::builder()
@@ -1544,6 +1549,7 @@ mod tests {
             rewrite_path: None,
             add_headers: HashMap::new(),
             middleware_refs: Vec::new(),
+            rr_counter: std::sync::Arc::new(std::sync::atomic::AtomicU64::new(0)),
         }
     }
 
@@ -1623,7 +1629,8 @@ mod tests {
             request_queue_timeout_ms: 0,
             subset_size: 0,
             remote_endpoints: Vec::new(),
-        });
+        })
+        .await;
 
         let _handler = test_handler(router, cfg.clone());
         let cluster = cfg.get_cluster("test-cluster").unwrap();
@@ -1715,7 +1722,8 @@ mod tests {
             request_queue_timeout_ms: 0,
             subset_size: 0,
             remote_endpoints: Vec::new(),
-        });
+        })
+        .await;
 
         let handler = Arc::new(test_handler(router, cfg));
 

dataplane/novaedge-dataplane/src/proxy/router.rs

@@ -1,7 +1,9 @@
 use std::collections::HashMap;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::Arc;
 
 /// A route configuration for the HTTP proxy.
-#[derive(Debug, Clone)]
+#[derive(Clone)]
 pub struct Route {
     pub name: String,
     pub hostnames: Vec<HostMatch>,
@@ -15,13 +17,35 @@ pub struct Route {
     pub rewrite_path: Option<String>,
     pub add_headers: HashMap<String, String>,
     pub middleware_refs: Vec<String>,
+    /// Per-route counter for deterministic weighted round-robin backend selection.
+    pub rr_counter: Arc<AtomicU64>,
+}
+
+impl std::fmt::Debug for Route {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("Route")
+            .field("name", &self.name)
+            .field("hostnames", &self.hostnames)
+            .field("paths", &self.paths)
+            .field("methods", &self.methods)
+            .field("headers", &self.headers)
+            .field("query_params", &self.query_params)
+            .field("backend_refs", &self.backend_refs)
+            .field("priority", &self.priority)
+            .field("rewrite_path", &self.rewrite_path)
+            .field("add_headers", &self.add_headers)
+            .field("middleware_refs", &self.middleware_refs)
+            .field("rr_counter", &self.rr_counter.load(Ordering::Relaxed))
+            .finish()
+    }
 }
 
 impl Route {
-    /// Select a backend cluster name using weighted random selection.
+    /// Select a backend cluster name using deterministic weighted round-robin.
     ///
-    /// If there is only one backend, returns it directly. If multiple, picks
-    /// one randomly proportional to weights.
+    /// If there is only one backend, returns it directly. If multiple, uses
+    /// a per-route atomic counter to cycle through backends proportional to
+    /// their weights. This is deterministic and thread-safe.
     pub fn select_backend(&self) -> Option<&str> {
         match self.backend_refs.len() {
             0 => None,
@@ -31,21 +55,13 @@ impl Route {
                 if total_weight == 0 {
                     return Some(&self.backend_refs[0].0);
                 }
-                let mut rng_val = {
-                    // Use a fast inline xorshift for weight selection to avoid
-                    // pulling in rand as a dependency just for this.
-                    use std::time::SystemTime;
-                    let seed = SystemTime::now()
-                        .duration_since(SystemTime::UNIX_EPOCH)
-                        .unwrap_or_default()
-                        .subsec_nanos();
-                    seed % total_weight
-                };
+                let slot = self.rr_counter.fetch_add(1, Ordering::Relaxed) % total_weight as u64;
+                let mut remaining = slot as u32;
                 for (name, weight) in &self.backend_refs {
-                    if rng_val < *weight {
+                    if remaining < *weight {
                         return Some(name);
                     }
-                    rng_val -= weight;
+                    remaining -= weight;
                 }
                 // Fallback (shouldn't reach here).
                 Some(&self.backend_refs[0].0)
@@ -303,6 +319,7 @@ mod tests {
             rewrite_path: None,
             add_headers: HashMap::new(),
             middleware_refs: Vec::new(),
+            rr_counter: Arc::new(AtomicU64::new(0)),
         }
     }