Conversation
`AsyncReqMessageClient.send()` enqueues `(future, message)` on `self._queue` and arms `_timeout_message()` to complete the future when the caller's timeout expires. When that fires the caller is released, but the queue entry is not removed -- it keeps pinning its serialized payload until the drain loop happens to reach it. Before 145a06e in-flight requests were tracked in `self._send_future_map`, a dict, so a timed-out entry could be removed by key. `Queue` has no removal-from-middle and `self._queue` has no maxsize, so an abandoned entry becomes unreachable garbage that still occupies the queue. A REQ socket permits one request/reply in flight, each drain iteration polls for up to 300ms, and `_send_recv()` restarts from scratch on every reconnect. Under sustained load the enqueue rate outruns the drain rate and the queue grows without bound. Observed on a 120-minion master: one process at ~53 GiB, OOM-killed roughly fortnightly. Skip requests whose future is already done: the reply can no longer be delivered to anyone, so there is no reason to spend a round trip on it. Two existing tests queued a single already-done future and relied on it reaching the send path to terminate the loop. With the entry now dropped they fall through to the idle poll branch, which calls `.result()` on the AsyncMock's coroutine. Both now queue a shutdown sentinel so the loop exits. Note this makes the saltstack#68506 repro structurally unreachable on this path -- a timed-out future never reaches `socket.send` -- so that test now asserts only its invariant, that the original timeout exception survives untouched. Fixes saltstack#68660
twangboy
approved these changes
Sep 23, 2026
This branch was successfully deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What does this PR do?
Skips requests whose caller future is already done when draining
AsyncReqMessageClient._queue, instead of sending them.What issues does this PR fix or reference?
Fixes #68660
Note: #68660 is currently closed, but the fix is not present on any
supported branch. I verified
3006.x,3007.x,3008.xandmaster— nonecontain a
future.done()check in the drain loop, and the regression commit145a06e ("Refactor send recv to avoid race conditions and limit memory usage")
is an ancestor of all four. Requesting reopen.
Previous Behavior
send()enqueues(future, message)and arms_timeout_message(), whichcompletes the future when the caller's timeout expires. The caller is released,
but its queue entry is not removed — it keeps pinning its serialized payload
until the drain loop reaches it.
Before 145a06e, in-flight requests lived in
self._send_future_map, a dict, soa timed-out entry was removed by key in O(1).
Queuehas noremoval-from-middle and
self._queuehas nomaxsize, so an abandoned entrybecomes unreachable garbage that still occupies the queue.
A REQ socket permits one request/reply in flight, each drain iteration polls
for up to 300 ms, and
_send_recv()restarts from scratch on every reconnect.Under sustained load the enqueue rate outruns the drain rate and the queue
grows without bound.
Observed on a 120-minion master: one process at ~53 GiB, OOM-killed roughly
fortnightly. The original reporter measured the same growth from 3006.16 and
confirmed 3006.15/3006.14 were unaffected.
New Behavior
A request whose future is already done is dropped — the reply can no longer be
delivered to anyone, so there is no reason to spend a round trip on it.
Tests written?
Yes.
test_client_send_recv_drops_abandoned_requestasserts the abandonedpayload never reaches
socket.send. Verified it fails without the fix(
AssertionError: Expected 'send' to not have been called. Called 1 times.)and passes with it.
Two existing tests queued a single already-done future and relied on it
reaching the send path to terminate the drain loop. With the entry now dropped
they fall through to the idle poll branch, which calls
.result()on anAsyncMock coroutine. Both now queue a
_REQ_QUEUE_SHUTDOWNsentinel so theloop exits.
Worth reviewer attention: this makes the #68506 repro structurally unreachable
on this path — a timed-out future never reaches
socket.send— sotest_client_send_recv_no_double_set_exception_after_timeoutnow asserts onlyits invariant (the original timeout exception survives untouched) rather than
exercising the double-set path. The
if not future.done()guards in theexception handlers are untouched and still protect the remaining paths.
Full
tests/pytests/unit/transport/test_zeromq.py: 35 passed, 2 skipped, 0failed.
pre-commitclean on the changed files; the onelint-saltfinding(
E0712atZeroMQSocketMonitor.stop) is pre-existing and byte-identical onunmodified
3006.x.Commits signed with GPG?
No