tcp punch: keep the simultaneous-open stream

Punch after PunchHoleSent instead of before, and hold the socket for 4s rather
than 30ms, so the window covers our own SYN retransmits. The first SYN reaches
the peer before it has been told to connect and is dropped; a retransmit finds
it in SYN_SENT and draws a SYN-ACK, which a gateway that drops a bare inbound
SYN still passes as the reply to our own outbound SYN. That is the one path a
listener cannot serve, since it answers a SYN-ACK with RST.

Use the stream when it connects instead of dropping it, which would RST the
connection the listener was going to accept.

4s, not CONNECT_TIMEOUT: it covers the first two retransmits on a 1s initial
RTO and still covers the first on a 3s one, while leaving the peer's own
retransmits at 7s and 15s to the listener. Anything longer only buys
resilience to packet loss, and binds the listener after the peer has given up.

Release the rendezvous connection before the window rather than after, so the
server keeps holding it for a round trip and not for the whole punch.

Based on #15904.

Co-authored-by: alonginwind <100897495+alonginwind@users.noreply.github.com>
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QeWzFF37in7Q3iLVn3kzfS

src/rendezvous_mediator.rs

@@ -700,19 +700,33 @@ impl RendezvousMediator {
             return Ok(());
         }
         log::debug!("Punch tcp hole to {:?}", peer_addr);
-        let mut socket = {
-            let socket = connect_tcp(&*self.host, CONNECT_TIMEOUT).await?;
-            let local_addr = socket.local_addr();
-            // key important here for punch hole to tell my gateway incoming peer is safe.
-            // it can not be async here, because local_addr can not be reused, we must close the connection before use it again.
-            allow_err!(socket_client::connect_tcp_local(peer_addr, Some(local_addr), 30).await);
-            socket
-        };
+        let mut socket = connect_tcp(&*self.host, CONNECT_TIMEOUT).await?;
+        let local_addr = socket.local_addr();
         let mut msg_out = Message::new();
         msg_out.set_punch_hole_sent(msg_punch);
         let bytes = msg_out.write_to_bytes()?;
         socket.send_raw(bytes).await?;
-        crate::accept_connection(server.clone(), socket, peer_addr, true, meta).await;
+        drop(socket);
+        // key important here for punch hole to tell my gateway incoming peer is safe.
+        // Punched after the message so the window covers our own SYN retransmits: the first SYN
+        // reaches the peer before it has been told to connect and is dropped, while a retransmit
+        // finds it in SYN_SENT and draws a SYN-ACK, which a gateway that drops a bare inbound SYN
+        // still passes as the reply to our own outbound SYN. Keep that stream rather than dropping
+        // it, or the RST kills the connection the listener would have accepted. The listener can
+        // not be bound meanwhile, because local_addr can not be reused while this socket holds it.
+        // 4s, not CONNECT_TIMEOUT: it covers our first two SYN retransmits on a 1s initial RTO and
+        // still covers the first one if the RTO is 3s, while leaving the peer's own retransmits at
+        // 7s and 15s to the listener in case a gateway drops that inbound SYN less consistently
+        // than it looks. The retransmits past those only buy resilience to packet loss.
+        match socket_client::connect_tcp_local(peer_addr, Some(local_addr), 4_000).await {
+            Ok(stream) => {
+                crate::server::create_tcp_connection(server, stream, peer_addr, true, meta).await?;
+            }
+            Err(err) => {
+                log::debug!("Punch tcp hole to {} failed: {}", peer_addr, err);
+                crate::accept_connection_at(server, local_addr, peer_addr, true, meta).await;
+            }
+        }
         Ok(())
     }
 

src/server.rs

@@ -169,12 +169,10 @@ pub fn new() -> ServerPtr {
 
 async fn accept_connection_(
     server: ServerPtr,
-    socket: Stream,
+    local_addr: SocketAddr,
     secure: bool,
     meta: ConnectionMeta,
 ) -> ResultType<()> {
-    let local_addr = socket.local_addr();
-    drop(socket);
     // even we drop socket, below still may fail if not use reuse_addr,
     // there is TIME_WAIT before socket really released, so sometimes we
     // see "Only one usage of each socket address is normally permitted" on windows sometimes,
@@ -279,7 +277,21 @@ pub async fn accept_connection(
     secure: bool,
     meta: ConnectionMeta,
 ) {
-    if let Err(err) = accept_connection_(server, socket, secure, meta).await {
+    let local_addr = socket.local_addr();
+    drop(socket);
+    accept_connection_at(server, local_addr, peer_addr, secure, meta).await
+}
+
+// Listen on a local_addr whose socket the caller already released, so the tcp punch path can
+// hand its rendezvous connection back to the server before the punch window instead of after it.
+pub async fn accept_connection_at(
+    server: ServerPtr,
+    local_addr: SocketAddr,
+    peer_addr: SocketAddr,
+    secure: bool,
+    meta: ConnectionMeta,
+) {
+    if let Err(err) = accept_connection_(server, local_addr, secure, meta).await {
         log::warn!("Failed to accept connection from {}: {}", peer_addr, err);
     }
 }