Merge pull request #864 from telemt/flow

Restore ME writer source IP for initial proxy request binding

Cargo.lock

@@ -2899,7 +2899,7 @@ checksum = "7b2093cf4c8eb1e67749a6762251bc9cd836b6fc171623bd0a9d324d37af2417"
 
 [[package]]
 name = "telemt"
-version = "3.4.19"
+version = "3.4.21"
 dependencies = [
  "aes",
  "anyhow",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "telemt"
-version = "3.4.19"
+version = "3.4.21"
 edition = "2024"
 
 [features]

src/transport/middle_proxy/send.rs

@@ -464,7 +464,9 @@ impl MePool {
                 if !self.writer_accepts_new_binding(w) {
                     continue;
                 }
-                let (payload, meta) = build_routed_payload(our_addr);
+                // Keep the advertised proxy IP aligned with the selected ME writer source.
+                let effective_our_addr = SocketAddr::new(w.source_ip, our_addr.port());
+                let (payload, meta) = build_routed_payload(effective_our_addr);
                 match w.tx.clone().try_reserve_owned() {
                     Ok(permit) => {
                         if !self.registry.bind_writer(conn_id, w.id, meta).await {
@@ -519,7 +521,9 @@ impl MePool {
             }
             self.stats
                 .increment_me_writer_pick_blocking_fallback_total();
-            let (payload, meta) = build_routed_payload(our_addr);
+            // Keep the advertised proxy IP aligned with the selected ME writer source.
+            let effective_our_addr = SocketAddr::new(w.source_ip, our_addr.port());
+            let (payload, meta) = build_routed_payload(effective_our_addr);
             let reserve_result =
                 if let Some(timeout) = self.route_runtime.me_route_blocking_send_timeout {
                     match tokio::time::timeout(timeout, w.tx.clone().reserve_owned()).await {

src/transport/middle_proxy/tests/send_adversarial_tests.rs

@@ -323,7 +323,7 @@ async fn send_proxy_req_prunes_iterative_stale_bind_failures_without_data_replay
 }
 
 #[tokio::test]
-async fn send_proxy_req_preserves_client_facing_our_addr_when_writer_source_ip_differs() {
+async fn send_proxy_req_uses_writer_source_ip_when_advertised_our_addr_differs() {
     let (pool, _rng) = make_pool().await;
     pool.rr.store(0, Ordering::Relaxed);
 
@@ -363,5 +363,8 @@ async fn send_proxy_req_preserves_client_facing_our_addr_when_writer_source_ip_d
     let payload = recv_first_data_payload(&mut live_rx, Duration::from_millis(50))
         .await
         .expect("writer must receive routed payload");
-    assert_eq!(proxy_req_our_addr_from_payload(&payload), our_addr);
+    assert_eq!(
+        proxy_req_our_addr_from_payload(&payload),
+        SocketAddr::new(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 31)), our_addr.port())
+    );
 }