From a26cfa38037f07a259e40a5dff6dc1dda3cb318a Mon Sep 17 00:00:00 2001
From: Derek <derek@hyperi.io>
Date: Tue, 3 Nov 2026 09:45:57 +1100
Subject: [PATCH] fix(runner): read results from a file sink

About 1 run in 1,400 returned one event for a two-event input, with exit 0 and empty stderr. This failed test_runner_processes_logs_through_the_real_binary on main.

The console sink writes through Tokio's stdout, which queues each write on a blocking thread and never flushes. Vector exits when stdin ends, and a write that has not started by then is dropped.

The file sink flushes and syncs before finishing. The runner reads results from the file after Vector exits. On two pinned cores the old sink lost an event in 19 of 27,000 runs and the file sink in 0 of 24,000.
---
 .../src/vector_vrl/vector_test_utils.py       | 34 ++++++++++++++-----
 .../integration/test_vector_test_runner.py    | 17 ++++++++++
 2 files changed, 42 insertions(+), 9 deletions(-)

diff --git a/vector-vrl/src/vector_vrl/vector_test_utils.py b/vector-vrl/src/vector_vrl/vector_test_utils.py
index cb8e7e64..373b8d61 100644
--- a/vector-vrl/src/vector_vrl/vector_test_utils.py
+++ b/vector-vrl/src/vector_vrl/vector_test_utils.py
@@ -99,9 +99,10 @@ def run_remap_chain(
     ) -> tuple[bool, list[dict], str]:
         """Run `input_logs` through `transforms`, in order, in a real Vector.
 
-        Vector reads the logs from stdin and exits once every event is written,
-        so the run ends when the work does. `timeout` bounds it: a run still
-        going then is killed and reported as a failure.
+        Vector reads the logs from stdin and writes the results to a file, then
+        exits once the file is flushed, so the run ends when the work does.
+        `timeout` bounds it: a run still going then is killed and reported as a
+        failure.
 
         Each transform keeps its own options (`drop_on_error`, `drop_on_abort`
         and so on), and only its `inputs` is rewired to the one before it.
@@ -117,9 +118,10 @@ def run_remap_chain(
         try:
             data_dir = test_dir / "vector_data"
             data_dir.mkdir()
+            output_file = test_dir / "output.ndjson"
             config_file = test_dir / "config.json"
             config_file.write_text(
-                json.dumps(self._vector_config(transforms, data_dir)),
+                json.dumps(self._vector_config(transforms, data_dir, output_file)),
                 encoding="utf-8",
                 newline="\n",
             )
@@ -146,7 +148,13 @@ def run_remap_chain(
 
             results: list[dict] = []
             undecodable: list[int] = []
-            for number, line in enumerate(completed.stdout.splitlines(), start=1):
+            # The file sink creates the file on its first event, so none means no output.
+            output = (
+                output_file.read_text(encoding="utf-8", errors="replace")
+                if output_file.exists()
+                else ""
+            )
+            for number, line in enumerate(output.splitlines(), start=1):
                 if not line.strip():
                     continue
                 try:
@@ -172,9 +180,16 @@ def run_remap_chain(
 
     @staticmethod
     def _vector_config(
-        transforms: list[tuple[str, dict[str, Any]]], data_dir: Path
+        transforms: list[tuple[str, dict[str, Any]]],
+        data_dir: Path,
+        output_file: Path,
     ) -> dict[str, Any]:
-        """A stdin -> transforms -> stdout Vector config, transforms chained in order."""
+        """A stdin -> transforms -> file Vector config, transforms chained in order.
+
+        The sink is a file because the console sink hands its last write to Tokio's
+        blocking pool without a flush, and Vector's exit drops that write when it
+        has not started: one event of a short run goes missing now and then.
+        """
         chained: dict[str, Any] = {}
         upstream = "runner_stdin"
         for name, transform in transforms:
@@ -185,8 +200,9 @@ def _vector_config(
             "sources": {"runner_stdin": {"type": "stdin"}},
             "transforms": chained,
             "sinks": {
-                "runner_stdout": {
-                    "type": "console",
+                "runner_output": {
+                    "type": "file",
+                    "path": str(output_file),
                     "inputs": [upstream],
                     "encoding": {"codec": "json"},
                 }
diff --git a/vector-vrl/tests/integration/test_vector_test_runner.py b/vector-vrl/tests/integration/test_vector_test_runner.py
index 694f5c12..1a1e9868 100644
--- a/vector-vrl/tests/integration/test_vector_test_runner.py
+++ b/vector-vrl/tests/integration/test_vector_test_runner.py
@@ -27,6 +27,23 @@ def test_runner_processes_logs_through_the_real_binary(vector_binary):
     assert all(result["flag"] is True for result in results)
 
 
+def test_runner_reads_results_from_a_file_sink_not_the_console(tmp_path):
+    """The console sink's last write is lost when Vector exits first.
+
+    About 1 run in 1,500 of a two-event input came back with one event, exit 0 and
+    empty stderr, so no repeat-run test catches it reliably. The file sink flushes
+    and syncs before it finishes.
+    """
+    output_file = tmp_path / "out.ndjson"
+    config = VectorTestRunner._vector_config(  # noqa: SLF001
+        [("t", {"type": "remap", "source": "."})], tmp_path, output_file
+    )
+
+    (sink,) = config["sinks"].values()
+    assert sink["type"] == "file"
+    assert sink["path"] == str(output_file)
+
+
 def test_runner_reports_a_config_that_vector_rejects(vector_binary):
     runner = VectorTestRunner(vector_binary)