Ah, of course. Possibly this then: def _setup_transport(self): # Setup to _write() directly to the socket, and # do our own buffered reads.

Ah, of course. Possibly this then: def _setup_transport(self): # Setup to _write() directly to the socket, and # do our own buffered reads. self._write = self.sock.sendall self._read_buffer = EMPTY_BUFFER self._quick_recv = self.sock.recv (And similarly in SSLTransport.) Then, the OSError exception raised inside Transport.close also prevents Connection.collect from destroying the Transport object. Again, I'm not disagreeing with the original PR/fix. My point is only that when the socket object is destroyed, the socket is closed. I'll bet a try/except around this (Connection.collect) would also fix it: def collect(self): if self._transport: self._transport.close() Edit: we're discussing this in two places. My apologies. For interested readers: https://github.com/celery/celery/issues/4843#issuecomment-998220198 https://github.com/celery/celery/issues/4843#issuecomment-99... I'm satisfied I understand what's going on now, and that it was a bug in celery/py-aqmp retaining the socket object due to over-broad exception handling. Python closes sockets as expected when they are released.