From 38d6604cde025d5bdf1fef6c92244ac234288d45 Mon Sep 17 00:00:00 2001
From: Fupan Li <fupan.lfp@antgroup.com>
Date: Tue, 24 Jun 2025 09:37:13 +0800
Subject: [PATCH] sync: fix the connection fd leak on the server side

We need to ensure that only when the connection is successfully
inserted into the connections map can the handler send fd
to notify the reaper to recycle the connection.

Otherwise, the handler may have been executed and sent fd to notify
the reaper to recycle the connection, and the connection has
not been inserted into the map. Then the reaper cannot find the corresponding
connection after receiving the fd, and loses the opportunity to recycle the
connection, causing fd leakage;

Signed-off-by: Fupan Li <fupan.lfp@antgroup.com>
---
 src/sync/server.rs | 43 +++++++++++++++++++++++++++++--------------
 1 file changed, 29 insertions(+), 14 deletions(-)

diff --git a/src/sync/server.rs b/src/sync/server.rs
index d19b0558..555224de 100644
--- a/src/sync/server.rs
+++ b/src/sync/server.rs
@@ -401,7 +401,9 @@ impl Server {
                     let reaper_tx_child = reaper_tx.clone();
                     let pipe_connection_child = pipe_connection.clone();
 
-                    let handler = thread::Builder::new()
+                    let (sync_tx, sync_rx) = channel();
+
+                    let _ = thread::Builder::new()
                         .name("client_handler".into())
                         .spawn(move || {
                             debug!("Got new client");
@@ -505,25 +507,38 @@ impl Server {
                             drop(workload_rx);
                             handler.join().unwrap_or(());
                             reader.join().unwrap_or(());
+
+                            //wait untile this connection had been inserted connections map;
+                            sync_rx.recv().unwrap_or(());
+
                             // client_handler should not close fd before exit
                             // , which prevent fd reuse issue.
                             reaper_tx_child.send(pipe.id()).unwrap();
 
                             debug!("client thread quit");
                         })
-                        .unwrap();
-
-                    let mut cns = connections.lock().unwrap();
-
-                    let id = pipe_connection.id();
-                    cns.insert(
-                        id,
-                        Connection {
-                            connection: pipe_connection,
-                            handler: Some(handler),
-                            quit: quit.clone(),
-                        },
-                    );
+                        .map(|handler| {
+                            let mut cns = connections.lock().unwrap();
+                            let id = pipe_connection.id();
+                            cns.insert(
+                                id,
+                                Connection {
+                                    connection: pipe_connection,
+                                    handler: Some(handler),
+                                    quit: quit.clone(),
+                                },
+                            );
+
+                            // We need to ensure that only when the connection is successfully inserted into the
+                            // connections map can the handler send fd to notify the reaper to recycle the connection.
+                            // Otherwise, the handler may have been executed and sent fd to notify the reaper to recycle
+                            // the connection, and the connection has not been inserted into the map. Then the reaper cannot
+                            // find the corresponding connection after receiving the fd, and loses the opportunity to recycle
+                            // the connection, causing fd leakage;
+                            // Notify the handler continue, to send fd to reaper thread.
+                            sync_tx.send(()).unwrap_or(());
+                        })
+                        .map_err(|e| warn!("spawn handler thread failed with: {:?}", e));
                 } // end loop
 
                 // notify reaper thread to exit.