Restore ME writer source IP for initial proxy request binding

This commit is contained in:
Alexey
2026-06-29 12:37:31 +03:00
parent 7aee991416
commit b9c5c71dbc
2 changed files with 11 additions and 4 deletions
+6 -2
View File
@@ -464,7 +464,9 @@ impl MePool {
if !self.writer_accepts_new_binding(w) { if !self.writer_accepts_new_binding(w) {
continue; 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() { match w.tx.clone().try_reserve_owned() {
Ok(permit) => { Ok(permit) => {
if !self.registry.bind_writer(conn_id, w.id, meta).await { if !self.registry.bind_writer(conn_id, w.id, meta).await {
@@ -519,7 +521,9 @@ impl MePool {
} }
self.stats self.stats
.increment_me_writer_pick_blocking_fallback_total(); .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 = let reserve_result =
if let Some(timeout) = self.route_runtime.me_route_blocking_send_timeout { if let Some(timeout) = self.route_runtime.me_route_blocking_send_timeout {
match tokio::time::timeout(timeout, w.tx.clone().reserve_owned()).await { match tokio::time::timeout(timeout, w.tx.clone().reserve_owned()).await {
@@ -323,7 +323,7 @@ async fn send_proxy_req_prunes_iterative_stale_bind_failures_without_data_replay
} }
#[tokio::test] #[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; let (pool, _rng) = make_pool().await;
pool.rr.store(0, Ordering::Relaxed); 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)) let payload = recv_first_data_payload(&mut live_rx, Duration::from_millis(50))
.await .await
.expect("writer must receive routed payload"); .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())
);
} }