Retry the comunica warm-up probe instead of failing on a refused port

The cross-engine agreement job failed on a commit that changed only a
docstring, and its message named a timeout that had not happened:

  comunica could not read /work/.../expanded.nt within 3600s
  (raise --comunica-warmup-timeout, or use --engine qlever):
  <urlopen error [Errno 111] Connection refused>

It had waited milliseconds. `timeout=` on urlopen is the socket timeout
-- how long to wait for a response -- and a refused connection comes
back from the kernel at once, so the budget never applied. `_await_bind`
already loops and tolerates refusal; `_await_warm` made exactly one
attempt. That asymmetry was the bug: comunica answers the bind probe and
then hands the socket to its worker, so the next connection can be
refused for a moment while the process is perfectly healthy.

The probe now retries until the budget is genuinely spent, and the
reported duration is measured rather than read back from the flag.

Retrying must not become tolerance for a dead engine, so two conditions
still fail immediately: the server process exiting, which keeps its exit
code and log tail, and the endpoint *answering* with an error. An
HTTPError means comunica replied -- a source it cannot parse would
answer every probe identically, and retrying it would spin for the whole
budget before reporting what it knew on the first attempt.

This gates merges, and it is timing-dependent rather than rare: the
image floats its base and its apt nodejs and python3, so a rebuild can
shift the timing that exposes it. `HdtEngine` shares the mixin, so it
was exposed to the same failure.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: ecrum19

docs/roadmap.md

@@ -33,7 +33,7 @@ a record with `ALT=.` satisfies both the shape and `vcfc:missingValuePolicy`.
 
 ### ~~1. The validation suite has not been migrated~~ — **done**
 
-The suite passes: 703 tests, mutation score 96/113 (85%) across 60 mutations.
+The suite passes: 707 tests, mutation score 96/113 (85%) across 60 mutations.
 The runner and the wrapper now share one vocabulary module instead of mirroring
 each other, and the census derives every new resource family from the VCF. Ten
 of the eighteen new mutations are recorded gaps, each with what would close it.

docs/validation-migration-notes.md

@@ -7,7 +7,7 @@ and follows each input's declared VCF version. The validation suite has been
 migrated with it. This file records what changed, what it found, and what is
 still open.
 
-**Status: the suite passes.** 703 tests, no failures. The mutation score is
+**Status: the suite passes.** 707 tests, no failures. The mutation score is
 96/113 (85%) across 60 mutations, each listed with its expected detector.
 
 ---

docs/validation.md

@@ -97,6 +97,29 @@ vcf-rdfizer --mode full -i cohort.vcf.gz --validate \
   --validation-query-timeout 14400 --validation-time-budget 86400 -o ./out
 ```
 
+### A startup probe that fails on a healthy engine
+
+Comunica's endpoint is probed twice at startup: `_await_bind` waits for the
+port to answer at all, then `_await_warm` proves the endpoint can actually read
+the source, so that readiness is charged to setup rather than billed to the
+first query.
+
+The warm-up probe used to be a single request. Comunica answers the bind probe
+and then hands the socket to its worker, so the next connection can be refused
+for a moment while the process is perfectly healthy — and a refused connection
+comes back from the kernel immediately, not after the warm-up budget. The result
+was an engine that failed in milliseconds while reporting
+
+```
+comunica could not read /work/…/expanded.nt within 3600s
+```
+
+a timeout it had never waited for. The probe now retries until the budget is
+genuinely spent, still failing at once if the server process exits or if the
+endpoint *answers* with an error, and the duration in the message is measured
+rather than read back from the flag. If you see that message now, the wait was
+real: raise `--comunica-warmup-timeout`, or use `--engine qlever`.
+
 ### A decode that reports success and writes garbage
 
 Every decode into N-Triples is now checked structurally before anything reads

src/validation/validation_runner.py

@@ -2254,27 +2254,72 @@ def _await_bind(self, server_log: Path) -> None:
             f"{self.bind_timeout}s: {last_error}"
         )
 
+    @staticmethod
+    def _is_connection_level_error(error: BaseException) -> bool:
+        """Is this the socket never opening, rather than the server answering?
+
+        An HTTPError means the endpoint replied, so whatever it said is a real
+        result and must not be retried. Everything else at this layer -- a
+        refused, reset or aborted connection, or a socket that timed out -- is
+        about reaching the process, not about what it thinks of the query.
+        """
+        import urllib.error
+
+        if isinstance(error, urllib.error.HTTPError):
+            return False
+        if isinstance(error, urllib.error.URLError):
+            return isinstance(error.reason, (OSError, TimeoutError))
+        return isinstance(error, (ConnectionError, TimeoutError))
+
     def _await_warm(self, server_log: Path) -> None:
         """Prove the endpoint can actually read the source, before the suite runs.
 
         Charged to setup rather than to the first query, so one query is not
         billed for readiness that every later query got for free.
+
+        A refused connection is retried rather than fatal. ``_await_bind``
+        returns as soon as the port answers one probe, but comunica hands the
+        socket to its worker after that reply, so the next connection can be
+        refused for a moment while the process is perfectly healthy. Treating
+        that as failure made the whole engine unusable on a timing accident --
+        and reported it as a timeout that never happened, because a refused
+        connection comes back from the kernel at once rather than after the
+        warm-up budget. Only the server exiting, an answer from the endpoint,
+        or the budget genuinely running out ends this wait.
         """
         assert self.endpoint is not None
         started = time.monotonic()
-        try:
-            self._post(
-                self.endpoint, "SELECT * WHERE { ?s ?p ?o } LIMIT 1",
-                timeout=self.warmup_timeout,
-            )
-        except Exception as error:  # noqa: BLE001 - surfaced as engine failure
+        deadline = started + self.warmup_timeout
+        last_error: BaseException | None = None
+        while True:
             self._assert_alive(server_log)
-            raise RuntimeError(
-                f"{self.name} could not read {self._endpoint_source_argument()} "
-                f"within {self.warmup_timeout}s (raise --comunica-warmup-timeout, "
-                f"or use --engine qlever): {error}"
-            ) from error
-        self.warmup_seconds = time.monotonic() - started
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                break
+            try:
+                self._post(
+                    self.endpoint, "SELECT * WHERE { ?s ?p ?o } LIMIT 1",
+                    timeout=remaining,
+                )
+            except Exception as error:  # noqa: BLE001 - surfaced as engine failure
+                self._assert_alive(server_log)
+                if not self._is_connection_level_error(error):
+                    raise RuntimeError(
+                        f"{self.name} could not read "
+                        f"{self._endpoint_source_argument()}: {error}"
+                    ) from error
+                last_error = error
+                time.sleep(min(0.5, max(0.0, deadline - time.monotonic())))
+            else:
+                self.warmup_seconds = time.monotonic() - started
+                return
+        self._assert_alive(server_log)
+        raise RuntimeError(
+            f"{self.name} could not read {self._endpoint_source_argument()} "
+            f"within {time.monotonic() - started:.0f}s "
+            f"(raise --comunica-warmup-timeout, or use --engine qlever): "
+            f"{last_error or 'the endpoint never accepted a connection'}"
+        )
 
     def _assert_alive(self, server_log: Path) -> None:
         if self.server is not None and self.server.poll() is not None:

test/test_validation_engines_unit.py

@@ -438,6 +438,141 @@ def fake_popen(command, stdout=None, **kwargs):
                     engine.start()
             self.assertIn("index not found", str(caught.exception))
 
+    def _comunica_engine_awaiting_warm(self, tmp_path, *, warmup_timeout):
+        """A started comunica engine, parked just before the warm-up probe."""
+        source = tmp_path / "cohort.nt"
+        source.write_bytes(TRIPLES)
+        engine = V.build_engine(
+            "comunica", source, raw_dir=tmp_path, scratch=tmp_path,
+            options={"comunica_warmup_timeout": warmup_timeout},
+        )
+        engine.endpoint = f"http://127.0.0.1:{engine.port}/sparql"
+        engine.server = mock.MagicMock()
+        engine.server.poll.return_value = None
+        return engine
+
+    def test_a_refused_connection_during_warmup_is_retried(self):
+        """The bind probe is not a promise that the next connection is accepted.
+
+        comunica answers the bind probe and then hands the socket to its
+        worker, so a connection a moment later can be refused while the process
+        is perfectly healthy. That briefly-closed port used to fail the whole
+        engine - and the run it gated - on a timing accident.
+        """
+        import urllib.error
+
+        with tempfile.TemporaryDirectory() as td:
+            tmp_path = Path(td)
+            engine = self._comunica_engine_awaiting_warm(tmp_path, warmup_timeout=30)
+            attempts = []
+
+            def flaky_post(endpoint, query, *, timeout):
+                attempts.append(timeout)
+                if len(attempts) < 3:
+                    raise urllib.error.URLError(ConnectionRefusedError(111, "refused"))
+                return b'{"results": {"bindings": []}}'
+
+            with mock.patch.object(engine, "_post", flaky_post), \
+                    mock.patch.object(V.time, "sleep"):
+                engine._await_warm(tmp_path / "server.log")
+
+            self.assertEqual(len(attempts), 3)
+            self.assertIsNotNone(engine.warmup_seconds)
+            # Each attempt is given what is left of the budget, never the whole
+            # of it again, so retrying cannot outlive --comunica-warmup-timeout.
+            self.assertLessEqual(attempts[-1], attempts[0])
+
+    def test_a_server_that_dies_during_warmup_fails_at_once(self):
+        """Retrying a refused port must not paper over a dead engine."""
+        import urllib.error
+
+        with tempfile.TemporaryDirectory() as td:
+            tmp_path = Path(td)
+            server_log = tmp_path / "server.log"
+            server_log.write_text("comunica: out of memory\n", encoding="utf-8")
+            engine = self._comunica_engine_awaiting_warm(tmp_path, warmup_timeout=3600)
+            engine.server.poll.return_value = 137
+            engine.server.returncode = 137
+
+            def refused(endpoint, query, *, timeout):
+                raise urllib.error.URLError(ConnectionRefusedError(111, "refused"))
+
+            started = time.monotonic()
+            with mock.patch.object(engine, "_post", refused):
+                with self.assertRaises(RuntimeError) as caught:
+                    engine._await_warm(server_log)
+
+            # Fails on the dead process, not after the 3600s budget.
+            self.assertLess(time.monotonic() - started, 30)
+            self.assertIn("137", str(caught.exception))
+            self.assertIn("out of memory", str(caught.exception))
+
+    def test_an_answered_error_is_not_retried(self):
+        """An HTTP reply is the endpoint's verdict, not a connection problem.
+
+        A source comunica cannot parse answers every probe the same way, so
+        retrying it would spin for the whole warm-up budget before reporting
+        what it already knew on the first attempt.
+        """
+        import urllib.error
+
+        with tempfile.TemporaryDirectory() as td:
+            tmp_path = Path(td)
+            engine = self._comunica_engine_awaiting_warm(tmp_path, warmup_timeout=3600)
+            attempts = []
+
+            def http_error(endpoint, query, *, timeout):
+                attempts.append(timeout)
+                raise urllib.error.HTTPError(
+                    endpoint, 500, "Parse error", {}, None
+                )
+
+            with mock.patch.object(engine, "_post", http_error):
+                with self.assertRaises(RuntimeError) as caught:
+                    engine._await_warm(tmp_path / "server.log")
+
+            self.assertEqual(len(attempts), 1)
+            message = str(caught.exception)
+            self.assertIn("could not read", message)
+            # The old message claimed a timeout it had never waited for.
+            self.assertNotIn("3600s", message)
+
+    def test_an_exhausted_warmup_budget_reports_the_time_it_waited(self):
+        """The duration in the message must be measured, not the flag's value.
+
+        The original reported the configured budget whatever had happened, so a
+        connection refused in milliseconds was announced as an hour-long
+        timeout - pointing every reader at the wrong thing entirely.
+        """
+        import itertools
+        import urllib.error
+
+        with tempfile.TemporaryDirectory() as td:
+            tmp_path = Path(td)
+            engine = self._comunica_engine_awaiting_warm(tmp_path, warmup_timeout=30)
+            attempts = []
+
+            def refused(endpoint, query, *, timeout):
+                attempts.append(timeout)
+                raise urllib.error.URLError(ConnectionRefusedError(111, "refused"))
+
+            # A clock that runs 10s per reading, so the budget expires after one
+            # attempt without the test waiting for any of it.
+            clock = itertools.count(0, 10)
+            with mock.patch.object(V.time, "monotonic", lambda: next(clock)), \
+                    mock.patch.object(V.time, "sleep"), \
+                    mock.patch.object(engine, "_post", refused):
+                with self.assertRaises(RuntimeError) as caught:
+                    engine._await_warm(tmp_path / "server.log")
+
+            self.assertEqual(len(attempts), 1)
+            message = str(caught.exception)
+            # 40s elapsed against a 30s budget: the figure is measured, and is
+            # not simply the flag read back out.
+            self.assertIn("within 40s", message)
+            self.assertIn("refused", message)
+            self.assertIn("--comunica-warmup-timeout", message)
+
     def test_qlever_command_lines_are_overridable(self):
         """QLever's CLI varies by release, so both argv are user-overridable."""
         with tempfile.TemporaryDirectory() as td: