pybridge: Avoid BrokenPipeError crash of SubprocessTransport.write()

src/cockpit/peer.py

@@ -99,9 +99,11 @@ def _connect_task_done(task: asyncio.Task) -> None:
             #   - other transport exception
             init_message = await self.init_future
 
-        except PeerExited:
-            # This is a fairly generic error.  If the connection process is
-            # still running, perhaps we'd get a better error message from it.
+        except (PeerExited, BrokenPipeError):
+            # These are fairly generic errors. PeerExited means that we observed the process exiting.
+            # BrokenPipeError means that we got EPIPE when attempting to write() to it. In both cases,
+            # the process is gone, but it's not clear why. If the connection process is still running,
+            # perhaps we'd get a better error message from it.
             await connect_task
             # Otherwise, re-raise
             raise

src/cockpit/transports.py

@@ -219,7 +219,13 @@ def _create_write_queue(self, data: bytes) -> None:
         self._protocol.pause_writing()
 
     def write(self, data: bytes) -> None:
-        assert not self._closing
+        # this is a race condition with subprocesses: if we get and process the the "exited"
+        # event before seeing BrokenPipeError, we'll try to write to a closed pipe.
+        # Do what the standard library does and ignore, instead of assert
+        if self._closing:
+            logger.debug('ignoring write() to closing transport fd %i', self._out_fd)
+            return
+
         assert not self._eof
 
         if self._queue is not None:

test/pytest/test_peer.py

@@ -1,13 +1,16 @@
 import asyncio
 import os
 import sys
+import time
 
 import pytest
 
+from cockpit.channel import ChannelError
 from cockpit.packages import BridgeConfig
 from cockpit.peer import ConfiguredPeer, PeerRoutingRule
 from cockpit.protocol import CockpitProtocolError
 from cockpit.router import Router
+from cockpit.transports import SubprocessTransport
 
 from . import mockpeer
 from .mocktransport import MockTransport
@@ -174,3 +177,34 @@ async def test_await_cancellable_connect_close(monkeypatch, event_loop, bridge):
     while len(asyncio.all_tasks()) > 1:
         await asyncio.sleep(0.1)
     assert peer.was_cancelled
+
+
+@pytest.mark.asyncio
+async def test_spawn_broken_pipe(bridge):
+    class BrokenPipePeer(ConfiguredPeer):
+        def __init__(self, *, specific_error=False):
+            super().__init__(bridge, PEER_CONFIG)
+            self.specific_error = specific_error
+
+        async def do_connect_transport(self) -> None:
+            transport = await self.spawn(['sh', '-c', 'exit 9'], ())
+            assert isinstance(transport, SubprocessTransport)
+            time.sleep(0.1)  # sync! increase chance for process to end already
+            transport.write(b'abcdefg\n')
+            while transport.get_returncode() is None:
+                await asyncio.sleep(0.1)
+            if self.specific_error:
+                raise ChannelError('not-supported', message='kaputt')
+
+    # BrokenPipe bubbles up without an error returned by do_connect_transport
+    peer = BrokenPipePeer(specific_error=False)
+    with pytest.raises(BrokenPipeError):
+        await peer.start()
+    peer.close()
+
+    # BrokenPipe gets trumped by specific error returned by do_connect_transport
+    peer = BrokenPipePeer(specific_error=True)
+    with pytest.raises(ChannelError) as raises:
+        await peer.start()
+    assert raises.value.kwargs == {'message': 'kaputt', 'problem': 'not-supported'}
+    peer.close()

test/verify/check-client

@@ -20,7 +20,7 @@
 import testlib
 
 
-@testlib.skipImage("needs pybridge", "debian-stable", "ubuntu-2204", "rhel-8*", "centos-8*")
+@testlib.skipImage("needs pybridge", "rhel-8*", "centos-8*")
 # enable this once our cockpit/ws container can beiboot
 @testlib.skipOstree("client setup does not work with ws container")
 class TestClient(testlib.MachineCase):