diff --git a/src/validation/validation_runner.py b/src/validation/validation_runner.py
index e830f84..3987504 100644
--- a/src/validation/validation_runner.py
+++ b/src/validation/validation_runner.py
@@ -2024,6 +2024,64 @@ def merge_shapes_graph(shapes: list[Path]):
return graph
+#: pyshacl loads the whole graph into memory, so the shape layer needs a size
+#: gate. This one is in triples rather than bytes, because what pyshacl pays
+#: for is the graph, not the packaging -- see the comment at its call site for
+#: the failure that motivated it.
+#:
+#: 50M is chosen to sit above the largest graph the published campaign actually
+#: validated with shapes (17.1M triples, the 100,000-record HG005 slice) and
+#: below the one that exhausted 31 GB (170.9M). It is deliberately not derived
+#: from measured bytes-per-triple: rdflib's footprint depends on term sharing
+#: and IRI length, so a constant here is a conservative guard rather than a
+#: prediction, and it is overridable for a machine that can afford more.
+DEFAULT_SHACL_MAX_TRIPLES = 50_000_000
+
+#: Node's V8 heap ceiling for the Comunica-backed endpoints (comunica, hdt,
+#: cottas -- all three go through ComunicaHttpEndpointMixin).
+#:
+#: Node sizes its old-space from a default that does not track the machine, so
+#: on a 31 GB host the endpoint still died with "Reached heap limit Allocation
+#: failed - JavaScript heap out of memory" on the sample-level query of a
+#: 170.9M-triple HDT graph, with 25 GB free. The kernel OOM killer was never
+#: involved: the process aborted itself inside a ceiling it chose.
+#:
+#: None keeps Node's default, which is the behaviour every published result was
+#: produced under. Set a value to raise it.
+DEFAULT_NODE_HEAP_MB: int | None = None
+
+
+def node_endpoint_env(heap_mb: int | None, base: dict[str, str] | None = None) -> dict[str, str]:
+ """Environment for a Comunica endpoint, with an optional heap ceiling.
+
+ Appends rather than replaces NODE_OPTIONS, so a caller's own setting is
+ kept and only the heap is added.
+ """
+ env = dict(base if base is not None else os.environ)
+ if not heap_mb:
+ return env
+ option = f"--max-old-space-size={heap_mb}"
+ existing = env.get("NODE_OPTIONS", "").strip()
+ env["NODE_OPTIONS"] = f"{existing} {option}".strip() if existing else option
+ return env
+
+
+
+def shacl_exceeds_limit(triple_count: int | None, limit: int | None) -> bool:
+ """Whether a decoded graph is too large to hand to pyshacl.
+
+ An unknown count is treated as too large, for the same reason the wrapper
+ treats an unreadable size that way: skipping a check is recoverable and is
+ recorded, while exhausting memory mid-run loses the whole validation.
+ A limit of 0 disables the gate.
+ """
+ if not limit:
+ return False
+ if triple_count is None:
+ return True
+ return triple_count > limit
+
+
def validate_shacl(
source: Path,
shapes: Path | list[Path],
@@ -2435,6 +2493,9 @@ def _init_endpoint(self, options: dict[str, Any]) -> None:
options.get("comunica_warmup_timeout")
or max(DEFAULT_COMUNICA_WARMUP_TIMEOUT, self.query_timeout)
)
+ # None leaves Node's own default, which is what every published result
+ # was produced under.
+ self.node_heap_mb = options.get("node_heap_mb") or DEFAULT_NODE_HEAP_MB
self.server: subprocess.Popen | None = None
self.executable: str | None = None
self.endpoint: str | None = None
@@ -2486,6 +2547,7 @@ def _start_endpoint(self) -> None:
stdout=server_log.open("wb"),
stderr=subprocess.STDOUT,
start_new_session=True,
+ env=node_endpoint_env(getattr(self, "node_heap_mb", None)),
)
self._await_bind(server_log)
self._await_warm(server_log)
@@ -3958,11 +4020,52 @@ def run_validation(args: argparse.Namespace) -> int:
)
rdf_validation = validate_ntriples(decoded, results_dir)
shacl_result = None
- if args.shacl_shapes is not None:
+ # The authoritative size gate for pyshacl, because this is the only
+ # point that knows what pyshacl will actually load.
+ #
+ # The wrapper also gates, on the PACKAGED artifact's bytes, and that
+ # measure is wrong in a way that bites hardest on the best format.
+ # On a 170,935,101-triple graph the same graph landed on both sides
+ # of the wrapper's 512 MiB gate purely by packaging:
+ #
+ # cottas 390,728,158 B under -> shapes attempted -> OOM
+ # nt.gz 756,594,166 B over -> skipped
+ # hdt 1,182,206,289 B over -> skipped
+ #
+ # The COTTAS run was SIGKILLed at 32.2 GB RSS on a 31 GB machine
+ # after decoding and rapper had both succeeded on every triple. So
+ # the better a format compresses, the likelier it was to exhaust
+ # memory -- the guard inverted. Gating on the triple count fixes
+ # that, because a graph's cost to pyshacl does not depend on how it
+ # arrived.
+ decoded_triples = rdf_validation.get("tripleCount")
+ shacl_skipped: dict[str, Any] | None = None
+ shacl_limit = getattr(
+ args, "shacl_max_triples", DEFAULT_SHACL_MAX_TRIPLES
+ )
+ if args.shacl_shapes is not None and shacl_exceeds_limit(
+ decoded_triples, shacl_limit
+ ):
+ shacl_skipped = {
+ "status": "SKIPPED_TOO_LARGE",
+ "tripleCount": decoded_triples,
+ "limitTriples": shacl_limit,
+ "reason": (
+ f"the decoded graph holds {decoded_triples:,} triples, above the "
+ f"--shacl-max-triples limit of {shacl_limit:,}. "
+ f"pyshacl is in-memory, so attempting it risks exhausting memory "
+ f"mid-run; skipping is recoverable and is recorded here."
+ ),
+ }
+ eprint(f"[{args.dataset_id}] shapes skipped: {shacl_skipped['reason']}")
+ if args.shacl_shapes is not None and shacl_skipped is None:
progress.emit("progress", completed=0, detail="validating SHACL shapes")
shacl_result = validate_shacl(
decoded, args.shacl_shapes, results_dir, args.shacl_ontology
)
+ if shacl_skipped is not None:
+ shacl_result = shacl_skipped
+ write_json(results_dir / "shacl.json", shacl_skipped)
write_json(results_dir / "rdf-validation.json", rdf_validation)
materialization["decodedTripleCount"] = rdf_validation.get("tripleCount")
write_json(results_dir / "materialization.json", materialization)
@@ -3976,6 +4079,7 @@ def run_validation(args: argparse.Namespace) -> int:
"comunica_port": args.comunica_port,
"hdt_port": args.hdt_port,
"comunica_bind_timeout": args.comunica_bind_timeout,
+ "node_heap_mb": getattr(args, "node_heap_mb", DEFAULT_NODE_HEAP_MB),
"comunica_warmup_timeout": args.comunica_warmup_timeout,
}
engine_options["artifact_path"] = str(args.rdf)
@@ -4351,6 +4455,29 @@ def build_arg_parser() -> argparse.ArgumentParser:
"checked."
),
)
+ parser.add_argument(
+ "--node-heap-mb",
+ type=int,
+ default=DEFAULT_NODE_HEAP_MB,
+ help=(
+ "Raise the V8 old-space ceiling (MB) for the Comunica-backed "
+ "endpoints (comunica, hdt, cottas). Node does not size its heap "
+ "from the machine, so an endpoint can abort with a JavaScript "
+ "heap-out-of-memory while the host still has free memory. Unset "
+ "keeps Node's default"
+ ),
+ )
+ parser.add_argument(
+ "--shacl-max-triples",
+ type=int,
+ default=DEFAULT_SHACL_MAX_TRIPLES,
+ help=(
+ "Skip the shape layer when the decoded graph holds more triples "
+ "than this, recording the skip and its reason (0 disables the "
+ "gate). pyshacl is in-memory, and its cost tracks the graph rather "
+ f"than the artifact it arrived in (default: {DEFAULT_SHACL_MAX_TRIPLES:,})"
+ ),
+ )
parser.add_argument("--filter-oracle", choices=("auto", "bcftools", "cyvcf2"), default="auto")
parser.add_argument("--scratch-dir", type=Path, default=Path("/work"))
parser.add_argument(
diff --git a/test/test_query_selection_unit.py b/test/test_query_selection_unit.py
index 57d6aba..5544e6e 100644
--- a/test/test_query_selection_unit.py
+++ b/test/test_query_selection_unit.py
@@ -422,6 +422,11 @@ def _args(self, tmp_path, queries):
filter_oracle="cyvcf2",
dataset_id="sample",
queries=queries,
+ # Mirror the real argument surface. run_validation tolerates these
+ # being absent (see RunValidationToleratesAMinimalNamespaceTests),
+ # but a fixture that omits them stops exercising the real path.
+ shacl_max_triples=V.DEFAULT_SHACL_MAX_TRIPLES,
+ node_heap_mb=V.DEFAULT_NODE_HEAP_MB,
)
def _run(self, queries, *, corrupt=None):
diff --git a/test/test_shacl_gate_and_node_heap_unit.py b/test/test_shacl_gate_and_node_heap_unit.py
new file mode 100644
index 0000000..1aaa067
--- /dev/null
+++ b/test/test_shacl_gate_and_node_heap_unit.py
@@ -0,0 +1,225 @@
+"""Two memory guards, each fixed after it failed on a 170.9M-triple graph.
+
+**The shape-layer gate measured the wrong thing.** pyshacl is in-memory, so the
+wrapper size-gates it -- but on the PACKAGED artifact's bytes, while what
+pyshacl pays for is the graph. One 170,935,101-triple graph therefore landed on
+both sides of one 512 MiB gate purely by packaging:
+
+ cottas 390,728,158 B under -> shapes attempted -> OOM
+ nt.gz 756,594,166 B over -> skipped
+ hdt 1,182,206,289 B over -> skipped
+
+The COTTAS run was SIGKILLed at 32.2 GB RSS on a 31 GB machine, after COTTAS
+decoding and rapper had both succeeded on every triple. The better a format
+compresses, the likelier it was to exhaust memory: the guard inverted. The fix
+gates on the decoded triple count, which no packaging can change.
+
+**Node chose a ceiling the machine did not.** On the same graph the HDT
+endpoint aborted with "Reached heap limit Allocation failed - JavaScript heap
+out of memory" while ~25 GB was free; the kernel OOM killer was never involved.
+Node does not size its old-space from the host, so the endpoints now accept an
+explicit ceiling.
+"""
+
+import importlib.util
+import os
+import unittest
+from pathlib import Path
+
+from test.helpers import VerboseTestCase
+
+RUNNER_PATH = Path(__file__).resolve().parents[1] / "src" / "validation" / "validation_runner.py"
+_spec = importlib.util.spec_from_file_location("validation_runner_guards", RUNNER_PATH)
+V = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(V)
+
+# The real numbers from the run that motivated the fix.
+GRAPH_TRIPLES = 170_935_101
+ARTIFACT_BYTES = {
+ "cottas": 390_728_158,
+ "nt.gz": 756_594_166,
+ "hdt": 1_182_206_289,
+}
+WRAPPER_BYTE_GATE = 512 * 1024 * 1024
+
+
+class ShaclGateTests(VerboseTestCase):
+ def test_the_packaging_no_longer_decides(self):
+ """The property the bug violated: same graph, same verdict.
+
+ Every packaging of one graph must get one answer. This is the
+ regression test for the inversion itself.
+ """
+ verdicts = {
+ kind: V.shacl_exceeds_limit(GRAPH_TRIPLES, V.DEFAULT_SHACL_MAX_TRIPLES)
+ for kind in ARTIFACT_BYTES
+ }
+ self.assertEqual(set(verdicts.values()), {True},
+ "the decoded graph is the same for every artifact, so "
+ "every artifact must reach the same decision")
+
+ def test_the_old_byte_gate_did_disagree_across_packagings(self):
+ """Pins why the fix was needed, so nobody reinstates the byte gate.
+
+ Not a test of current behaviour -- a record of the defect, expressed in
+ the numbers that produced it.
+ """
+ byte_verdicts = {k: b <= WRAPPER_BYTE_GATE for k, b in ARTIFACT_BYTES.items()}
+ self.assertTrue(byte_verdicts["cottas"], "cottas slipped under the gate")
+ self.assertFalse(byte_verdicts["nt.gz"])
+ self.assertFalse(byte_verdicts["hdt"])
+ self.assertEqual(len(set(byte_verdicts.values())), 2,
+ "the byte gate split one graph three ways; that was the bug")
+
+ def test_the_graph_that_oomed_is_refused(self):
+ self.assertTrue(V.shacl_exceeds_limit(GRAPH_TRIPLES, V.DEFAULT_SHACL_MAX_TRIPLES))
+
+ def test_the_campaigns_largest_validated_graph_still_passes(self):
+ """17.1M triples is the 100,000-record HG005 slice the campaign ran
+ shapes on. The gate must not retroactively disable published behaviour."""
+ self.assertFalse(V.shacl_exceeds_limit(17_098_746, V.DEFAULT_SHACL_MAX_TRIPLES))
+
+ def test_an_unknown_count_is_treated_as_too_large(self):
+ """Skipping is recoverable and recorded; exhausting memory is not."""
+ self.assertTrue(V.shacl_exceeds_limit(None, V.DEFAULT_SHACL_MAX_TRIPLES))
+
+ def test_zero_disables_the_gate(self):
+ for count in (0, 1, GRAPH_TRIPLES):
+ with self.subTest(count=count):
+ self.assertFalse(V.shacl_exceeds_limit(count, 0))
+ self.assertFalse(V.shacl_exceeds_limit(None, 0))
+
+ def test_the_boundary_is_inclusive(self):
+ limit = V.DEFAULT_SHACL_MAX_TRIPLES
+ self.assertFalse(V.shacl_exceeds_limit(limit, limit))
+ self.assertTrue(V.shacl_exceeds_limit(limit + 1, limit))
+
+ def test_the_default_sits_between_the_two_observations(self):
+ """It must admit what worked and refuse what died, or it is arbitrary."""
+ self.assertGreater(V.DEFAULT_SHACL_MAX_TRIPLES, 17_098_746)
+ self.assertLess(V.DEFAULT_SHACL_MAX_TRIPLES, GRAPH_TRIPLES)
+
+
+class NodeHeapEnvTests(VerboseTestCase):
+ def test_unset_leaves_the_environment_alone(self):
+ """Published results were produced under Node's default; keep it."""
+ env = V.node_endpoint_env(None, {"PATH": "/usr/bin"})
+ self.assertNotIn("NODE_OPTIONS", env)
+ self.assertEqual(env["PATH"], "/usr/bin")
+
+ def test_zero_is_also_unset(self):
+ self.assertNotIn("NODE_OPTIONS", V.node_endpoint_env(0, {}))
+
+ def test_a_ceiling_is_applied(self):
+ env = V.node_endpoint_env(16384, {})
+ self.assertEqual(env["NODE_OPTIONS"], "--max-old-space-size=16384")
+
+ def test_an_existing_setting_is_kept_not_replaced(self):
+ """A caller's own NODE_OPTIONS must survive; only the heap is added."""
+ env = V.node_endpoint_env(8192, {"NODE_OPTIONS": "--enable-source-maps"})
+ self.assertIn("--enable-source-maps", env["NODE_OPTIONS"])
+ self.assertIn("--max-old-space-size=8192", env["NODE_OPTIONS"])
+
+ def test_the_base_environment_is_not_mutated(self):
+ base = {"NODE_OPTIONS": "--enable-source-maps"}
+ V.node_endpoint_env(4096, base)
+ self.assertEqual(base["NODE_OPTIONS"], "--enable-source-maps")
+
+ def test_it_defaults_to_the_process_environment(self):
+ env = V.node_endpoint_env(None)
+ self.assertEqual(env.get("PATH"), os.environ.get("PATH"))
+
+ def test_the_default_is_node_s_own(self):
+ """Unset by default, so this change alters no existing measurement."""
+ self.assertIsNone(V.DEFAULT_NODE_HEAP_MB)
+
+
+if __name__ == "__main__":
+ unittest.main()
+
+
+class RunValidationToleratesAMinimalNamespaceTests(VerboseTestCase):
+ """A new option must not become a required attribute of run_validation.
+
+ run_validation is driven by hand-built namespaces in several places -- this
+ suite and the mutation harness among them -- so reading a new option with
+ plain attribute access breaks those callers at runtime rather than at
+ import. That is exactly how adding --shacl-max-triples and --node-heap-mb
+ broke CI: five tests failed with "'Namespace' object has no attribute
+ 'node_heap_mb'", and the failure surfaced in a merged feature's tests
+ rather than in the change that caused it.
+
+ The convention the file already used for progress_path and quiet is
+ getattr with the documented default. This pins it, so the next option
+ added cannot reintroduce the same break silently.
+ """
+
+ def test_run_validation_accepts_a_namespace_without_the_new_options(self):
+ """Behavioural, not textual: drive run_validation and look for the break.
+
+ An earlier version of this test asserted on the source text and failed
+ twice on formatting, which is the wrong instrument -- it constrains how
+ the guard is written rather than that it works.
+ """
+ import argparse
+ import json
+ import tempfile
+ from unittest import mock
+
+ with tempfile.TemporaryDirectory() as td:
+ tmp = Path(td)
+ (tmp / "scratch").mkdir()
+ (tmp / "s.vcf").write_text("##fileformat=VCFv4.2\n#CHROM\tPOS\n")
+ (tmp / "s.nt").write_text(" .\n")
+ # Deliberately missing shacl_max_triples and node_heap_mb.
+ args = argparse.Namespace(
+ results_dir=tmp / "results", representation="expanded",
+ info_representation="structured", header_representation="structured",
+ progress_path=None, quiet=True, scratch_dir=tmp / "scratch",
+ rdf=tmp / "s.nt", rdf_format="nt", engine="comunica",
+ engines=["comunica"], mapping_policy="strict",
+ strict_conformance=False, shacl_shapes=None,
+ query_timeout=60, validation_time_budget=0,
+ stop_after_query_timeout=False, qlever_memory_gb=4,
+ qlever_port=7019, comunica_port=7020, hdt_port=7021,
+ comunica_bind_timeout=60, comunica_warmup_timeout=60,
+ qlever_startup_timeout=60, qlever_index_arg=[],
+ qlever_server_arg=[], vcf=tmp / "s.vcf",
+ filter_oracle="cyvcf2", dataset_id="sample", queries=None,
+ )
+ engine = mock.MagicMock()
+ engine.describe.return_value = {"engine": "comunica"}
+ engine.execute.return_value = {"status": "FAILED"}
+ with mock.patch.object(V, "parse_vcf", return_value={
+ "totalRecords": 1, "sampleCount": 0, "gtRecordCount": 0,
+ "sourceSha256": "0" * 64}), \
+ mock.patch.object(V, "attach_census_expectations",
+ side_effect=lambda p, *a, **k: p), \
+ mock.patch.object(V, "validate_ntriples",
+ return_value={"status": "PASS", "tripleCount": 1}), \
+ mock.patch.object(V, "materialize_ntriples",
+ return_value=(tmp / "s.nt", {})), \
+ mock.patch.object(V, "build_manifest", return_value={}), \
+ mock.patch.object(V, "build_engine", return_value=engine):
+ V.run_validation(args)
+ summary = json.loads((args.results_dir / "summary.json").read_text())
+
+ # The run may fail for its own reasons -- the engine here is a stub --
+ # but never because an option was read by attribute.
+ self.assertNotIn(
+ "has no attribute", str(summary.get("error") or ""),
+ "run_validation read a new option by attribute; use getattr with "
+ "its documented default so callers that build their own Namespace "
+ "keep working",
+ )
+
+ def test_a_namespace_without_them_still_resolves(self):
+ """The behaviour the getattr guard buys, checked rather than assumed."""
+ import argparse
+
+ args = argparse.Namespace()
+ self.assertEqual(
+ getattr(args, "shacl_max_triples", V.DEFAULT_SHACL_MAX_TRIPLES),
+ V.DEFAULT_SHACL_MAX_TRIPLES,
+ )
+ self.assertIsNone(getattr(args, "node_heap_mb", V.DEFAULT_NODE_HEAP_MB))
diff --git a/vcf_rdfizer.py b/vcf_rdfizer.py
index 817b7d7..78bb7b8 100644
--- a/vcf_rdfizer.py
+++ b/vcf_rdfizer.py
@@ -8972,6 +8972,11 @@ def run_validation_mode(
# A subset selection. The runner turns this into a TIMING_ONLY result
# rather than a validation verdict -- see its --queries help.
("--queries", "queries"),
+ # The authoritative shape-layer gate lives in the runner, which is the
+ # only place that knows the decoded graph's size. The size gate here
+ # sees the packaged artifact and cannot.
+ ("--shacl-max-triples", "shacl_max_triples"),
+ ("--node-heap-mb", "node_heap_mb"),
):
value = options.get(key)
if value is not None:
@@ -9580,6 +9585,24 @@ def main():
default=None,
help="Per-query timeout in seconds for validation (default: engine default)",
)
+ parser.add_argument(
+ "--shacl-max-triples",
+ default=None,
+ help=(
+ "Skip the shape layer when the decoded graph exceeds this many "
+ "triples, recording the skip (0 disables). pyshacl is in-memory "
+ "and its cost tracks the graph, not the artifact it arrived in"
+ ),
+ )
+ parser.add_argument(
+ "--node-heap-mb",
+ default=None,
+ help=(
+ "Raise the V8 old-space ceiling (MB) for the Comunica-backed "
+ "validation engines (comunica, hdt, cottas). Node does not size "
+ "its heap from the machine"
+ ),
+ )
parser.add_argument(
"--validation-queries",
default=None,
@@ -9742,6 +9765,20 @@ def main():
validation_engine_options["stop_after_query_timeout"] = True
if args.validation_queries is not None:
validation_engine_options["queries"] = args.validation_queries
+ if args.shacl_max_triples is not None:
+ # 0 is meaningful here: it disables the gate, the way
+ # --validation-time-budget 0 means no ceiling.
+ try:
+ limit = int(args.shacl_max_triples)
+ except (TypeError, ValueError):
+ raise ValueError("--shacl-max-triples must be an integer")
+ if limit < 0:
+ raise ValueError("--shacl-max-triples must be zero or a positive integer")
+ validation_engine_options["shacl_max_triples"] = limit
+ if args.node_heap_mb is not None:
+ validation_engine_options["node_heap_mb"] = parse_positive_int(
+ args.node_heap_mb, name="--node-heap-mb"
+ )
if args.qlever_index_arg:
validation_engine_options["qlever_index_args"] = list(args.qlever_index_arg)
if args.qlever_server_arg: