diff --git a/test_runner/fixtures/broker.py b/test_runner/fixtures/broker.py
index fa8b816e69..8aca90a097 100644
--- a/test_runner/fixtures/broker.py
+++ b/test_runner/fixtures/broker.py
@@ -54,7 +54,10 @@ class NeonBroker:
             else:
                 break  # success
 
-    def stop(self):
+    def stop(self, immediate: bool = False):
         if self.handle is not None:
-            self.handle.terminate()
+            if immediate:
+                self.handle.kill()
+            else:
+                self.handle.terminate()
             self.handle.wait()
diff --git a/test_runner/fixtures/neon_fixtures.py b/test_runner/fixtures/neon_fixtures.py
index 111070bd93..c139fa215c 100644
--- a/test_runner/fixtures/neon_fixtures.py
+++ b/test_runner/fixtures/neon_fixtures.py
@@ -754,20 +754,12 @@ class NeonEnvBuilder:
         # Stop all the nodes.
         if self.env:
             log.info("Cleaning up all storage and compute nodes")
-            self.env.endpoints.stop_all()
-            for sk in self.env.safekeepers:
-                sk.stop(immediate=True)
-
-            for pageserver in self.env.pageservers:
+            self.env.stop(
+                immediate=True,
                 # if the test threw an exception, don't check for errors
                 # as a failing assertion would cause the cleanup below to fail
-                if exc_type is not None:
-                    pageserver.assert_no_metric_errors()
-
-                pageserver.stop(immediate=True)
-
-            self.env.attachment_service.stop(immediate=True)
-
+                ps_assert_metric_no_errors=(exc_type is None),
+            )
             cleanup_error = None
 
             if self.scrub_on_exit:
@@ -950,6 +942,20 @@ class NeonEnv:
         for safekeeper in self.safekeepers:
             safekeeper.start()
 
+    def stop(self, immediate=False, ps_assert_metric_no_errors=False):
+        """
+        After this method returns, there should be no child processes running.
+        """
+        self.endpoints.stop_all()
+        for sk in self.safekeepers:
+            sk.stop(immediate=immediate)
+        for pageserver in self.pageservers:
+            if ps_assert_metric_no_errors:
+                pageserver.assert_no_metric_errors()
+            pageserver.stop(immediate=immediate)
+        self.attachment_service.stop(immediate=immediate)
+        self.broker.stop(immediate=immediate)
+
     @property
     def pageserver(self) -> NeonPageserver:
         """