Make stub server fail hard on OS errors

boltstub/__init__.py

@@ -235,16 +235,16 @@ def play(self):
                     # `try_skip_to_end` being called from the interrupt handler
                     # in `__main__.py` which spawns a new thread. Without this
                     # `sleep`, the main thread that keeps calling
-                    # `script.consume` only releases Script's internal lock  so
-                    # briefly that `try_skip_to_end` hangs unnecessarily long
+                    # `script.consume` only releases Script's internal lock so
+                    # briefly that `try_skip_to_end` can't reliably acquire it
+                    # thus hanging unnecessarily long.
                     time.sleep(0.000001)
                     continue
         except OSError as e:
-            # It's likely the client has gone away, so we can
-            # safely drop out and silence the error. There's no
-            # point in flagging a broken client from a test helper.
             self.log("S: <BROKEN> %r", e)
-            return
+            self.script.try_skip_to_end(self.channel)
+            if not self.script.done(self.channel):
+                raise
         self.log("Script finished")
 
     def try_skip_to_end(self):

boltstub/parsing.py

@@ -1213,14 +1213,10 @@ def has_deterministic_end(self) -> bool:
         return self.blocks[-1].has_deterministic_end()
 
     def init(self, channel):
-        while True:
+        while self.index < len(self.blocks):
             block = self.blocks[self.index]
             block.init(channel)
-            if (
-                not block.has_deterministic_end()
-                or not block.done(channel)
-                or self.index + 1 >= len(self.blocks)
-            ):
+            if not block.has_deterministic_end() or not block.done(channel):
                 break
             self.index += 1
 

boltstub/tests/test_integration.py

@@ -478,7 +478,7 @@ def test_initial_response(server_version, response_tag, response_name,
     con.read(4)
     res = con.read_message()
     assert res[:2] == b"\xb0" + response_tag
-    with pytest.raises(socket.timeout):
+    with pytest.raises(BrokenSocket):
         con.read(1)
     assert not server.service.exceptions
 
@@ -496,10 +496,17 @@ def test_restarting(server_factory, restarting, concurrent,
         "!: ALLOW CONCURRENT\n" if concurrent else "",
     )
 
-    server = server_factory(parse(script))
+    if restarting and concurrent:
+        with pytest.warns(
+            Warning, match="concurrent scripts are implicitly restarting"
+        ):
+            server = server_factory(parse(script))
+    else:
+        server = server_factory(parse(script))
+
     for i in range(3):
         if i > 0 and not (restarting or concurrent):
-            with pytest.raises((ConnectionError, OSError, BrokenSocket)):
+            with pytest.raises((OSError, BrokenSocket)):
                 con = connection_factory("localhost", 7687)
                 con.write(b"\x60\x60\xb0\x17")
                 con.write(server_version_to_version_request((4, 3)))

tests/stub/configuration_hints/test_connection_recv_timeout_seconds.py

@@ -7,7 +7,10 @@
     get_driver_name,
     TestkitTestCase,
 )
-from tests.stub.shared import StubServer
+from tests.stub.shared import (
+    StubServer,
+    StubServerUncleanExitError,
+)
 
 
 class TestDirectConnectionRecvTimeout(TestkitTestCase):
@@ -114,7 +117,8 @@ def test_timeout_unmanaged_tx(self):
         res = tx.run("in time")
         res.next()
         tx.commit()
-        self._server.done()
+        with self.assertRaises(StubServerUncleanExitError):
+            self._server.done()
         self._assert_is_timeout_exception(exc.exception)
         self.assertEqual(self._server.count_responses("<ACCEPT>"), 2)
         self.assertEqual(self._server.count_responses("<HANGUP>"), 2)
@@ -143,7 +147,8 @@ def test_timeout_unmanaged_tx_should_fail_subsequent_usage_after_timeout(
         if get_driver_name() in ["java", "ruby"]:
             tx.rollback()
 
-        self._server.done()
+        with self.assertRaises(StubServerUncleanExitError):
+            self._server.done()
         self._assert_is_timeout_exception(first_run_error.exception)
         self._assert_is_client_exception(second_run_error.exception)
         self.assertEqual(self._server.count_responses("<ACCEPT>"), 1)
@@ -177,7 +182,8 @@ def work(tx):
 
         self._start_server("1_second_exceeds_tx_retry.script")
         self._session.execute_write(work)
-        self._server.done()
+        with self.assertRaises(StubServerUncleanExitError):
+            self._server.done()
         self.assertEqual(retries, 2)
         self.assertIsInstance(record, types.Record)
         self.assertEqual(record.values, [types.CypherInt(1)])

tests/stub/routing/scripts/v3_no_routing/optional_hello.script

@@ -1,7 +1,8 @@
 !: BOLT #VERSION#
 
-?: GOODBYE
-C: HELLO {"scheme": "basic", "credentials": "c", "principal": "p", "user_agent": "#USER_AGENT#" #ROUTING#}
-S: SUCCESS {"server": "#SERVER_AGENT#", "connection_id": "bolt-123456789"}
+{?
+    C: HELLO {"scheme": "basic", "credentials": "c", "principal": "p", "user_agent": "#USER_AGENT#" #ROUTING#}
+    S: SUCCESS {"server": "#SERVER_AGENT#", "connection_id": "bolt-123456789"}
+?}
 *: RESET
 ?: GOODBYE

tests/stub/routing/scripts/v4x1_no_routing/optional_hello.script

@@ -1,7 +1,8 @@
 !: BOLT #VERSION#
 
-?: GOODBYE
-C: HELLO {"scheme": "basic", "credentials": "c", "principal": "p", "user_agent": "#USER_AGENT#" #ROUTING#}
-S: SUCCESS {"server": "#SERVER_AGENT#", "connection_id": "bolt-123456789"}
+{?
+    C: HELLO {"scheme": "basic", "credentials": "c", "principal": "p", "user_agent": "#USER_AGENT#" #ROUTING#}
+    S: SUCCESS {"server": "#SERVER_AGENT#", "connection_id": "bolt-123456789"}
+?}
 *: RESET
 ?: GOODBYE