Do not report a successful connection when the transport dropped - #2

• base_client.py — initialise self._connection_dropped = False.
• client.py / async_client.py — reset it at the start of each connect(), set it in _handle_eio_disconnect(), and raise ConnectionError('Connection dropped') instead of assigning connected = True when it is set.

Why a flag rather than checking self.eio.state. My first attempt gated the assignment on self.eio.state != 'connected'. It works at runtime but breaks 16 existing tests, because they drive a real engineio.Client with only eio.connect mocked, so state never leaves 'disconnected'. Narrowing to == 'disconnected' did not help for the same reason. The flag records the invariant we actually care about — did a disconnect land during this attempt — and leaves the test doubles alone.

A fix that did not work, recorded so nobody retries it: moving self.namespaces = {} out of the if self.connected: guard, so connect()'s existing set(self.namespaces) != set(self.connection_namespaces) check would fire. It fails 25/25, because _receive_packet dispatches with run_async=True, so _handle_connect runs after the disconnect cleanup and repopulates namespaces.

Self-contained, socketio + aiohttp only, no monkeypatching and no writes to any client attribute. The server accepts the namespace and then immediately closes.

import asyncio
import sys

import socketio
from aiohttp import web

OPEN = '0{"sid":"S","upgrades":[],"pingInterval":25000,"pingTimeout":20000,"maxPayload":1000000}'


async def server(request):
    ws = web.WebSocketResponse()
    await ws.prepare(request)
    await ws.send_str(OPEN)
    async for msg in ws:
        if str(msg.data).startswith('40'):
            await ws.send_str('40{"sid":"NS"}')  # accept the namespace
            await ws.send_str('1')               # then immediately close
            await ws.close()
            break
    return ws


async def main():
    app = web.Application()
    app.router.add_route('*', '/socket.io/', server)
    runner = web.AppRunner(app)
    await runner.setup()
    await web.TCPSite(runner, '127.0.0.1', 0).start()
    url = f'http://127.0.0.1:{runner.addresses[0][1]}'

    sio = socketio.AsyncClient(reconnection=False)
    try:
        await sio.connect(url, transports=['websocket'], wait_timeout=3)
        outcome = 'returned normally'
    except socketio.exceptions.ConnectionError as e:
        outcome = f'raised ConnectionError({str(e)!r})'
    await asyncio.sleep(0.2)

    wedged = sio.connected and sio.eio.state != 'connected'
    print(f'connect() {outcome}')
    print(f'connected={sio.connected!r}  eio.state={sio.eio.state!r}  wedged={wedged}')

    try:
        await sio.connect(url, transports=['websocket'], wait_timeout=3)
        reusable = 'yes'
    except socketio.exceptions.ConnectionError as e:
        reusable = 'no' if 'Already connected' in str(e) else 'yes (dropped again, as expected)'
    print(f'client still usable: {reusable}')

    await runner.cleanup()
    print('RESULT:', 'WEDGED' if wedged else 'HEALTHY')
    sys.exit(1 if wedged else 0)


asyncio.run(main())

Before:

connect() returned normally
connected=True  eio.state='disconnected'  wedged=True
client still usable: no
RESULT: WEDGED

After:

connect() raised ConnectionError('Connection dropped')
connected=False  eio.state='disconnected'  wedged=False
client still usable: yes (dropped again, as expected)
RESULT: HEALTHY

Deterministic — the wedge reproduced 25/25 on the unpatched build (macOS 15 arm64 and Linux python:3.12-slim), and a control that adds one 0.1 s server-side delay before closing reproduced 0/25.

What this proves: the wedged state is reachable and unrecoverable, and the fix removes it. What it does not prove: that a cooperative server produces it unaided — see below.

A plain cooperative AsyncServer calling sio.disconnect(sid) from its own connect handler does not reproduce, at any delay from 0 to 5 ms (0/45 attempts). With no proxy and no client-side await, the client always wins the race — the window is narrower than a network round-trip.

So the trigger above is fair to call staged. Two things widen the window in practice:

1. _receive_packet dispatches with run_async=True, so the message handler runs in a separate task, decoupled from the read loop.
2. _handle_connect populates self.namespaces[ns] and then awaits the user's connect handler before _connect_event.set(). Any await in that handler — an emit, a DB call, an HTTP request — widens it to the handler's duration.

A variant using a genuine AsyncServer with an ordinary async connect handler, severed in the network by a small TCP proxy, reproduces identically with a real server-issued sid in namespaces. That is what a load balancer, proxy or restarting pod does. Happy to add it if useful.

Precision note: _handle_eio_disconnect() does clear self.callbacks, self._binary_packet and self.sid unconditionally. Only the namespace teardown and self.connected = False sit inside the guard — which is enough.

disconnect() still cannot clear connected when the transport is already gone, because eio.disconnect() is guarded by if self.state == 'connected'. This PR removes the known path into that state rather than making the state escapable. Making disconnect() always reset connected would be a good belt-and-braces follow-up, but it is a separate behavioural change and I did not want to bundle it.

Found while investigating a downstream report of devices becoming permanently unreachable after a network change, with connected == True and no TCP connection in the process's socket list.