chore(cubestore): Fix for timeout in worker services (#7359)

Author: waralexrom

rust/cubestore/cubestore/src/cluster/worker_pool.rs

@@ -342,6 +342,7 @@ impl<C: Configurator, P: WorkerProcessing, S: ServicesTransport> WorkerProcess<C
                 processor: PhantomData::<(C, P, S)>::default(),
                 services_sender: service_request_tx,
                 services_reciever: service_response_rx,
+                timeout: self.timeout.clone(),
             },
             &[title],
             &envs,
@@ -365,6 +366,7 @@ pub struct WorkerProcessArgs<C: Configurator, P: WorkerProcessing, S: ServicesTr
     processor: PhantomData<(C, P, S)>,
     services_sender: IpcSender<S::TransportRequest>,
     services_reciever: IpcReceiver<S::TransportResponse>,
+    timeout: Duration,
 }
 
 pub fn worker_main<C, P, S>(a: WorkerProcessArgs<C, P, S>) -> i32
@@ -373,8 +375,13 @@ where
     P: WorkerProcessing<Config = C::Config>,
     S: ServicesTransport<Request = C::ServicesRequest, Response = C::ServicesResponse>,
 {
-    let (rx, tx, services_sender, services_reciever) =
-        (a.args, a.results, a.services_sender, a.services_reciever);
+    let (rx, tx, services_sender, services_reciever, timeout) = (
+        a.args,
+        a.results,
+        a.services_sender,
+        a.services_reciever,
+        a.timeout,
+    );
     let mut tokio_builder = Builder::new_multi_thread();
     tokio_builder.enable_all();
     tokio_builder.thread_name("cubestore-worker");
@@ -386,7 +393,7 @@ where
     let runtime = tokio_builder.build().unwrap();
     worker_setup(&runtime);
     runtime.block_on(async move {
-        let services_client = S::connect(services_sender, services_reciever);
+        let services_client = S::connect(services_sender, services_reciever, timeout);
         let config = match C::configure(services_client).await {
             Err(e) => {
                 error!(

rust/cubestore/cubestore/src/cluster/worker_services.rs

@@ -80,8 +80,9 @@ pub trait ServicesTransport {
     fn connect(
         sender: IpcSender<Self::TransportRequest>,
         reciever: IpcReceiver<Self::TransportResponse>,
+        timeout: Duration,
     ) -> Arc<Self::Client> {
-        Self::Client::connect(sender, reciever)
+        Self::Client::connect(sender, reciever, timeout)
     }
 }
 
@@ -131,6 +132,7 @@ pub trait ServicesClient: Callable {
     fn connect(
         sender: IpcSender<Self::TransportRequest>,
         reciever: IpcReceiver<Self::TransportResponse>,
+        timeout: Duration,
     ) -> Arc<Self>;
 
     fn stop(&self);
@@ -158,9 +160,10 @@ impl<P: Callable> ServicesClient for ServicesClientImpl<P> {
     fn connect(
         sender: IpcSender<Self::TransportRequest>,
         reciever: IpcReceiver<Self::TransportResponse>,
+        timeout: Duration,
     ) -> Arc<Self> {
         let queue = Arc::new(unlimited::Queue::new());
-        let handle = Self::processing_loop(sender, reciever, queue.clone());
+        let handle = Self::processing_loop(sender, reciever, queue.clone(), timeout);
         Arc::new(Self {
             processor: PhantomData,
             handle,
@@ -197,6 +200,7 @@ impl<P: Callable> ServicesClientImpl<P> {
                 ServicesClientMessage<<Self as Callable>::Request, <Self as Callable>::Response>,
             >,
         >,
+        timeout: Duration,
     ) -> JoinHandle<()> {
         let (message_broadcast_tx, _) = broadcast::channel(10000);
 
@@ -252,7 +256,7 @@ impl<P: Callable> ServicesClientImpl<P> {
                 cube_ext::spawn(async move {
                     loop {
                         let broadcast_message = tokio::select! {
-                            _ = tokio::time::sleep(Duration::from_secs(5)) => { //TODO! config
+                            _ = tokio::time::sleep(timeout) => {
                                 Err(CubeError::internal(format!(
                                     "Worker service timeout for message id: {}",
                                     message_id