From 550751d0ff8b24351a6440ab583f649ac5d56c5c Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 27 Jul 2026 01:51:53 +0200 Subject: [PATCH] Fix the file descriptor send retry in asynctlvdata When sendmsg() reports EAGAIN, _sendmsg rescheduled itself with loop.add_reader(fd, _sendmsg, loop, fut, sock, fd) which waits for the socket to become readable rather than writable, and passes four of the six required arguments, so the callback raised TypeError once it did fire. Wait for writability and pass the message and descriptors through. Also skip the work in _recvmsg if the future was cancelled while waiting for data, as _sendmsg already does, so a cancelled read does not end in InvalidStateError from set_result. This module is imported by the server as well, so both paths are reached by the daemon whenever a descriptor is passed over the local socket. --- confluent_client/confluent/asynctlvdata.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/confluent_client/confluent/asynctlvdata.py b/confluent_client/confluent/asynctlvdata.py index 809880b5..e27b20d4 100644 --- a/confluent_client/confluent/asynctlvdata.py +++ b/confluent_client/confluent/asynctlvdata.py @@ -99,9 +99,9 @@ class ClientFile(object): -def _sendmsg(loop, fut, sock, msg, fds, rfd): - if rfd is not None: - loop.remove_reader(rfd) +def _sendmsg(loop, fut, sock, msg, fds, wfd): + if wfd is not None: + loop.remove_writer(wfd) if fut.cancelled(): return try: @@ -110,7 +110,7 @@ def _sendmsg(loop, fut, sock, msg, fds, rfd): [(socket.SOL_SOCKET, socket.SCM_RIGHTS, array.array("i", fds))]) except (BlockingIOError, InterruptedError): fd = sock.fileno() - loop.add_reader(fd, _sendmsg, loop, fut, sock, fd) + loop.add_writer(fd, _sendmsg, loop, fut, sock, msg, fds, fd) except Exception as exc: fut.set_exception(exc) else: @@ -127,6 +127,8 @@ def send_fds(sock, msg, fds): def _recvmsg(loop, fut, sock, msglen, maxfds, rfd): if rfd is not None: loop.remove_reader(rfd) + if fut.cancelled(): + return fds = array.array("i") # Array of ints try: msg, ancdata, flags, addr = sock.recvmsg(