# 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()

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

        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()

"""
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)