diff -urN a/salt/transport/ipc.py b/salt/transport/ipc.py
--- a/salt/transport/ipc.py	2023-12-13 13:00:00.000000000 +0100
+++ b/salt/transport/ipc.py	2024-01-16 13:12:36.365271409 +0100
@@ -91,7 +91,9 @@
         "close",
     ]
 
-    def __init__(self, socket_path, io_loop=None, payload_handler=None):
+    def __init__(
+        self, socket_path, io_loop=None, payload_handler=None, ratelimiter=None
+    ):
         """
         Create a new Tornado IPC server
 
@@ -115,6 +117,7 @@
         self.sock = None
         self.io_loop = io_loop or salt.ext.tornado.ioloop.IOLoop.current()
         self._closing = False
+        self._ratelimiter = ratelimiter
 
     def start(self):
         """
@@ -179,12 +182,18 @@
         else:
             msgpack_kwargs = {"encoding": "utf-8"}
         unpacker = salt.utils.msgpack.Unpacker(**msgpack_kwargs)
+        if self._ratelimiter is not None:
+            self._ratelimiter.set_queue_handlers(self.io_loop, self.payload_handler, _null)
         while not stream.closed():
             try:
                 wire_bytes = yield stream.read_bytes(4096, partial=True)
                 unpacker.feed(wire_bytes)
                 for framed_msg in unpacker:
                     body = framed_msg["body"]
+                    if self._ratelimiter is not None and self._ratelimiter.exceeds(
+                        framed_msg
+                    ):
+                        continue
                     self.io_loop.spawn_callback(
                         self.payload_handler,
                         body,
@@ -438,7 +447,7 @@
     # FIXME timeout unimplemented
     # FIXME tries unimplemented
     @salt.ext.tornado.gen.coroutine
-    def send(self, msg, timeout=None, tries=None):
+    def send(self, msg, timeout=None, tries=None, header=None):
         """
         Send a message to an IPC socket
 
@@ -448,9 +457,27 @@
         :param int timeout: Timeout when sending message (Currently unimplemented)
         """
         if not self.connected():
-            yield self.connect()
-        pack = salt.transport.frame.frame_msg_ipc(msg, raw_body=True)
-        yield self.stream.write(pack)
+            max_retries = 5
+            retry = 1
+            while retry <= max_retries:
+                try:
+                    self.stream = None
+                    yield self.connect()
+                    break
+                except StreamClosedError:
+                    log.warning(
+                        "IPCMessageClient: Unable to reconnect IPC stream, retry %d of %d",
+                        retry,
+                        max_retries
+                    )
+                yield salt.ext.tornado.gen.sleep(3)
+                retry += 1
+        pack = salt.transport.frame.frame_msg_ipc(msg, raw_body=True, header=header)
+        try:
+            yield self.stream.write(pack)
+        except StreamClosedError:
+            if not self._closing:
+                log.warning("IPCMessageClient: Stream was closed")
 
 
 class IPCMessageServer(IPCServer):
@@ -493,7 +520,7 @@
     but using either UNIX domain sockets or TCP sockets
     """
 
-    def __init__(self, opts, socket_path, io_loop=None):
+    def __init__(self, opts, socket_path, io_loop=None, ratelimiter=None):
         """
         Create a new Tornado IPC server
         :param dict opts: Salt options
@@ -514,6 +541,7 @@
         # Placeholders for attributes to be populated by method calls
         self.sock = None
         self.io_loop = io_loop or IOLoop.current()
+        self._ratelimiter = ratelimiter
         self._closing = False
         self.streams = set()
 
@@ -546,14 +574,21 @@
     def _write(self, stream, pack):
         try:
             yield stream.write(pack)
+            if self._ratelimiter is not None:
+                self._ratelimiter.update_out(-1)
+                self._ratelimiter.update_out_str(id(stream), -1)
         except StreamClosedError:
             log.trace("Client disconnected from IPC %s", self.socket_path)
             self.streams.discard(stream)
+            if self._ratelimiter is not None:
+                self._ratelimiter.update_out_drop(id(stream))
         except Exception as exc:  # pylint: disable=broad-except
             log.error("Exception occurred while handling stream: %s", exc)
             if not stream.closed():
                 stream.close()
             self.streams.discard(stream)
+            if self._ratelimiter is not None:
+                self._ratelimiter.update_out_drop(id(stream))
 
     def publish(self, msg):
         """
@@ -563,8 +598,12 @@
             return
 
         pack = salt.transport.frame.frame_msg_ipc(msg, raw_body=True)
+        if self._ratelimiter is not None:
+            self._ratelimiter.update_out(len(self.streams))
         for stream in self.streams:
             self.io_loop.spawn_callback(self._write, stream, pack)
+            if self._ratelimiter is not None:
+                self._ratelimiter.update_out_str(id(stream), 1)
 
     def handle_connection(self, connection, address):
         log.trace("IPCServer: Handling connection to address: %s", address)
@@ -579,6 +618,7 @@
             with salt.utils.asynchronous.current_ioloop(self.io_loop):
                 stream = IOStream(connection, **kwargs)
             self.streams.add(stream)
+            log.critical("DBG: stream: %s, connect: %s", id(stream), connection)
 
             def discard_after_closed():
                 self.streams.discard(stream)
@@ -691,6 +731,9 @@
                     self.unpacker.feed(wire_bytes)
                     first_sync_msg = True
                     for framed_msg in self.unpacker:
+                        if not isinstance(framed_msg, dict):
+                            log.debug("Improper message received: %s", framed_msg)
+                            continue
                         if callback:
                             self.io_loop.spawn_callback(callback, framed_msg["body"])
                         elif first_sync_msg:
diff -urN a/salt/utils/asynchronous.py b/salt/utils/asynchronous.py
--- a/salt/utils/asynchronous.py	2023-12-13 13:00:00.000000000 +0100
+++ b/salt/utils/asynchronous.py	2024-01-16 07:43:11.672103008 +0100
@@ -111,30 +111,10 @@
 
     def _wrap(self, key):
         def wrap(*args, **kwargs):
-            results = []
-            thread = threading.Thread(
-                target=self._target,
-                args=(key, args, kwargs, results, self.io_loop),
-            )
-            thread.start()
-            thread.join()
-            if results[0]:
-                return results[1]
-            else:
-                exc_info = results[1]
-                raise exc_info[1].with_traceback(exc_info[2])
+            return self.io_loop.run_sync(lambda: getattr(self.obj, key)(*args, **kwargs))
 
         return wrap
 
-    def _target(self, key, args, kwargs, results, io_loop):
-        try:
-            result = io_loop.run_sync(lambda: getattr(self.obj, key)(*args, **kwargs))
-            results.append(True)
-            results.append(result)
-        except Exception:  # pylint: disable=broad-except
-            results.append(False)
-            results.append(sys.exc_info())
-
     def __enter__(self):
         return self
 
diff -urN a/salt/utils/event.py b/salt/utils/event.py
--- a/salt/utils/event.py	2023-12-13 13:00:00.000000000 +0100
+++ b/salt/utils/event.py	2024-01-14 11:10:07.305545940 +0100
@@ -76,6 +76,10 @@
 import salt.utils.stringutils
 import salt.utils.zeromq
 
+from salt._logging.impl import get_lowest_log_level, set_lowest_log_level_by_opts
+from salt.utils.ratelimiter import RateLimiter
+from salt.exceptions import SaltDeserializationError
+
 log = logging.getLogger(__name__)
 
 # The SUB_EVENT set is for functions that require events fired based on
@@ -270,6 +274,10 @@
             # the buffer will grow resulting in big memory usage.
             self.connect_pub()
 
+        set_lowest_log_level_by_opts(opts)
+        log_level = get_lowest_log_level()
+        self._trace_enabled = log_level is not None and log_level <= logging.TRACE
+
     @classmethod
     def __load_cache_regex(cls):
         """
@@ -461,7 +469,11 @@
             salt.utils.stringutils.to_bytes(TAGEND)
         )  # split tag from data
         mtag = salt.utils.stringutils.to_str(mtag)
-        data = salt.payload.loads(mdata, encoding="utf-8")
+        try:
+            data = salt.payload.loads(mdata, encoding="utf-8")
+        except SaltDeserializationError:
+            log.warning("SaltDeserializationError on reading data, the payload could be incomplete")
+            raise
         return mtag, data
 
     def _get_match_func(self, match_type=None):
@@ -592,15 +604,18 @@
                     pmatch_func(ret["tag"], ptag)
                     for ptag, pmatch_func in self.pending_tags
                 ):
-                    log.trace("get_event() caching unwanted event = %s", ret)
+                    if self._trace_enabled:
+                        log.trace("get_event() caching unwanted event = %s", ret)
                     self.pending_events.append(ret)
                 if wait:  # only update the wait timeout if we had one
                     wait = timeout_at - time.time()
                 continue
 
-            log.trace("get_event() received = %s", ret)
+            if self._trace_enabled:
+                log.trace("get_event() received = %s", ret)
             return ret
-        log.trace("_get_event() waited %s seconds and received nothing", wait)
+        if self._trace_enabled:
+            log.trace("_get_event() waited %s seconds and received nothing", wait)
         return None
 
     def get_event(
@@ -657,7 +672,8 @@
         request, it MUST subscribe the result to ensure the response is not lost
         should other regions of code call get_event for other purposes.
         """
-        log.trace("Get event. tag: %s", tag)
+        if self._trace_enabled:
+            log.trace("Get event. tag: %s", tag)
         assert self._run_io_loop_sync
 
         match_func = self._get_match_func(match_type)
@@ -676,9 +692,16 @@
                             self.close_pub()
                             self.connect_pub(timeout=wait)
                             continue
+                        except SaltDeserializationError:
+                            log.warning("Unable to deserialize received Salt event")
+                            ret = None
                     self.raise_errors = raise_errors
                 else:
-                    ret = self._get_event(wait, tag, match_func, no_block)
+                    try:
+                        ret = self._get_event(wait, tag, match_func, no_block)
+                    except SaltDeserializationError:
+                        log.warning("Unable to deserialize received Salt event")
+                        ret = None
 
         if ret is None or full:
             return ret
@@ -832,7 +855,10 @@
         if self._run_io_loop_sync:
             with salt.utils.asynchronous.current_ioloop(self.io_loop):
                 try:
-                    self.pusher.send(msg)
+                    header = {"_tag": tag, "_ts": time.time()}
+                    if "id" in data:
+                        header["_id"] = data["id"]
+                    self.pusher.send(msg, header=header)
                 except Exception as exc:  # pylint: disable=broad-except
                     log.debug(
                         "Publisher send failed with exception: %s",
@@ -1175,6 +1201,7 @@
         self.io_loop = None
         self.puller = None
         self.publisher = None
+        self._ratelimiter = None
 
     def run(self):
         """
@@ -1199,14 +1226,17 @@
                 epub_uri = os.path.join(self.opts["sock_dir"], "master_event_pub.ipc")
                 epull_uri = os.path.join(self.opts["sock_dir"], "master_event_pull.ipc")
 
+            self._ratelimiter = RateLimiter(self.opts)
+
             self.publisher = salt.transport.ipc.IPCMessagePublisher(
-                self.opts, epub_uri, io_loop=self.io_loop
+                self.opts, epub_uri, io_loop=self.io_loop, ratelimiter=self._ratelimiter
             )
 
             self.puller = salt.transport.ipc.IPCMessageServer(
                 epull_uri,
                 io_loop=self.io_loop,
                 payload_handler=self.handle_publish,
+                ratelimiter=self._ratelimiter,
             )
 
             # Start the master event publisher
diff -urN a/salt/utils/ratelimiter.py b/salt/utils/ratelimiter.py
--- a/salt/utils/ratelimiter.py	1970-01-01 01:00:00.000000000 +0100
+++ b/salt/utils/ratelimiter.py	2024-01-17 11:38:03.885303558 +0100
@@ -0,0 +1,299 @@
+import itertools
+import logging
+
+import salt.ext.tornado.gen
+
+from threading import Thread
+from time import sleep, time
+
+from salt.ext.tornado.ioloop import PeriodicCallback
+
+log = logging.getLogger(__name__)
+
+
+class Counter:
+    def __init__(self):
+        self._incs = itertools.count()
+        self._decs = itertools.count()
+
+    def inc(self):
+        next(self._incs)
+
+    def dec(self):
+        next(self._decs)
+
+    def update(self, num):
+        if num > 0:
+            cnt = self._incs
+        else:
+            cnt = self._decs
+
+        for _ in range(abs(num)):
+            next(cnt)
+
+    def value(self):
+        return next(self._incs) - next(self._decs)
+
+
+class RateLimiter:
+    def __init__(self, opts={}):
+        rate_limit_opts = opts.get("event_ratelimiter", {})
+        self._enabled = rate_limit_opts.get("enabled", False)
+        self._report_interval = rate_limit_opts.get("report_interval", 0)
+        self._report_count = rate_limit_opts.get("report_count", 20)
+        self._time_frame = rate_limit_opts.get("time_frame", 600)
+        self._max_auth = rate_limit_opts.get("max_auth", 10)
+        self._ignore_time = rate_limit_opts.get("ignore_time", 20)
+        self._ignore_force_remove = rate_limit_opts.get("ignore_force_remove", 120)
+        self._ignore_cleanup_interval = rate_limit_opts.get(
+            "ignore_cleanup_interval", 20
+        )
+        self._queue_enabled = rate_limit_opts.get("queue_enabled", False)
+        self._queue_start_on = rate_limit_opts.get("queue_start_on", 100000)
+        self._queue_out_max = rate_limit_opts.get("queue_out_max", 200000)
+        self._queue_check_interval = rate_limit_opts.get("queue_check_interval", 20)
+        self._queue_max_batch = rate_limit_opts.get("queue_max_batch", 200)
+        self._queue_interval = rate_limit_opts.get("queue_interval", 1000)
+
+        self._io_loop = None
+        self._payload_handler = None
+        self._write_callback = None
+
+        self._msgs_queue_in = Counter()
+        self._msgs_queue_out = Counter()
+        self._msgs_ignored = Counter()
+        self._msgs_passed = Counter()
+
+        self._msgs_stream = {}
+
+        self._msgs_idx = 0
+
+        self._incoming = []
+
+        self._data = {}
+        self._data_last = {}
+        self._data_max = {}
+        self._ignore = {}
+
+        self._queue = []
+        self._queue_armed = False
+        self._queue_max_flex = 0
+
+        if self._report_interval > 0:
+            self._thread = Thread(target=self.report_rates)
+            self._thread.start()
+
+        if self._enabled:
+            self._service_thread = Thread(target=self._service)
+            self._service_thread.start()
+
+        self._queue_periodic_cb = None
+
+    def set_queue_handlers(self, io_loop, payload_handler, write_callback):
+        self._io_loop = io_loop
+        self._payload_handler = payload_handler
+        self._write_callback = write_callback
+        if self._queue_periodic_cb is None:
+            self._queue_periodic_cb = PeriodicCallback(
+                self._process_queue, self._queue_interval, io_loop=self._io_loop
+            )
+            self._queue_periodic_cb.start()
+
+    def exceeds(self, msg):
+        if not self._enabled:
+            self._msgs_queue_in.inc()
+            self._msgs_passed.inc()
+            return False
+
+        header = msg.get("head", {})
+
+        tag = header.get("_tag")
+        minion_id = header.get("_id")
+        ts = header.get("_ts")
+
+        if tag is None or tag != "salt/auth" or minion_id is None or ts is None:
+            self._msgs_queue_in.inc()
+            self._msgs_passed.inc()
+            return False
+
+        self._incoming.append((minion_id, ts))
+
+        if self._queue_armed:
+            self._queue.append((header, msg["body"]))
+            return True
+
+        if minion_id in self._ignore:
+            self._msgs_ignored.inc()
+            return True
+
+        self._msgs_queue_in.inc()
+        self._msgs_passed.inc()
+        return False
+
+    def update_out(self, num):
+        if num > 0:
+            self._msgs_queue_in.dec()
+            self._msgs_idx = num
+        self._msgs_queue_out.update(num)
+
+    def update_out_str(self, sid, num):
+        if sid not in self._msgs_stream:
+            self._msgs_stream[sid] = Counter()
+        self._msgs_stream[sid].update(num)
+
+    def update_out_drop(self, sid):
+        if sid not in self._msgs_stream:
+            return
+        counter = self._msgs_stream.pop(sid).value()
+        self.update_out(counter * -1)
+        log.critical("DBG: stream: %s dropped with counter: %d", sid, counter)
+
+    def _service(self):
+        cts = time()
+        last_ignore_cleanup = cts
+        last_queue_check = cts
+        while True:
+            sleep(0.2)
+            cts = time()
+            wipe_before = cts - self._time_frame
+            incoming = []
+            while self._incoming:
+                (minion_id, ts) = self._incoming.pop(0)
+                if minion_id not in self._data:
+                    self._data[minion_id] = []
+                self._data[minion_id].append(ts)
+                i = len(self._data[minion_id])
+                while i > 0:
+                    i -= 1
+                    if self._data[minion_id][i] < wipe_before:
+                        del self._data[minion_id][i]
+                minion_ev_cnt = len(self._data[minion_id])
+                self._data_max[minion_id] = max(
+                    self._data_max.get(minion_id, 0), minion_ev_cnt
+                )
+                self._data_last[minion_id] = max(self._data_last.get(minion_id, ts), ts)
+                if (
+                    len(self._data[minion_id]) >= self._max_auth
+                    and minion_id not in self._ignore
+                ):
+                    self._ignore[minion_id] = ts
+                    log.warning(
+                        "Minion '%s' exceeded the amount of %d 'salt/auth' events per %d sec. "
+                        "Ignoring all further 'salt/auth' events for it.",
+                        minion_id,
+                        self._max_auth,
+                        self._time_frame,
+                    )
+            if self._queue_enabled and cts > (
+                last_queue_check + self._queue_check_interval
+            ):
+                last_queue_check = cts
+                cur_queue_in = self._msgs_queue_in.value()
+                cur_queue_out = self._msgs_queue_out.value()
+                self._queue_armed = cur_queue_out > self._queue_start_on
+                self._queue_max_flex = max(
+                    int(
+                        (
+                            max(self._queue_out_max - cur_queue_out, 1)
+                            / max(self._msgs_idx, 1)
+                        )
+                        / self._queue_check_interval
+                    ),
+                    1,
+                )
+            if cts > (last_ignore_cleanup + self._ignore_cleanup_interval):
+                last_ignore_cleanup = cts
+                to_remove = []
+                ignore_remove_before = cts - self._ignore_time
+                ignore_remove_force_before = cts - self._ignore_force_remove
+                for minion_id, its in self._ignore.items():
+                    i = len(self._data[minion_id])
+                    while i > 0:
+                        i -= 1
+                        if self._data[minion_id][i] < wipe_before:
+                            del self._data[minion_id][i]
+                    force_remove = its < ignore_remove_force_before
+                    if (
+                        its < ignore_remove_before
+                        and len(self._data[minion_id]) < self._max_auth
+                    ) or force_remove:
+                        log.warning(
+                            "Removing minion '%s' from the ignore list after beeing ignored for %d sec.%s",
+                            minion_id,
+                            round(cts - its),
+                            " (force remove)" if force_remove else "",
+                        )
+                        to_remove.append((minion_id, force_remove))
+                for minion_id, force_remove in to_remove:
+                    del self._ignore[minion_id]
+                    if force_remove:
+                        del self._data[minion_id]
+                to_remove = []
+                for minion_id, ts in self._data_last.items():
+                    if ts < wipe_before:
+                        to_remove.append(minion_id)
+                for minion_id in to_remove:
+                    del self._data[minion_id]
+                    del self._data_last[minion_id]
+
+    @salt.ext.tornado.gen.coroutine
+    def _process_queue(self):
+        if not self._queue:
+            return
+        ignore_before = time() - self._time_frame
+        pushed = 0
+        max_to_push = min(self._queue_max_batch, self._queue_max_flex)
+        while pushed < self._queue_max_batch and self._queue:
+            header, body = self._queue.pop()
+            if header["_id"] in self._ignore or header["_ts"] < ignore_before:
+                self._msgs_ignored.inc()
+                continue
+            self._io_loop.spawn_callback(
+                self._payload_handler,
+                body,
+                self._write_callback,
+            )
+            self._msgs_queue_in.inc()
+            self._msgs_passed.inc()
+            pushed += 1
+
+    def _report_rates(self):
+        log.critical(
+            "Ratelimit report: queued: %d[%d]->%d->%d, passed: %d, ignored: %d",
+            len(self._queue),
+            self._queue_max_flex,
+            self._msgs_queue_in.value(),
+            self._msgs_queue_out.value(),
+            self._msgs_passed.value(),
+            self._msgs_ignored.value(),
+        )
+        log.critical(
+            "Ratelimit report: %d minion(s) in the ignore list",
+            len(self._ignore),
+        )
+        counters = len(self._msgs_stream)
+        max_streams = [
+            (sid, counter.value()) for sid, counter in self._msgs_stream.items()
+        ]
+        max_streams = sorted(max_streams, key=lambda x: x[1], reverse=True)[:10]
+        for sid, counter in max_streams:
+            log.critical(
+                "Ratelimit report: stream counter %s: %d",
+                sid,
+                counter,
+            )
+        data_max = self._data_max
+        self._data_max = {}
+        max_rates = sorted(data_max.items(), key=lambda x: x[1], reverse=True)[
+            : self._report_count
+        ]
+        for minion_it, rate in max_rates:
+            log.critical("Ratelimit top rate: %s: %d", minion_it, rate)
+
+    def report_rates(self):
+        while True:
+            sleep(self._report_interval)
+            try:
+                self._report_rates()
+            except Exception as e:
+                log.warning("Exception while reporting rates: %s", e, exc_info=True)
diff -urN a/salt/utils/reactor.py b/salt/utils/reactor.py
--- a/salt/utils/reactor.py	2023-04-18 23:05:02.000000000 +0200
+++ b/salt/utils/reactor.py	2024-01-13 19:06:24.559989091 +0100
@@ -218,7 +218,7 @@
         ) as event:
             self.wrap = ReactWrap(self.opts)
 
-            for data in event.iter_events(full=True):
+            for data in event.iter_events(full=True, auto_reconnect=True):
                 # skip all events fired by ourselves
                 if data["data"].get("user") == self.wrap.event_user:
                     continue
