refactor task bridge to handle async tasks concurrently, preventing potential duplicate submission

.github/workflows/checks.yml

@@ -18,7 +18,6 @@ env:
   LANG: C.UTF-8 
   LC_ALL: C.UTF-8
 
-
 jobs:
   check:
     name: Format & Lint
@@ -57,4 +56,4 @@ jobs:
         if: success() || failure()  
         run: |
           redis-server --version
-          RUST_BACKTRACE=1 cargo test -- --nocapture
+          RUST_BACKTRACE=1 cargo test -- --nocapture

crates/worker/Cargo.toml

@@ -34,7 +34,7 @@ anyhow = { workspace = true }
 thiserror = "2.0.11"
 toml = { workspace = true }
 ctrlc = "3.4.5"
-tokio-util = { workspace = true }
+tokio-util = { workspace = true, features = ["rt"] }
 futures = { workspace = true }
 chrono = { workspace = true }
 serial_test = "0.5.1"
@@ -55,4 +55,5 @@ iroh = { workspace = true }
 rand_v8 = { workspace = true }
 rand_core_v6 = { workspace = true }
 dashmap = "6.1.0"
+tokio-stream = { version = "0.1.17", features = ["net"] }
 homedir = "0.3"

crates/worker/src/cli/command.rs

@@ -453,7 +453,7 @@ pub async fn execute_command(
                 gpu,
                 system_memory,
                 task_bridge
-                    .socket_path
+                    .get_socket_path()
                     .to_str()
                     .expect("path is valid utf-8 string")
                     .to_string(),
@@ -469,11 +469,10 @@ pub async fn execute_command(
 
             let bridge_cancellation_token = cancellation_token.clone();
             tokio::spawn(async move {
-                let bridge_clone = task_bridge.clone();
                 tokio::select! {
                     _ = bridge_cancellation_token.cancelled() => {
                     }
-                    _ = bridge_clone.run() => {
+                    _ = task_bridge.run() => {
                     }
                 }
             });