From ca288e87429aeb0dda2dbd96eb63853731267bb6 Mon Sep 17 00:00:00 2001
From: vzhestkov <vzhestkov@suse.com>
Date: Wed, 28 Feb 2024 09:54:44 +0100
Subject: [PATCH 1/3] Implement timeout and tries to
 transport.ipc.IPCClient.send

---
 salt/transport/ipc.py | 59 ++++++++++++++++++++++++++++++++++++++-----
 1 file changed, 52 insertions(+), 7 deletions(-)

diff --git a/salt/transport/ipc.py b/salt/transport/ipc.py
index cee100b086..5d1bec0c6d 100644
--- a/salt/transport/ipc.py
+++ b/salt/transport/ipc.py
@@ -2,7 +2,6 @@
 IPC transport classes
 """
 
-
 import errno
 import logging
 import socket
@@ -435,8 +434,6 @@ class IPCMessageClient(IPCClient):
         "close",
     ]
 
-    # FIXME timeout unimplemented
-    # FIXME tries unimplemented
     @salt.ext.tornado.gen.coroutine
     def send(self, msg, timeout=None, tries=None):
         """
@@ -445,12 +442,60 @@ def send(self, msg, timeout=None, tries=None):
         If the socket is not currently connected, a connection will be established.
 
         :param dict msg: The message to be sent
-        :param int timeout: Timeout when sending message (Currently unimplemented)
+        :param int timeout: Timeout when sending message
+        :param int tries: Maximum numer of tries to send message
         """
-        if not self.connected():
-            yield self.connect()
+        if tries is None or tries < 1:
+            tries = 1
+        due_time = None
+        if timeout is not None:
+            due_time = time.time() + timeout
+        _try = 1
+        exc_count = 0
         pack = salt.transport.frame.frame_msg_ipc(msg, raw_body=True)
-        yield self.stream.write(pack)
+        while _try <= tries:
+            if not self.connected():
+                self.close()
+                self.stream = None
+                self._closing = False
+                try:
+                    yield self.connect(
+                        timeout=(
+                            None if due_time is None else max(due_time - time.time(), 1)
+                        )
+                    )
+                except StreamClosedError:
+                    log.warning(
+                        "IPCMessageClient: Unable to reconnect IPC stream on sending message with ID: 0x%016x%s",
+                        id(msg),
+                        f", retry {_try} of {tries}" if tries > 1 else "",
+                    )
+                    exc_count += 1
+            if self.connected():
+                try:
+                    yield self.stream.write(pack)
+                    return
+                except StreamClosedError:
+                    if self._closing:
+                        break
+                    log.warning(
+                        "IPCMessageClient: Stream was closed on sending message with ID: 0x%016x",
+                        id(msg),
+                    )
+                    exc_count += 1
+                    if exc_count == 1:
+                        # Give one more chance in case if stream was detected as closed
+                        # on the first write attempt
+                        continue
+            cur_time = time.time()
+            _try += 1
+            if _try > tries or (due_time is not None and cur_time > due_time):
+                return
+            yield salt.ext.tornado.gen.sleep(
+                1
+                if due_time is None
+                else (due_time - cur_time) / max(tries - _try + 1, 1)
+            )
 
 
 class IPCMessageServer(IPCServer):

From 17d29cd87e9da863f27cea9c5617822b32ea68f6 Mon Sep 17 00:00:00 2001
From: vzhestkov <vzhestkov@suse.com>
Date: Wed, 28 Feb 2024 13:00:34 +0100
Subject: [PATCH 2/3] Make timeout and tries configurable for fire_event

---
 salt/utils/event.py | 21 +++++++++++++++++++--
 1 file changed, 19 insertions(+), 2 deletions(-)

diff --git a/salt/utils/event.py b/salt/utils/event.py
index e6d7b00520..e0ad254546 100644
--- a/salt/utils/event.py
+++ b/salt/utils/event.py
@@ -269,6 +269,10 @@ def __init__(
             # and don't read out events from the buffer on an on-going basis,
             # the buffer will grow resulting in big memory usage.
             self.connect_pub()
+        self.pusher_send_timeout = self.opts.get(
+            "pusher_send_timeout", self.opts.get("timeout")
+        )
+        self.pusher_send_tries = self.opts.get("pusher_send_tries", 3)
 
     @classmethod
     def __load_cache_regex(cls):
@@ -829,10 +833,18 @@ def fire_event(self, data, tag, timeout=1000):
             ]
         )
         msg = salt.utils.stringutils.to_bytes(event, "utf-8")
+        if timeout is None:
+            timeout_s = self.pusher_send_timeout
+        else:
+            timeout_s = float(timeout) / 1000
         if self._run_io_loop_sync:
             with salt.utils.asynchronous.current_ioloop(self.io_loop):
                 try:
-                    self.pusher.send(msg)
+                    self.pusher.send(
+                        msg,
+                        timeout=timeout_s,
+                        tries=self.pusher_send_tries,
+                    )
                 except Exception as exc:  # pylint: disable=broad-except
                     log.debug(
                         "Publisher send failed with exception: %s",
@@ -841,7 +853,12 @@ def fire_event(self, data, tag, timeout=1000):
                     )
                     raise
         else:
-            self.io_loop.spawn_callback(self.pusher.send, msg)
+            self.io_loop.spawn_callback(
+                self.pusher.send,
+                msg,
+                timeout=timeout_s,
+                tries=self.pusher_send_tries,
+            )
         return True
 
     def fire_master(self, data, tag, timeout=1000):

