Fix issue with socket folders being held by non existing process and bump dependencies

Signed-off-by: avifenesh <[email protected]>

Author: avifenesh

Diff for: glide-core/Cargo.toml

@@ -9,7 +9,7 @@ authors = ["Valkey GLIDE Maintainers"]
 
 [dependencies]
 bytes = "1"
-futures = "^0.3"
+futures = "0.3"
 redis = { path = "./redis-rs/redis", features = [
     "aio",
     "tokio-comp",
@@ -21,27 +21,27 @@ redis = { path = "./redis-rs/redis", features = [
 telemetrylib = { path = "./telemetry" }
 tokio = { version = "1", features = ["macros", "time"] }
 logger_core = { path = "../logger_core" }
-dispose = "0.5.0"
-tokio-util = { version = "^0.7", features = ["rt"], optional = true }
-num_cpus = { version = "^1.15", optional = true }
-tokio-retry2 = {version = "0.5", features = ["jitter"]}
+dispose = "0.5"
+tokio-util = { version = "0.7", features = ["rt"], optional = true }
+num_cpus = { version = "1", optional = true }
+tokio-retry2 = { version = "0.5", features = ["jitter"] }
 
 protobuf = { version = "3", features = [
     "bytes",
     "with-bytes",
 ], optional = true }
-integer-encoding = { version = "4.0.0", optional = true }
-thiserror = "1"
-rand = { version = "0.8.5" }
-futures-intrusive = "0.5.0"
-directories = { version = "5.0", optional = true }
-once_cell = "1.18.0"
-sha1_smol = "1.0.0"
-nanoid = "0.4.0"
-async-trait = { version = "0.1.24" }
+integer-encoding = { version = "4", optional = true }
+thiserror = "2"
+rand = "0.9"
+futures-intrusive = "0.5"
+directories = { version = "6", optional = true }
+once_cell = "1"
+sha1_smol = "1"
+nanoid = "0.4"
+async-trait = { version = "0.1" }
 serde_json = "1"
 serde = { version = "1", features = ["derive"] }
-versions = "6.3"
+versions = "6"
 
 [features]
 socket-layer = [
@@ -57,11 +57,11 @@ standalone_heartbeat = []
 rsevents = "0.3.1"
 socket2 = "^0.5"
 tempfile = "3.3.0"
-rstest = "^0.23"
+rstest = "^0.24"
 serial_test = "3"
 criterion = { version = "^0.5", features = ["html_reports", "async_tokio"] }
-which = "6"
-ctor = "0.2.2"
+which = "7"
+ctor = "0.3"
 redis = { path = "./redis-rs/redis", features = ["tls-rustls-insecure"] }
 iai-callgrind = "0.14"
 tokio = { version = "1", features = ["rt-multi-thread"] }

Diff for: glide-core/benches/rotating_buffer_benchmark.rs

@@ -11,7 +11,7 @@ use glide_core::{
 };
 use integer_encoding::VarInt;
 use protobuf::Message;
-use rand::{distributions::Alphanumeric, Rng};
+use rand::{distr::Alphanumeric, Rng};
 
 fn benchmark(
     c: &mut Criterion,
@@ -102,7 +102,7 @@ fn benchmark_split_data(
 }
 
 fn generate_random_string(length: usize) -> bytes::Bytes {
-    let s: String = rand::thread_rng()
+    let s: String = rand::rng()
         .sample_iter(&Alphanumeric)
         .take(length)
         .map(char::from)

Diff for: glide-core/src/client/standalone_client.rs

@@ -9,7 +9,6 @@ use crate::retry_strategies::RetryStrategy;
 use futures::{future, stream, StreamExt};
 use logger_core::log_debug;
 use logger_core::log_warn;
-use rand::Rng;
 use redis::aio::ConnectionLike;
 use redis::cluster_routing::{self, is_readonly_cmd, ResponsePolicy, Routable, RoutingInfo};
 use redis::{PushInfo, RedisError, RedisResult, Value};
@@ -125,7 +124,7 @@ impl StandaloneClient {
         let tls_mode = connection_request.tls_mode;
         let node_count = connection_request.addresses.len();
         // randomize pubsub nodes, maybe a batter option is to always use the primary
-        let pubsub_node_index = rand::thread_rng().gen_range(0..node_count);
+        let pubsub_node_index = rand::random_range(0..node_count);
         let pubsub_addr = &connection_request.addresses[pubsub_node_index];
         let discover_az = matches!(
             connection_request.read_from,

Diff for: glide-core/src/rotating_buffer.rs

@@ -29,18 +29,17 @@ impl RotatingBuffer {
                 let start_pos = prev_position + bytes_read;
                 if (start_pos + request_len as usize) > buffer_len {
                     break;
-                } else {
-                    match T::parse_from_tokio_bytes(
-                        &buffer.slice(start_pos..start_pos + request_len as usize),
-                    ) {
-                        Ok(request) => {
-                            prev_position += request_len as usize + bytes_read;
-                            results.push(request);
-                        }
-                        Err(err) => {
-                            log_error("parse input", format!("Failed to parse request: {err}"));
-                            return Err(err.into());
-                        }
+                }
+                match T::parse_from_tokio_bytes(
+                    &buffer.slice(start_pos..start_pos + request_len as usize),
+                ) {
+                    Ok(request) => {
+                        prev_position += request_len as usize + bytes_read;
+                        results.push(request);
+                    }
+                    Err(err) => {
+                        log_error("parse input", format!("Failed to parse request: {err}"));
+                        return Err(err.into());
                     }
                 }
             } else {
@@ -68,7 +67,7 @@ mod tests {
     use crate::command_request::{command, command_request};
     use crate::command_request::{Command, CommandRequest, RequestType};
     use bytes::BufMut;
-    use rand::{distributions::Alphanumeric, Rng};
+    use rand::{distr::Alphanumeric, Rng};
     use rstest::rstest;
 
     fn write_length(buffer: &mut BytesMut, length: u32) {
@@ -161,7 +160,7 @@ mod tests {
     }
 
     fn generate_random_string(length: usize) -> String {
-        rand::thread_rng()
+        rand::rng()
             .sample_iter(&Alphanumeric)
             .take(length)
             .map(char::from)

Diff for: glide-core/src/socket_listener.rs

@@ -11,7 +11,6 @@ use crate::errors::{error_message, error_type, RequestErrorType};
 use crate::response;
 use crate::response::Response;
 use bytes::Bytes;
-use directories::BaseDirs;
 use logger_core::{log_debug, log_error, log_info, log_trace, log_warn};
 use once_cell::sync::Lazy;
 use protobuf::{Chars, Message};
@@ -22,6 +21,7 @@ use redis::cluster_routing::{ResponsePolicy, Routable};
 use redis::{ClusterScanArgs, Cmd, PushInfo, RedisError, ScanStateRC, Value};
 use std::cell::Cell;
 use std::collections::HashSet;
+use std::path::PathBuf;
 use std::ptr::from_mut;
 use std::rc::Rc;
 use std::sync::RwLock;
@@ -39,7 +39,9 @@ use ClosingReason::*;
 use PipeListeningResult::*;
 
 /// The socket file name
-const SOCKET_FILE_NAME: &str = "glide-socket";
+const SOCKET_FILE_NAME: &str = "glide-socket.soc";
+/// The socket folder
+const SOCKET_FOLDER: &str = "glide";
 
 /// The maximum length of a request's arguments to be passed as a vector of
 /// strings instead of a pointer
@@ -569,7 +571,7 @@ async fn handle_requests(
 
 pub fn close_socket(socket_path: &String) {
     log_info("close_socket", format!("closing socket at {socket_path}"));
-    let _ = std::fs::remove_file(socket_path);
+    remove_socket_dir();
 }
 
 async fn create_client(
@@ -783,29 +785,39 @@ struct ClosingError {
 /// Get the socket full path.
 /// The socket file name will contain the process ID and will try to be saved into the user's runtime directory
 /// (e.g. /run/user/1000) in Unix systems. If the runtime dir isn't found, the socket file will be saved to the temp dir.
-/// For Windows, the socket file will be saved to %AppData%\Local.
 pub fn get_socket_path_from_name(socket_name: String) -> String {
-    let base_dirs = BaseDirs::new().expect("Failed to create BaseDirs");
-    let tmp_dir;
-    let folder = if cfg!(windows) {
-        base_dirs.data_local_dir()
-    } else {
-        base_dirs.runtime_dir().unwrap_or({
-            tmp_dir = env::temp_dir();
-            tmp_dir.as_path()
-        })
-    };
-    folder
+    get_socket_dir()
         .join(socket_name)
         .into_os_string()
         .into_string()
-        .expect("Couldn't create socket path")
+        .expect("Failed to create socket path from name")
+}
+
+/// Get the socket directory path.
+fn get_socket_dir() -> PathBuf {
+    // Use XDG_RUNTIME_DIR if available, else fallback to temp directory
+    if let Ok(runtime_dir) = env::var("XDG_RUNTIME_DIR") {
+        PathBuf::from(runtime_dir)
+            .join(SOCKET_FOLDER)
+            .join(std::process::id().to_string())
+    } else {
+        env::temp_dir()
+            .join(SOCKET_FOLDER)
+            .join(std::process::id().to_string())
+    }
+}
+
+/// Remove socket dir of the process
+pub fn remove_socket_dir() {
+    let socket_dir = get_socket_dir();
+    if socket_dir.exists() {
+        let _ = std::fs::remove_dir_all(socket_dir);
+    }
 }
 
 /// Get the socket path as a string
 pub fn get_socket_path() -> String {
-    let socket_name = format!("{}-{}", SOCKET_FILE_NAME, std::process::id());
-    get_socket_path_from_name(socket_name)
+    get_socket_path_from_name(SOCKET_FILE_NAME.to_string())
 }
 
 /// This function is exposed only for the sake of testing with a nonstandard `socket_path`.
@@ -820,7 +832,10 @@ pub fn start_socket_listener_internal<InitCallback>(
     static INITIALIZED_SOCKETS: Lazy<RwLock<HashSet<String>>> =
         Lazy::new(|| RwLock::new(HashSet::new()));
 
-    let socket_path = socket_path.unwrap_or_else(get_socket_path);
+    let socket_path = match socket_path {
+        Some(path) => path,
+        None => get_socket_path(),
+    };
 
     {
         // Optimize for already initialized
@@ -841,7 +856,6 @@ pub fn start_socket_listener_internal<InitCallback>(
         init_callback(Ok(socket_path.clone()));
         return;
     }
-
     let (tx, rx) = std::sync::mpsc::channel();
     let socket_path_cloned = socket_path.clone();
     let init_callback_cloned = init_callback.clone();
@@ -860,8 +874,16 @@ pub fn start_socket_listener_internal<InitCallback>(
                     }
                     Ok(runtime) => runtime,
                 };
-
                 runtime.block_on(async move {
+                    // Clean up any leftover from previous runs and create socket dir
+                    remove_socket_dir();
+                    if let Err(err) = std::fs::create_dir_all(get_socket_dir()) {
+                        log_error(
+                            "listen_on_socket",
+                            format!("Failed to create socket directory: {err}"),
+                        );
+                    }
+
                     let listener_socket = match UnixListener::bind(socket_path_cloned.clone()) {
                         Err(err) => {
                             log_error(
@@ -874,8 +896,6 @@ pub fn start_socket_listener_internal<InitCallback>(
                     };
 
                     // Signal initialization is successful.
-                    // IMPORTANT:
-                    // tx.send() must be called before init_callback_cloned() to ensure runtimes, such as Python, can properly complete the main function
                     let _ = tx.send(true);
                     init_callback_cloned(Ok(socket_path_cloned.clone()));
 
@@ -900,7 +920,9 @@ pub fn start_socket_listener_internal<InitCallback>(
                     drop(listener_socket);
                     let _ = std::fs::remove_file(socket_path_cloned.clone());
 
-                    // no more listening on socket - update the sockets db
+                    // Clean the entire process-id socket directory on close
+                    remove_socket_dir();
+
                     let mut sockets_write_guard = INITIALIZED_SOCKETS
                         .write()
                         .expect("Failed to acquire sockets db write guard");
@@ -917,7 +939,6 @@ pub fn start_socket_listener_internal<InitCallback>(
         })
         .expect("Thread spawn failed. Cannot report error because callback was moved.");
 
-    // wait for thread initialization signaling, callback invocation is done in the thread
     let _ = rx.recv().map(|res| {
         if res {
             sockets_write_guard.insert(socket_path);

Diff for: glide-core/tests/utilities/mod.rs

@@ -7,7 +7,7 @@ use glide_core::{
     connection_request::{self, AuthenticationInfo, NodeAddress, ProtocolVersion},
 };
 use once_cell::sync::Lazy;
-use rand::{distributions::Alphanumeric, Rng};
+use rand::{distr::Alphanumeric, Rng};
 use redis::{
     cluster_routing::{MultipleNodeRoutingInfo, RoutingInfo},
     ConnectionAddr, GlideConnectionOptions, PushInfo, RedisConnectionInfo, RedisResult, Value,
@@ -517,7 +517,7 @@ pub fn get_address_info(address: &ConnectionAddr) -> NodeAddress {
 }
 
 pub fn generate_random_string(length: usize) -> String {
-    rand::thread_rng()
+    rand::rng()
         .sample_iter(&Alphanumeric)
         .take(length)
         .map(char::from)