asyncio: _call_connection_lost() is called twice when resume_writing() closes the transport

[TOKUJI]

Bug report

Bug description:

Bug description:

A protocol that writes the next chunk from resume_writing() and closes the transport
when the source is exhausted causes a selector transport to call
_call_connection_lost() twice. The second call raises AttributeError into the event
loop's exception handler, once per connection.

PEP 3156 anticipates the reentrancy — resume_writing() "may be called directly by the
transport's write() method (as opposed to being called indirectly using
call_soon())" — and asyncio re-enters the transport from that callback itself:
SSLProtocol.resume_writing() writes via _process_outgoing(), and
test_write_buffer_after_close, merged with gh-128037, drives transport.write() from
resume_writing(). I found no documentation, docstring or PEP 3156 restriction on what
a protocol may do there, nor any indication that close() is treated differently from
write().

Expected: _call_connection_lost() is called at most once per connection, and the
program below completes without reaching the loop's exception handler.
Actual: the exception handler fires 20 times for 20 connections.

import asyncio, socket, sys

CHUNK = b'x' * (4 * 1024 * 1024)   # large enough to outrun the socket send buffer
CHUNKS = 2
errors = []


class Server(asyncio.Protocol):
    """Streams a fixed number of chunks under flow control, then closes."""

    def connection_made(self, transport):
        self.transport = transport
        self.chunks_left = CHUNKS
        transport.set_write_buffer_limits(high=64 * 1024, low=16 * 1024)
        self._send_next()

    def _send_next(self):
        if self.chunks_left:
            self.chunks_left -= 1
            self.transport.write(CHUNK)
        else:
            self.transport.close()         # end of stream

    def resume_writing(self):
        self._send_next()

    def connection_lost(self, exc):
        pass


async def client(port):
    reader, writer = await asyncio.open_connection('127.0.0.1', port)
    while await reader.read(65536):
        pass
    writer.close()


async def main():
    loop = asyncio.get_running_loop()
    loop.set_exception_handler(lambda l, ctx: errors.append(ctx))
    server = await loop.create_server(Server, '127.0.0.1', 0, family=socket.AF_INET)
    port = server.sockets[0].getsockname()[1]
    for _ in range(20):
        await client(port)
        await asyncio.sleep(0)             # let the scheduled callback run
    server.close()
    await server.wait_closed()


asyncio.run(main())
print('python', '.'.join(map(str, sys.version_info[:3])))
print('failures:', sum('_call_connection_lost' in c.get('message', '') for c in errors), '/ 20')

python 3.12.14  failures: 20 / 20
python 3.13.15  failures: 20 / 20
python 3.14.7   failures: 20 / 20

Ubuntu 24.04.4, kernel 6.17.0-1022-azure, EpollSelector. Also 20/20 on 3.12.3 /
3.13.14 / 3.14.6 on an unrelated host.

The client receives every byte, and the protocol gets one connection_made and one
connection_lost per connection; the failure is limited to the duplicate teardown. It is
not silent, though: a server written this way logs an unhandled-exception traceback for
every connection it serves.

Exception in callback _SelectorSocketTransport._call_connection_lost()
Traceback (most recent call last):
  File ".../asyncio/selector_events.py", line 910, in _call_connection_lost
    self._protocol.connection_lost(exc)
AttributeError: 'NoneType' object has no attribute 'connection_lost'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File ".../asyncio/events.py", line 94, in _run
    self._context.run(self._callback, *self._args)
  File ".../asyncio/selector_events.py", line 1199, in _call_connection_lost
    super()._call_connection_lost(exc)
  File ".../asyncio/selector_events.py", line 912, in _call_connection_lost
    self._sock.close()
AttributeError: 'NoneType' object has no attribute 'close'

(line numbers are 3.14.6)

Why it happens

_SelectorSocketTransport._write_sendmsg calls _maybe_resume_protocol() and then acts
on self._buffer and self._closing without re-reading what that call may have changed
(main, 24e5a55ccb):

        else:
            self._maybe_resume_protocol()  # May append to buffer.
            if not self._buffer:
                self._loop._remove_writer(self._sock_fd)
                if self._empty_waiter is not None:
                    self._empty_waiter.set_result(None)
                if self._closing:
                    self._call_connection_lost(None)
                ...

On one connection, in order:

1. _write_sendmsg drains the buffer and calls _maybe_resume_protocol().
2. resume_writing() calls close(). The buffer is already empty, so close() sets
   _closing, increments _conn_lost, and does call_soon(self._call_connection_lost).
3. Control returns to the block above. _closing is now true and _buffer is empty, so
   _write_sendmsg calls _call_connection_lost(None) synchronously — ahead of the
   callback scheduled in step 2. It delivers connection_lost(None) normally, then sets
   _sock, _protocol and _loop to None.
4. The callback scheduled in step 2 then runs from events.Handle._run, against the
   transport torn down in step 3. _protocol_connected is still true while _protocol is
   None, so it raises.

close() (if self._closing: return) and _force_close() (if self._conn_lost: return)
each prevent their own teardown path from being entered twice. The synchronous call in
_write_sendmsg bypasses both guards.

On all three versions, abort() in place of close() also reproduces 20/20, as does
forcing the non-sendmsg path (_write_send);
_SelectorDatagramTransport._sendto_ready appears by inspection to have the same
reentrancy, but I have not tested it. The streams API is unaffected (0/20 on the same
workload with await writer.drain()): FlowControlMixin.resume_writing() only
completes a future and never touches the transport.

Related issues

gh-128037 does not fix this: the reproducer fails on 3.12.3, which predates that change,
and on 3.14.6, which includes it; the call site has the same relevant shape as far back
as v3.4.0.
gh-115514 reports the same reentrancy with a different symptom
(Exception in callback None()), but provides no reproducer because its trigger is
timing-dependent. gh-128037 removed that symptom, not the double invocation; closing at
end of stream reaches the same state deterministically.

Candidate fix

gh-98703 gave the proactor transport a _called_connection_lost guard in 2022, so
one-shot teardown under a reentrant callback is already recognised on the other transport
family. selector_events.py — the path on every platform except Windows' proactor loop —
has no occurrence of that name, which suggests it may need the same protection.

A smaller variant exists: _conn_lost already distinguishes these cases at the
synchronous call site — it is 0 when close() deferred teardown because the buffer
was non-empty, and 1 when close() or _force_close() has already scheduled the
connection-lost callback. I tested if self._closing and not self._conn_lost: at the
three _call_connection_lost() call sites on 3.13.14: the reproducer goes from 20/20 to
0/20, and test_asyncio stays green (2,688 tests, 31 files, same before and after).

loop.call_soon(transport.close) in place of the direct close() avoids the issue
(0/20), so I am not blocked. I can prepare a PR for the selector-side guard if that
approach is preferred.

CPython versions tested on:

3.12, 3.13, 3.14

Operating systems tested on:

Linux