Wire ipc_write_buffer through to TCP transport per-stream cap

changelog/69930.fixed.md

@@ -0,0 +1 @@
+Wired the ``ipc_write_buffer`` master option through to the TCP transport in 3008.x.  The option remained in the config schema after the legacy ``salt.transport.ipc`` module was removed but was no longer read by any code path, so setting it in ``master.conf`` had no effect.  It now caps the per-stream Tornado outbound ``max_write_buffer_size`` on both ``PubServer`` (event-bus subscribers, plaintext and SSL-delayed paths) and ``SaltMessageServer`` (request/reply clients), matching the semantics of the legacy IPC module's per-connection cap.  The default (unset / ``0``) preserves the existing unlimited-buffer behavior; operators opt in by setting an explicit byte value.

salt/transport/tcp.py

@@ -737,11 +737,17 @@ def post_fork(self, message_handler, io_loop, **kwargs):
             ctx = None
             if self.ssl is not None:
                 ctx = salt.transport.base.ssl_context(self.ssl, server_side=True)
+            # See issue #69930: pass the configured cap through to the
+            # per-stream Tornado outbound write buffer.  ``ipc_write_buffer``
+            # is the legacy option name kept for master.conf compatibility;
+            # it was a no-op on 3008.x until this wiring was added.
+            max_write_buffer_size = self.opts.get("ipc_write_buffer") or None
             if USE_LOAD_BALANCER:
                 self.req_server = LoadBalancerWorker(
                     self.socket_queue,
                     self.handle_message,
                     ssl_options=ctx,
+                    max_write_buffer_size=max_write_buffer_size,
                 )
             else:
                 if salt.utils.platform.is_windows():
@@ -754,6 +760,7 @@ def post_fork(self, message_handler, io_loop, **kwargs):
                     self.handle_message,
                     ssl_options=ctx,
                     io_loop=io_loop,
+                    max_write_buffer_size=max_write_buffer_size,
                 )
                 self.req_server.add_socket(self._socket)
                 self._socket.listen(self.backlog)
@@ -806,6 +813,11 @@ class SaltMessageServer(tornado.tcpserver.TCPServer):
 
     def __init__(self, message_handler, *args, **kwargs):
         io_loop = kwargs.pop("io_loop", None) or tornado.ioloop.IOLoop.current()
+        # ``ipc_write_buffer`` (the legacy option name preserved for
+        # backwards-compat with ``master.conf``) caps the per-stream
+        # Tornado outbound write buffer.  ``0`` / ``None`` == unlimited
+        # (Tornado default), matching prior behavior.
+        self.max_write_buffer_size = kwargs.pop("max_write_buffer_size", None) or None
         self._closing = False
         super().__init__(*args, **kwargs)
         self.io_loop = io_loop
@@ -823,6 +835,12 @@ async def handle_stream(  # pylint: disable=arguments-differ,invalid-overridden-
         Handle incoming streams and add messages to the incoming queue
         """
         log.trace("Req client %s connected", address)
+        if self.max_write_buffer_size:
+            # See issue #69930: cap the outbound IOStream buffer per accepted
+            # request/reply client so a slow consumer can't grow it without
+            # bound.  Tornado's ``TCPServer`` builds the ``IOStream`` before
+            # dispatching to ``handle_stream``, so we set the attribute here.
+            stream.max_write_buffer_size = self.max_write_buffer_size
         self.clients.append((stream, address))
         unpacker = salt.utils.msgpack.Unpacker()
         try:
@@ -1384,11 +1402,28 @@ def handle_stream(self, stream, address):
                     self._validate_ssl_and_add_client(stream, address)
                 )
                 return
+        self._apply_write_buffer_cap(stream)
         client = Subscriber(stream, address)
         self.clients.add(client)
         stream.set_close_callback(self._discard_on_close(client))
         self.io_loop.create_task(self._stream_read(client))
 
+    def _apply_write_buffer_cap(self, stream):
+        """
+        Cap the accepted stream's outbound write buffer per ``ipc_write_buffer``.
+
+        See issue #69930: the legacy ``salt.transport.ipc`` module was
+        removed in 3008.x but the ``ipc_write_buffer`` opt remained in
+        the config schema.  Without this cap, Tornado defaults the
+        per-stream write buffer to unlimited, so a slow / blocked
+        event-bus subscriber lets the master's outbound bytearray grow
+        without bound (RSS growth observed on prod masters under event
+        burst).  ``0`` / falsy preserves prior behavior (unlimited).
+        """
+        cap = self.opts.get("ipc_write_buffer") or None
+        if cap:
+            stream.max_write_buffer_size = cap
+
     async def _validate_ssl_and_add_client(self, stream, address):
         """
         Validate SSL handshake completed successfully before accepting client.
@@ -1409,6 +1444,7 @@ async def _validate_ssl_and_add_client(self, stream, address):
                     return
 
                 # Successfully got cert - add client
+                self._apply_write_buffer_cap(stream)
                 client = Subscriber(stream, address)
                 self.clients.add(client)
                 stream.set_close_callback(self._discard_on_close(client))

tests/pytests/unit/transport/test_tcp.py

@@ -1507,3 +1507,170 @@ def closed(self):
     # The "boom" was dropped by the except-log-and-continue guard; the
     # other two got through.
     assert handled == ["ok1", "ok2"]
+
+
+# ---------------------------------------------------------------------------
+# issue #69930: ipc_write_buffer wired through to per-stream cap.
+# ---------------------------------------------------------------------------
+
+
+async def test_salt_message_server_applies_ipc_write_buffer(master_opts):
+    """
+    ``SaltMessageServer.handle_stream`` must set the accepted stream's
+    ``max_write_buffer_size`` to the ``ipc_write_buffer`` value passed
+    in.  Without this wiring (regression on 3008.x after the legacy
+    ``salt.transport.ipc`` module was dropped), setting
+    ``ipc_write_buffer`` in ``master.conf`` was a no-op and the
+    outbound IOStream buffer grew without bound under slow-consumer
+    conditions.  See issue #69930.
+    """
+
+    def handler(stream, body, header):  # pylint: disable=unused-argument
+        return None
+
+    cap = 12345
+    server = salt.transport.tcp.SaltMessageServer(handler, max_write_buffer_size=cap)
+
+    class Stream:
+        def __init__(self):
+            self.max_write_buffer_size = None
+
+        def read_bytes(self, *args, **kwargs):
+            raise tornado.iostream.StreamClosedError()
+
+    stream = Stream()
+    await server.handle_stream(stream, "client-cap")
+
+    assert stream.max_write_buffer_size == cap
+
+
+async def test_salt_message_server_no_cap_by_default(master_opts):
+    """
+    Not passing ``max_write_buffer_size`` (or passing 0) must leave the
+    stream untouched -- preserves Tornado's default (unlimited) and
+    matches prior behavior when ``ipc_write_buffer`` is not set in
+    ``master.conf``.
+    """
+
+    def handler(stream, body, header):  # pylint: disable=unused-argument
+        return None
+
+    server = salt.transport.tcp.SaltMessageServer(handler)
+    assert server.max_write_buffer_size is None
+
+    server_zero = salt.transport.tcp.SaltMessageServer(handler, max_write_buffer_size=0)
+    assert server_zero.max_write_buffer_size is None
+
+    class Stream:
+        def __init__(self):
+            self.max_write_buffer_size = "sentinel"
+
+        def read_bytes(self, *args, **kwargs):
+            raise tornado.iostream.StreamClosedError()
+
+    stream = Stream()
+    await server.handle_stream(stream, "client-nocap")
+    # Untouched -- the sentinel is still there.
+    assert stream.max_write_buffer_size == "sentinel"
+
+
+def test_pub_server_applies_ipc_write_buffer(master_opts, io_loop):
+    """
+    ``PubServer.handle_stream`` must set the accepted stream's
+    ``max_write_buffer_size`` to ``opts['ipc_write_buffer']`` when set.
+    See issue #69930.
+    """
+    master_opts["ipc_write_buffer"] = 54321
+    server = salt.transport.tcp.PubServer(master_opts, io_loop=io_loop)
+
+    class Stream:
+        def __init__(self):
+            self.max_write_buffer_size = None
+            self.socket = MagicMock()
+            self.socket.getpeercert.return_value = None
+            self._closed = False
+
+        def set_close_callback(self, cb):
+            pass
+
+        def close(self):
+            self._closed = True
+
+        def closed(self):
+            return self._closed
+
+    stream = Stream()
+    try:
+        with patch.object(
+            server, "_stream_read", MagicMock(return_value=None)
+        ), patch.object(server.io_loop, "create_task"):
+            server.handle_stream(stream, ("127.0.0.1", 12345))
+    finally:
+        server.close()
+
+    assert stream.max_write_buffer_size == 54321
+
+
+def test_pub_server_no_cap_when_ipc_write_buffer_zero(master_opts, io_loop):
+    """
+    ``ipc_write_buffer == 0`` (the default when the operator hasn't
+    opted in) must leave the stream's ``max_write_buffer_size``
+    untouched -- preserving Tornado's unlimited-write-buffer default.
+    """
+    master_opts["ipc_write_buffer"] = 0
+    server = salt.transport.tcp.PubServer(master_opts, io_loop=io_loop)
+
+    class Stream:
+        def __init__(self):
+            self.max_write_buffer_size = "sentinel"
+            self.socket = MagicMock()
+            self.socket.getpeercert.return_value = None
+            self._closed = False
+
+        def set_close_callback(self, cb):
+            pass
+
+        def close(self):
+            self._closed = True
+
+        def closed(self):
+            return self._closed
+
+    stream = Stream()
+    try:
+        with patch.object(
+            server, "_stream_read", MagicMock(return_value=None)
+        ), patch.object(server.io_loop, "create_task"):
+            server.handle_stream(stream, ("127.0.0.1", 12345))
+    finally:
+        server.close()
+
+    assert stream.max_write_buffer_size == "sentinel"
+
+
+def test_pub_server_apply_write_buffer_cap_helper(master_opts, io_loop):
+    """
+    ``_apply_write_buffer_cap`` is the shared helper used by both the
+    plaintext ``handle_stream`` path and the SSL-delayed
+    ``_validate_ssl_and_add_client`` path.  Verify the helper's contract
+    directly so both call sites are covered.
+    """
+    master_opts["ipc_write_buffer"] = 99999
+    server = salt.transport.tcp.PubServer(master_opts, io_loop=io_loop)
+
+    class Stream:
+        max_write_buffer_size = None
+
+    stream = Stream()
+    server._apply_write_buffer_cap(stream)
+    assert stream.max_write_buffer_size == 99999
+
+    master_opts["ipc_write_buffer"] = 0
+    server2 = salt.transport.tcp.PubServer(master_opts, io_loop=io_loop)
+
+    class Stream2:
+        max_write_buffer_size = "sentinel"
+
+    stream2 = Stream2()
+    server2._apply_write_buffer_cap(stream2)
+    assert stream2.max_write_buffer_size == "sentinel"