bridge: call connection_lost "soon"

The standard Python asyncio Transport classes protect their caller from
having to deal with exceptions when calling .write().  In case an
exception does occur, they capture it and close the transport, passing
the exception on to the connection_lost() handler.  This makes it a lot
easier to interact with the transport: because OS-level errors are
handled out of band, protocols don't have to consider that any write
might fail.

Our Transport classes do the same thing, but there's an important
difference: the standard library defers the call to .connection_lost(),
but our transports invoke it immediately.

This creates a weird scenario whereby your connection_lost() handler can
get called in the middle of your connection_made() handler, if you
should happen to do a .write() there which fails.  Since
connection_lost() reasonably assumes that connection_made() has run, you
can get into all kinds of trouble.  Worst of all, once
.connection_lost() is done running, control flow returns back to your
.connection_made() handler, which continues running as if nothing has
happened in the meantime.

Let's adopt the standard Python behaviour here and defer the
.connection_lost() call.  We need to adjust some unit tests which
assumed the old behaviour: adjust them to ensure that we specifically
*don't* have that behaviour anymore.

src/cockpit/transports.py

@@ -142,7 +142,7 @@ def abort(self, exc: 'Exception | None' = None) -> None:
         self._closing = True
         self._close_reader()
         self._remove_write_queue()
-        self._protocol.connection_lost(exc)
+        self._loop.call_soon(self._protocol.connection_lost, exc)
         self._close()
 
     def can_write_eof(self) -> bool:

test/pytest/test_transport.py

@@ -255,6 +255,10 @@ async def test_cat(self) -> None:
         assert transport.get_returncode() == 0
         assert protocol.sent == protocol.received
         transport.close()
+        # make sure the connection_lost handler isn't called immediately
+        assert protocol.transport is not None
+        # ...but "soon" (in the very next mainloop iteration)
+        await asyncio.sleep(0.01)
         assert protocol.transport is None
 
     @pytest.mark.asyncio
@@ -330,6 +334,10 @@ async def test_broken_pipe(self) -> None:
         # Now let's write to the stdin with the other side closed.
         # This should be enough to immediately disconnect us (EPIPE)
         protocol.write(b'abc')
+        # make sure the connection_lost handler isn't called immediately
+        assert protocol.transport is not None
+        # ...but "soon" (in the very next mainloop iteration)
+        await asyncio.sleep(0.01)
         assert protocol.transport is None
         assert isinstance(protocol.exc, BrokenPipeError)
 
@@ -396,7 +404,11 @@ async def test_simple_close(self) -> None:
         assert protocol.transport
         assert protocol.transport.get_write_buffer_size() == 0
         protocol.transport.close()
-        assert protocol.transport is None  # make sure it closed immediately
+        # make sure the connection_lost handler isn't called immediately
+        assert protocol.transport is not None
+        # ...but "soon" (in the very next mainloop iteration)
+        await asyncio.sleep(0.01)
+        assert protocol.transport is None
         # we have another ref on the transport
         transport.close()  # should be idempotent