Make a COTTAS query killable, and make a timeout report as one

Dockerfile

@@ -5,6 +5,11 @@ ARG COMUNICA_VERSION=5.3.0
 # behind it. It provides native SPARQL over a .hdt artifact, so validation can
 # query the compressed representation directly instead of decoding it first.
 ARG COMUNICA_HDT_VERSION=5.0.1
+#: Native COTTAS querying. Replaces the pycottas/rdflib path, which pulled
+#: triples through the Store API in Python and could not answer a genotype
+#: query at benchmark scale. Ships a DuckDB addon: glibc base only, and the
+#: prebuild is per-architecture.
+ARG COMUNICA_COTTAS_VERSION=0.1.0
 # QLever is an optional second SPARQL engine for validation. Its binaries are
 # copied from the upstream published image rather than built here: compiling
 # QLever needs a large C++ toolchain and would dominate this image's build.
@@ -175,6 +180,7 @@ RUN python3 -m venv /opt/pycottas-venv \
     pyshacl==0.30.1
 
 ARG COMUNICA_HDT_VERSION
+ARG COMUNICA_COTTAS_VERSION
 
 # The HDT engine compiles native bindings, so a toolchain is needed at install
 # time but not afterwards; it is purged in the same layer to keep it out of the
@@ -184,9 +190,31 @@ RUN apt-get update \
   && npm install --global \
     "@comunica/query-sparql-file@${COMUNICA_VERSION}" \
     "@comunica/query-sparql-hdt@${COMUNICA_HDT_VERSION}" \
+    "@elias.crum/query-sparql-cottas@${COMUNICA_COTTAS_VERSION}" \
   && apt-get purge -y --auto-remove build-essential \
   && rm -rf /var/lib/apt/lists/*
 
+# A DuckDB addon that cannot load is a runtime failure inside a benchmark cell
+# hours later, so prove the whole path here: write a triple, build a .cottas
+# with pycottas, and answer a query over it with the comunica engine.
+#
+# Checking the binary alone is not enough. `--version` prints its table and
+# then exits 1, and the addon is ESM so `require()` cannot probe it either --
+# both look like failures when the engine is fine. Only a real query settles it,
+# and it also covers rdf2cottas, which is the other half of this engine.
+RUN set -eu; \
+    printf '<http://example.org/s> <http://example.org/p> <http://example.org/o> .\n' \
+      > /tmp/cottas-probe.nt; \
+    /opt/pycottas-venv/bin/python -c \
+      "import pycottas; pycottas.rdf2cottas('/tmp/cottas-probe.nt', '/tmp/cottas-probe.cottas')"; \
+    out="$(comunica-sparql-cottas cottas@/tmp/cottas-probe.cottas \
+            'SELECT ?s WHERE { ?s ?p ?o }' 2>&1)"; \
+    printf '%s' "$out" | grep -q 'example.org/s' \
+      || { echo "cottas engine built the artifact but cannot query it:"; \
+           printf '%s\n' "$out"; exit 1; }; \
+    rm -f /tmp/cottas-probe.nt /tmp/cottas-probe.cottas; \
+    echo "cottas engine verified end to end"
+
 RUN mkdir -p /opt/rmlstreamer \
   && curl -fsSL \
     -o /opt/rmlstreamer/RMLStreamer-v${RMLSTREAMER_VERSION}-standalone.jar \

src/validation/validation_runner.py

@@ -1971,6 +1971,7 @@ def validate_ntriples(source: Path, results_dir: Path) -> dict[str, Any]:
 #: The HDT endpoint runs alongside the N-Triples one within a single run, so
 #: they must not contend for a port.
 DEFAULT_HDT_ENDPOINT_PORT = 7021
+DEFAULT_COTTAS_ENDPOINT_PORT = 7022
 
 
 QLEVER_STATUS_FILE = Path("/opt/vcf-rdfizer/qlever-status.txt")
@@ -2118,6 +2119,25 @@ def run_query_process(
             return 124, f"query exceeded {timeout}s"
 
 
+def is_timeout_error(error: BaseException) -> bool:
+    """Whether an exception from an HTTP query is a timeout rather than a fault.
+
+    urlopen(timeout=...) surfaces a timeout either directly or wrapped in a
+    URLError, and the two are not interchangeable at the call site. Telling them
+    apart is what lets a timeout be reported as exit 124 instead of as a generic
+    execution failure -- which is the difference between "this query is too slow
+    for this engine" and "this engine is broken".
+    """
+    import socket
+    import urllib.error
+
+    if isinstance(error, (TimeoutError, socket.timeout)):
+        return True
+    if isinstance(error, urllib.error.URLError):
+        return isinstance(error.reason, (TimeoutError, socket.timeout))
+    return False
+
+
 def _terminate_process_group(process: subprocess.Popen) -> None:
     """Signal a process group, escalating to SIGKILL, and reap the leader."""
     for signal_number, grace in ((signal.SIGTERM, 10), (signal.SIGKILL, 5)):
@@ -2385,12 +2405,19 @@ def execute(self, query_id: str, query_path: Path) -> dict[str, Any]:
                 self.endpoint, query_path.read_text(encoding="utf-8"),
                 timeout=self.query_timeout,
             )
-        except Exception as error:  # noqa: BLE001 - reported as EXECUTION_FAILED
+        except Exception as error:  # noqa: BLE001 - reported below
             stderr_path.write_text(str(error), encoding="utf-8")
             raw_path.write_bytes(b"")
+            # 124 is the timeout convention the query loop keys on; anything
+            # else here really is an execution failure.
+            timed_out = is_timeout_error(error)
+            message = (
+                f"query exceeded {self.query_timeout}s" if timed_out else str(error)
+            )
             return self._envelope(
-                query_id, query_path, returncode=1, started=started,
-                raw_path=raw_path, stderr_path=stderr_path, error=str(error),
+                query_id, query_path, returncode=124 if timed_out else 1,
+                started=started, raw_path=raw_path, stderr_path=stderr_path,
+                error=message,
             )
         raw_path.write_bytes(payload)
         stderr_path.write_bytes(b"")
@@ -2618,12 +2645,17 @@ def execute(self, query_id: str, query_path: Path) -> dict[str, Any]:
             payload = self._post(
                 query_path.read_text(encoding="utf-8"), timeout=self.query_timeout
             )
-        except Exception as error:  # noqa: BLE001 - reported as EXECUTION_FAILED
+        except Exception as error:  # noqa: BLE001 - reported below
             stderr_path.write_text(str(error), encoding="utf-8")
             raw_path.write_bytes(b"")
+            timed_out = is_timeout_error(error)
+            message = (
+                f"query exceeded {self.query_timeout}s" if timed_out else str(error)
+            )
             return self._envelope(
-                query_id, query_path, returncode=1, started=started,
-                raw_path=raw_path, stderr_path=stderr_path, error=str(error),
+                query_id, query_path, returncode=124 if timed_out else 1,
+                started=started, raw_path=raw_path, stderr_path=stderr_path,
+                error=message,
             )
         raw_path.write_bytes(payload)
         stderr_path.write_bytes(b"")
@@ -2754,63 +2786,67 @@ def describe(self) -> dict[str, Any]:
         }
 
 
-class CottasEngine(NativeArtifactEngine):
-    """Query a .cottas artifact in place through pycottas's rdflib store.
+class CottasEngine(ComunicaHttpEndpointMixin, NativeArtifactEngine):
+    """Query a .cottas artifact in place, from one load, over Comunica's endpoint.
+
+    This used to go through pycottas's rdflib Store in-process. rdflib evaluates
+    SPARQL over a custom Store by pulling triples through the Store API in
+    Python, which is why that path was unusable rather than merely slow:
+    measured on a 39 MB artifact, opening the store took 3-15 ms but
+    `SELECT ?s ?p ?o LIMIT 1` took 12.7 s and COUNT(*) did not finish in two
+    minutes. 06_equivalence hung on one genotype query for 41 hours, and again
+    for 10 hours once bounded, because a bound stops a query without making the
+    engine able to answer it.
+
+    @elias.crum/query-sparql-cottas answers the same artifact through DuckDB
+    over the Parquet, behind the same long-lived endpoint the comunica and HDT
+    engines already use -- so the graph is opened once and every query is
+    charged only for itself.
 
-    pycottas exposes ``COTTASStore``, an rdflib Store backed by the Parquet
-    artifact, so this runs in-process rather than shelling out. The validator
-    already runs under the interpreter that owns pycottas.
+    Building the artifact still goes through pycottas: this replaces querying,
+    not rdf2cottas.
     """
 
     name = "cottas"
     artifact_format = "cottas"
+    endpoint_binary = "comunica-sparql-cottas-http"
+    endpoint_port_option = "cottas_port"
+    default_endpoint_port = DEFAULT_COTTAS_ENDPOINT_PORT
+    endpoint_missing_hint = (
+        "so COTTAS cannot be queried natively. Rebuild the image, or validate "
+        "the COTTAS artifact by decoding it (--rdf file.cottas --engine comunica)"
+    )
+
+    def __init__(self, source: Path, *, raw_dir: Path, scratch: Path, options: dict[str, Any]):
+        super().__init__(source, raw_dir=raw_dir, scratch=scratch, options=options)
+        self._init_endpoint(options)
 
     def build_artifact(self, target: Path) -> None:
         import pycottas
 
         pycottas.rdf2cottas(str(self.source), str(target))
 
-    def start(self) -> None:
-        try:
-            import pycottas  # noqa: F401
-            import rdflib  # noqa: F401
-        except ImportError as error:
-            raise RuntimeError(
-                f"pycottas and rdflib are required to query COTTAS natively: {error}"
-            ) from error
-        super().start()
-
-        import pycottas
-        import rdflib
-
-        self.graph = rdflib.Graph(store=pycottas.COTTASStore(str(self.artifact)))
+    def _endpoint_source_argument(self) -> str:
+        # Comunica needs the source type declared, exactly as for HDT: a bare
+        # path is treated as a link to dereference. The engine takes one local
+        # file -- no URLs, directories or globs.
+        return f"cottas@{self.artifact}"
 
-    def execute(self, query_id: str, query_path: Path) -> dict[str, Any]:
-        raw_path = self.raw_dir / f"{query_id}.sparql.json"
-        stderr_path = self.raw_dir / f"{query_id}.stderr.txt"
-        started = time.monotonic()
-        try:
-            result = self.graph.query(query_path.read_text(encoding="utf-8"))
-            raw_path.write_bytes(result.serialize(format="json"))
-            stderr_path.write_bytes(b"")
-            returncode, error = 0, None
-        except Exception as failure:  # noqa: BLE001 - reported as EXECUTION_FAILED
-            raw_path.write_text("{}", encoding="utf-8")
-            stderr_path.write_text(str(failure), encoding="utf-8")
-            returncode, error = 1, str(failure)
-        return self._envelope(
-            query_id, query_path, returncode=returncode, started=started,
-            raw_path=raw_path, stderr_path=stderr_path, error=error,
-        )
+    def start(self) -> None:
+        # Check the binary before building the artifact: rdf2cottas is the
+        # expensive step and there is no point paying it for an engine that
+        # cannot run.
+        self._require_endpoint_binary()
+        NativeArtifactEngine.start(self)
+        self._start_endpoint()
 
-    def stop(self) -> None:
-        graph = getattr(self, "graph", None)
-        if graph is not None:
-            try:
-                graph.close()
-            except Exception:  # noqa: BLE001 - teardown must not mask a result
-                pass
-            self.graph = None
+    def describe(self) -> dict[str, Any]:
+        return {
+            **NativeArtifactEngine.describe(self),
+            **self.endpoint_describe(),
+            "mode": "native cottas query over DuckDB, no decode",
+            "index_orders": "spo only (sibling .posg/.ospg files are not built)",
+        }
 
 
 ENGINE_CLASSES = {

test/test_validation_benchmark_unit.py

@@ -227,27 +227,46 @@ def test_hdt_without_the_native_binary_explains_the_decode_alternative(self):
                     engine.start()
         self.assertIn("--engine comunica", str(raised.exception))
 
-    def test_a_failing_cottas_query_is_reported_not_raised(self):
-        """One bad query must not abort the other engines in a benchmark run."""
+    def test_cottas_queries_are_addressed_with_comunicas_typed_source_prefix(self):
+        """Same requirement as HDT: a bare path is treated as a link."""
         with tempfile.TemporaryDirectory() as td:
             tmp_path = Path(td)
-            raw_dir = tmp_path / "raw"
-            raw_dir.mkdir()
-            query = tmp_path / "q.rq"
-            query.write_text("SELECT * WHERE { ?s ?p ?o }", encoding="utf-8")
+            (tmp_path / "raw").mkdir()
             engine = self._engine("cottas", tmp_path)
-            engine.graph = mock.Mock()
-            engine.graph.query.side_effect = RuntimeError("parquet is unreadable")
-            envelope = engine.execute("q01", query)
-
-            self.assertEqual(envelope["status"], "EXECUTION_FAILED")
-            self.assertIn("parquet is unreadable", envelope["error"])
-            self.assertEqual(json.loads(Path(envelope["rawResult"]).read_text()), {})
-            self.assertIn(
-                "parquet is unreadable",
-                Path(envelope["stderr"]).read_text(encoding="utf-8"),
+            engine.artifact = tmp_path / "cohort.cottas"
+            self.assertEqual(
+                engine._endpoint_source_argument(),
+                f"cottas@{tmp_path / 'cohort.cottas'}",
             )
 
+    def test_cottas_is_queried_over_an_endpoint_not_in_process(self):
+        """The in-process rdflib Store is what made this engine unusable.
+
+        rdflib evaluates SPARQL over a custom Store by pulling triples through
+        the Store API in Python: 12.7 s for LIMIT 1 on a 39 MB artifact, and a
+        genotype query that ran 41 hours without finishing. Querying over the
+        same long-lived endpoint the comunica and HDT engines use is what makes
+        06_equivalence possible at all, so it must not regress to in-process.
+        """
+        source = Path(V.__file__).read_text(encoding="utf-8")
+        cottas = source[source.index("class CottasEngine"):]
+        cottas = cottas[: cottas.index("\nENGINE_CLASSES")]
+        self.assertIn("ComunicaHttpEndpointMixin", cottas)
+        self.assertIn("comunica-sparql-cottas-http", cottas)
+        self.assertNotIn("self.graph.query", cottas)
+        self.assertNotIn("COTTASStore", cottas)
+        # Building the artifact is still pycottas' job; only querying moved.
+        self.assertIn("rdf2cottas", cottas)
+
+    def test_cottas_checks_the_binary_before_building_the_artifact(self):
+        """rdf2cottas is the expensive step; do not pay it for a dead engine."""
+        with tempfile.TemporaryDirectory() as td:
+            engine = self._engine("cottas", Path(td))
+            with mock.patch.object(V.shutil, "which", return_value=None):
+                with self.assertRaises(RuntimeError) as raised:
+                    engine.start()
+        self.assertIn("--engine comunica", str(raised.exception))
+
     def test_an_unknown_engine_name_names_the_supported_ones(self):
         with tempfile.TemporaryDirectory() as td:
             with self.assertRaises(ValueError) as raised:

test/test_validation_engines_unit.py

@@ -10,6 +10,7 @@
 import json
 import os
 import subprocess
+import sys
 import tempfile
 import time
 import unittest
@@ -877,3 +878,82 @@ def test_position_datatype_preflight_accepts_the_xsd_integer_family(self):
             self.assertIn(accepted, accepted_list)
         for rejected in ("xsd:string", "xsd:decimal", "xsd:double"):
             self.assertNotIn(rejected, accepted_list)
+
+
+class QueryTimeoutEnforcementTests(VerboseTestCase):
+    """A query that will not finish must be stopped and reported as a timeout.
+
+    Regression: 06_equivalence hung twice on the same cottas query -- 41 h, then
+    10 h with --validation-query-timeout 1800 and --validation-time-budget 14400
+    on the command line. Neither bound could fire. CottasEngine queried
+    in-process through rdflib, so nothing could interrupt it, and no engine ever
+    produced exit 124, so the loop's timeout handling was unreachable in any
+    case. run_query_process -- which does the right thing -- existed and had no
+    callers at all.
+    """
+
+    def test_run_query_process_reports_124_and_kills_the_tree(self):
+        with tempfile.TemporaryDirectory() as td:
+            tmp = Path(td)
+            out, err = tmp / "out.json", tmp / "err.txt"
+            marker = tmp / "child_survived"
+            # A parent that forks a child outliving it, exactly the shape the
+            # helper exists for: killing the leader must not leave the child.
+            source = (
+                "import subprocess, sys, time\n"
+                "subprocess.Popen([sys.executable, '-c',"
+                " \"import time,pathlib;time.sleep(8);"
+                f"pathlib.Path(r'{marker}').write_text('x')\"])\n"
+                "time.sleep(60)\n"
+            )
+            started = time.monotonic()
+            code, error = V.run_query_process(
+                [sys.executable, "-c", source],
+                stdout_path=out, stderr_path=err, timeout=2,
+            )
+            elapsed = time.monotonic() - started
+            self.assertEqual(code, 124)
+            self.assertIn("exceeded", error or "")
+            self.assertLess(elapsed, 30, "the timeout did not actually interrupt")
+            time.sleep(9)
+            self.assertFalse(
+                marker.exists(),
+                "the forked child outlived the group kill and kept running",
+            )
+
+    def test_cottas_is_not_queried_in_process(self):
+        """In-process querying is what made the hang unkillable.
+
+        It is now answered over comunica's endpoint, where the mixin's timeout
+        applies like any other HTTP engine -- so this engine inherits the 124
+        classification rather than needing its own kill path.
+        """
+        source = Path(V.__file__).read_text(encoding="utf-8")
+        cottas = source[source.index("class CottasEngine"):]
+        cottas = cottas[: cottas.index("\nENGINE_CLASSES")]
+        self.assertIn("ComunicaHttpEndpointMixin", cottas)
+        self.assertNotIn(
+            "self.graph.query", cottas,
+            "CottasEngine is querying in-process again; a hung query cannot be "
+            "stopped that way, which is the bug this replaced",
+        )
+
+    def test_http_timeouts_are_classified_as_timeouts(self):
+        """Exit 1 for a timeout makes it a crash; the loop keys on 124."""
+        import socket
+        import urllib.error
+        self.assertTrue(V.is_timeout_error(TimeoutError()))
+        self.assertTrue(V.is_timeout_error(socket.timeout()))
+        self.assertTrue(
+            V.is_timeout_error(urllib.error.URLError(socket.timeout()))
+        )
+        self.assertFalse(V.is_timeout_error(ValueError("bad query")))
+        self.assertFalse(
+            V.is_timeout_error(urllib.error.URLError(ConnectionRefusedError()))
+        )
+
+    def test_the_timeout_convention_has_a_producer(self):
+        """124 is what the query loop keys on, so something must emit it."""
+        source = Path(V.__file__).read_text(encoding="utf-8")
+        self.assertIn("return 124", source)
+        self.assertIn("returncode=124", source)