`_ProactorDatagramTransport` never resumes a paused protocol after a write error

[graingert]

Bug report

Bug description:

On ProactorEventLoop, if a UDP write fails synchronously or asynchronously while the
write buffer is non-empty, _loop_writing() reports the error via error_received()
and just returns. It doesn't re-arm the write loop and doesn't call
_maybe_resume_protocol(). If the protocol was paused at that point, it stays paused
forever anything left in self._buffer is stranded, and a well-behaved protocol that
respects pause_writing() will never call sendto() again to un-stick it. Lost wakeup.

# Lib/asyncio/proactor_events.py, _ProactorDatagramTransport._loop_writing
        except OSError as exc:
            self._protocol.error_received(exc)
        except Exception as exc:
            self._fatal_error(exc, 'Fatal write error on datagram transport')
        else:
            self._write_fut.add_done_callback(self._loop_writing)
            self._maybe_resume_protocol()

_SelectorDatagramTransport._sendto_ready() has the identical bug on OSError, but it's
harmless there the fd is still registered with add_writer(), so the loop calls it
again next iteration and drains the buffer anyway. The proactor has no equivalent
re-trigger, so the same code is self-healing on selectors and terminal on IOCP.

Reproducer

Attached: proactor_datagram_write_pause.py. No mocking, no ICMP needed an oversized
datagram gives a deterministic write failure, and set_write_buffer_limits(0) makes the
pause deterministic (this is the limit anyio's UDP sockets use).

Confirmed on Windows, both 3.10.11 and 3.14.7:

fail-in-flight:
    events   = ['pause_writing', 'error_received([WinError 1784] ...)']
    buffered = 6
    verdict  = STRANDED

pause_writing fires, the write fails, and resume_writing never comes the protocol
sits paused with data stuck in the buffer until something outside asyncio forces another
sendto().

Linux (3.12.3, selector loop) is fine, as expected: the buffer never has a chance to fill
in the first place, so pause_writing never fires.

Suggested fix

        except OSError as exc:
            self._protocol.error_received(exc)
            if self._buffer and not self._conn_lost:
                self._loop.call_soon(self._loop_writing)
            else:
                self._maybe_resume_protocol()

Related

• asyncio proactor udp transport stops responding after send to port that isn't listening #91227 / gh-91227: Ignore ERROR_PORT_UNREACHABLE #32011 the read-side twin of this, fixed in 3.12 by swallowing
  ERROR_PORT_UNREACHABLE in IocpProactor._finish_recvfrom (not backported to 3.11)
• asyncio DatagramProtocol stops calling callbacks after OSError, on Windows #88906 closed by the above
• asyncio proactor udp transport stops responding after send to port that isn't listening due to ConnectionResetError from WSARecvFrom #127057 (open) WSARecvFrom raising ConnectionResetError synchronously still
  kills _loop_reading() the same way, on all versions

Your environment

• CPython versions tested: 3.10.11, 3.14.7
• OS: Windows 11, `ProactorEventLoop

here's the reproducer:

"""
Does _ProactorDatagramTransport._loop_writing() strand a paused protocol?

Run on Windows (ProactorEventLoop).  See README.md for what the output means.
"""

from __future__ import annotations

import asyncio
import sys

OVERSIZED = b"\x00" * 70000  # guaranteed oversend error (WSAEMSGSIZE or, on some
# builds, WinError 1784 ERROR_INVALID_USER_BUFFER); either way it fails the write


class Protocol(asyncio.DatagramProtocol):
    def connection_made(self, transport: asyncio.DatagramTransport) -> None:
        self.transport = transport
        self.events: list[str] = []
        self.error = asyncio.get_running_loop().create_future()
        # a high water mark of 0 makes the pause deterministic; it is what
        # anyio's UDP sockets use
        transport.set_write_buffer_limits(0)

    def datagram_received(self, data: bytes, addr: object) -> None:
        self.events.append(f"datagram_received({len(data)} bytes)")

    def pause_writing(self) -> None:
        self.events.append("pause_writing")

    def resume_writing(self) -> None:
        self.events.append("resume_writing")

    def error_received(self, exc: Exception) -> None:
        self.events.append(f"error_received({exc})")
        if not self.error.done():
            self.error.set_result(exc)


async def scenario(name: str, first: bytes, second: bytes) -> None:
    loop = asyncio.get_running_loop()
    transport, protocol = await loop.create_datagram_endpoint(
        Protocol, local_addr=("127.0.0.1", 0)
    )
    addr = transport.get_extra_info("sockname")

    # the first sendto() arms the overlapped write, so the second one has to be
    # buffered behind it, which trips pause_writing() at a high water mark of 0
    transport.sendto(first, addr)
    transport.sendto(second, addr)

    try:
        await asyncio.wait_for(protocol.error, 5)
    except asyncio.TimeoutError:
        protocol.events.append("<no error reported>")

    await asyncio.sleep(0.5)
    paused = "pause_writing" in protocol.events
    resumed = "resume_writing" in protocol.events
    buffered = transport.get_write_buffer_size()
    print(f"{name}:")
    print(f"    events   = {protocol.events}")
    print(f"    buffered = {buffered}")
    print(f"    verdict  = {'STRANDED' if paused and not resumed else 'ok'}")

    # recovery probe: an application send is the only thing that can re-arm the
    # write loop, which a protocol honouring pause_writing() would never do
    if paused and not resumed:
        transport.sendto(b"probe", addr)
        await asyncio.sleep(0.5)
        print(f"    after an unsolicited sendto(): {protocol.events}")

    transport.close()
    await asyncio.sleep(0.1)


async def main() -> None:
    print(sys.version)
    print(type(asyncio.get_running_loop()).__name__)
    print()
    # covers WSAEMSGSIZE being reported at completion of the in-flight write
    await scenario("fail-in-flight", OVERSIZED, b"queued")
    # covers WSAEMSGSIZE being reported synchronously by WSASendTo, from the
    # completion callback of the preceding write
    await scenario("fail-from-callback", b"ok", OVERSIZED)

CPython versions tested on:

3.14

Operating systems tested on:

Windows