From b43075289e04c2a656681ae10dddce2bb79af2eb Mon Sep 17 00:00:00 2001 From: mptyl Date: Sun, 12 Jul 2026 06:46:06 +0200 Subject: [PATCH] fix(preprocess): bind DWH artifacts to workspace --- harness/tests/test_dwh_preprocess_job.py | 86 +++++++++++++++++++++--- harness/tht/cli/preprocess_cmd.py | 11 +-- harness/tht/config.py | 6 +- harness/tht/jobs/dwh_pipeline.py | 67 +++++++++++++++--- 4 files changed, 146 insertions(+), 24 deletions(-) diff --git a/harness/tests/test_dwh_preprocess_job.py b/harness/tests/test_dwh_preprocess_job.py index 7c85c7aa..32db8753 100644 --- a/harness/tests/test_dwh_preprocess_job.py +++ b/harness/tests/test_dwh_preprocess_job.py @@ -14,6 +14,18 @@ from tht.jobs.locking import _lock_name FP = "sha256:" + hashlib.sha256(b"test").hexdigest() +def snapshot_config(tmp_path, workspace_id="demo"): + from types import SimpleNamespace + + cfg = SimpleNamespace( + paths=SimpleNamespace(artifacts=tmp_path / "artifacts", indexes=tmp_path / "indexes"), + _workspace_id=workspace_id, + _config_source="test", + ) + cfg.model_dump_json = lambda: "test" + return cfg + + def test_dwh_and_evidence_jobs_have_distinct_lock_names(): assert _lock_name("demo", "dwh") != _lock_name("demo", "evidence") @@ -43,6 +55,68 @@ def test_selected_dwh_stages_run_in_declared_order(tmp_path): ] +def test_shared_root_rejects_other_workspace_before_builder_or_read(tmp_path): + calls = [] + owner = DwhPreprocessPipeline( + workspace_id="demo", workspace_root=tmp_path, + config_fingerprint=FP, input_fingerprint=FP, + introspect=lambda output: output.write_text("catalog"), + build_lsh=lambda physical, output: _write_lsh([], physical, output), + ) + published = owner.run() + contender = DwhPreprocessPipeline( + workspace_id="other", workspace_root=tmp_path, + config_fingerprint=FP, input_fingerprint=FP, + introspect=lambda output: calls.append("introspect"), + build_lsh=lambda physical, output: calls.append("lsh"), + ) + + import pytest + with pytest.raises(Exception, match="different workspace configuration"): + contender.run() + with pytest.raises(Exception, match="different workspace configuration"): + resolve_dwh_snapshot(snapshot_config(tmp_path, "other")) + + assert calls == [] + assert resolve_dwh_snapshot(snapshot_config(tmp_path)).generation == published.run_id + + +def test_shared_root_mismatch_fails_without_deadlock_while_owner_reader_is_active(tmp_path): + import threading + + owner = DwhPreprocessPipeline( + workspace_id="demo", workspace_root=tmp_path, + config_fingerprint=FP, input_fingerprint=FP, + introspect=lambda output: output.write_text("catalog"), + build_lsh=lambda physical, output: _write_lsh([], physical, output), + ) + owner.run() + contender = DwhPreprocessPipeline( + workspace_id="other", workspace_root=tmp_path, + config_fingerprint=FP, input_fingerprint=FP, + introspect=lambda output: (_ for _ in ()).throw(AssertionError("builder called")), + build_lsh=lambda physical, output: None, + ) + finished = threading.Event() + errors = [] + with lease_dwh_snapshot(snapshot_config(tmp_path)): + thread = threading.Thread( + target=lambda: (errors.append(_capture_error(contender.run)), finished.set()) + ) + thread.start() + assert finished.wait(2) + thread.join(2) + assert "different workspace configuration" in str(errors[0]) + + +def _capture_error(operation): + try: + operation() + except Exception as error: + return error + raise AssertionError("operation unexpectedly succeeded") + + def _write_lsh(calls, physical: Path, output: Path): calls.append("lsh") assert physical.read_text() == "catalog" @@ -189,8 +263,6 @@ def test_active_fsync_failure_restores_previous_pointer(monkeypatch, tmp_path): def test_snapshot_stays_on_one_generation_across_publish(tmp_path): - from types import SimpleNamespace - def pipeline(content): return DwhPreprocessPipeline( workspace_id="demo", workspace_root=tmp_path, @@ -203,9 +275,7 @@ def test_snapshot_stays_on_one_generation_across_publish(tmp_path): ) first = pipeline("old").run() - cfg = SimpleNamespace(paths=SimpleNamespace( - artifacts=tmp_path / "artifacts", indexes=tmp_path / "indexes" - )) + cfg = snapshot_config(tmp_path) snapshot = resolve_dwh_snapshot(cfg) pipeline("new").run() assert snapshot.generation == first.run_id @@ -328,8 +398,6 @@ def test_retention_candidate_swap_to_symlink_is_never_followed(monkeypatch, tmp_ def test_reader_lease_blocks_retain_one_publisher_until_file_reads_finish(tmp_path): import threading import time - from types import SimpleNamespace - def make(content, retain=1): return DwhPreprocessPipeline( workspace_id="demo", workspace_root=tmp_path, @@ -342,9 +410,7 @@ def test_reader_lease_blocks_retain_one_publisher_until_file_reads_finish(tmp_pa ) first = make("old").run() - cfg = SimpleNamespace(paths=SimpleNamespace( - artifacts=tmp_path / "artifacts", indexes=tmp_path / "indexes" - )) + cfg = snapshot_config(tmp_path) completed = threading.Event() with lease_dwh_snapshot(cfg) as snapshot: thread = threading.Thread(target=lambda: (make("new").run(), completed.set())) diff --git a/harness/tht/cli/preprocess_cmd.py b/harness/tht/cli/preprocess_cmd.py index 20d94c73..55634db3 100644 --- a/harness/tht/cli/preprocess_cmd.py +++ b/harness/tht/cli/preprocess_cmd.py @@ -20,9 +20,12 @@ def run_dwh_from_config( ): from tht.cli.lsh_cmd import build_lsh_artifacts from tht.cli.schema_cmd import _load_config_or_exit, physical_path, refresh_catalog - from tht.jobs.dwh_pipeline import DwhPreprocessPipeline, active_generation_dir, fingerprint + from tht.jobs.dwh_pipeline import ( + DwhPreprocessPipeline, active_generation_dir, config_dwh_binding, + ) cfg = _load_config_or_exit(config) + binding = config_dwh_binding(cfg) workspace_root = cfg.paths.artifacts.parent active = active_generation_dir(workspace_root) lsh_names = ( @@ -31,10 +34,10 @@ def run_dwh_from_config( f"{cfg.database.db_schema}_meta.json", ) pipeline = DwhPreprocessPipeline( - workspace_id=config.stem.lower().replace(".", "-").replace("_", "-"), + workspace_id=binding["workspace_id"], workspace_root=workspace_root, - config_fingerprint=fingerprint(cfg.model_dump_json()), - input_fingerprint=fingerprint(config.resolve().as_posix()), + config_fingerprint=binding["config_fingerprint"], + input_fingerprint=binding["input_fingerprint"], introspect=lambda output: refresh_catalog(cfg, output_path=output), build_lsh=lambda physical, output: build_lsh_artifacts( cfg, physical_file=physical, output_dir=output diff --git a/harness/tht/config.py b/harness/tht/config.py index 363f8b1f..808a73cc 100644 --- a/harness/tht/config.py +++ b/harness/tht/config.py @@ -5,7 +5,7 @@ from pathlib import Path from typing import Annotated, Any, Literal import yaml -from pydantic import BaseModel, Field, SecretStr, model_validator, ValidationError +from pydantic import BaseModel, Field, PrivateAttr, SecretStr, model_validator, ValidationError from tht.config_compat import translate_legacy_config @@ -283,6 +283,8 @@ class ExecutionConfig(BaseModel): class Config(BaseModel): + _workspace_id: str = PrivateAttr(default="default") + _config_source: str = PrivateAttr(default="direct") dwh: DwhResourceConfig vectors: VectorResourceConfig | None = None roots: WorkspaceRoots = WorkspaceRoots() @@ -376,6 +378,8 @@ def load_config(path: Path) -> Config: FutureWarning, stacklevel=2, ) + cfg._workspace_id = path.stem.lower().replace(".", "-").replace("_", "-") + cfg._config_source = path.resolve().as_posix() return cfg diff --git a/harness/tht/jobs/dwh_pipeline.py b/harness/tht/jobs/dwh_pipeline.py index af3a1488..ee73631d 100644 --- a/harness/tht/jobs/dwh_pipeline.py +++ b/harness/tht/jobs/dwh_pipeline.py @@ -30,6 +30,18 @@ _SAFE_FILE = re.compile(r"^[A-Za-z0-9_-]+\.(?:pkl|json)$") GENERATION_MANIFEST = "generation-manifest.json" +def config_dwh_binding(cfg) -> dict[str, str]: + workspace_id = getattr(cfg, "_workspace_id", None) + config_source = getattr(cfg, "_config_source", None) + if not isinstance(workspace_id, str) or not isinstance(config_source, str): + raise CorruptCheckpointError("DWH workspace identity is unavailable; reload configuration") + return { + "workspace_id": workspace_id, + "config_fingerprint": fingerprint(cfg.model_dump_json()), + "input_fingerprint": fingerprint(config_source), + } + + @dataclass(frozen=True) class DwhArtifactSnapshot: generation: str | None @@ -120,7 +132,9 @@ def _read_owned_at(directory_fd: int, name: str, *, readonly: bool) -> bytes: os.close(fd) -def validate_generation_fd(directory_fd: int, generation: str) -> dict: +def validate_generation_fd( + directory_fd: int, generation: str, expected_binding: dict[str, str] | None = None, +) -> dict: try: directory_info = os.fstat(directory_fd) if ( @@ -138,8 +152,17 @@ def validate_generation_fd(directory_fd: int, generation: str) -> dict: or not isinstance(files, dict) or not re.fullmatch(r"sha256:[0-9a-f]{64}", manifest["job_spec_fingerprint"]) or not re.fullmatch(r"[0-9a-f]{64}", manifest["artifact_manifest_sha256"]) + or not re.fullmatch(r"[a-z][a-z0-9_-]{0,63}", manifest["workspace_id"]) + or not re.fullmatch(r"sha256:[0-9a-f]{64}", manifest["config_fingerprint"]) + or not re.fullmatch(r"sha256:[0-9a-f]{64}", manifest["input_fingerprint"]) ): raise ValueError + if expected_binding is not None and any( + manifest.get(key) != value for key, value in expected_binding.items() + ): + raise CorruptCheckpointError( + "DWH artifacts belong to a different workspace configuration" + ) if set(os.listdir(directory_fd)) != set(files) | {GENERATION_MANIFEST}: raise ValueError for name, expected in files.items(): @@ -153,13 +176,15 @@ def validate_generation_fd(directory_fd: int, generation: str) -> dict: raise CorruptCheckpointError("published DWH generation is invalid") from error -def validate_generation(target: Path) -> dict: +def validate_generation( + target: Path, expected_binding: dict[str, str] | None = None, +) -> dict: try: fd = os.open(target, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) except OSError as error: raise CorruptCheckpointError("published DWH generation is invalid") from error try: - return validate_generation_fd(fd, target.name) + return validate_generation_fd(fd, target.name, expected_binding) finally: os.close(fd) @@ -170,7 +195,7 @@ def resolve_dwh_snapshot(cfg) -> DwhArtifactSnapshot: return DwhArtifactSnapshot( None, cfg.paths.artifacts / "mschema" / "physical.yaml", cfg.paths.indexes / "lsh" ) - validate_generation(target) + validate_generation(target, config_dwh_binding(cfg)) return DwhArtifactSnapshot(target.name, target / "physical.yaml", target) @@ -229,10 +254,29 @@ class DwhPreprocessPipeline: ): raise ValueError("LSH filenames must be unique flat safe names") + @property + def binding(self) -> dict[str, str]: + return { + "workspace_id": self.workspace_id, + "config_fingerprint": self.config_fingerprint, + "input_fingerprint": self.input_fingerprint, + } + + def _assert_active_binding(self) -> None: + active = active_generation_dir(self.workspace_root) + if active is not None: + validate_generation(active, self.binding) + def run( self, steps: tuple[str, ...] = DWH_STAGE_IDS, *, resume_run_id: str | None = None ) -> JobReport: self._validate_steps(steps) + lease_fd = _acquire_generation_lock(self.workspace_root, exclusive=False) + try: + self._assert_active_binding() + finally: + fcntl.flock(lease_fd, fcntl.LOCK_UN) + os.close(lease_fd) if resume_run_id is not None: self._validate_resume_publication(resume_run_id) spec = JobSpec( @@ -290,7 +334,7 @@ class DwhPreprocessPipeline: if running is None: return set() target = self.workspace_root / ".tht-dwh" / "generations" / source.run_id - validate_generation(target) + validate_generation(target, self.binding) active = active_generation_dir(self.workspace_root) if active != target: raise CorruptCheckpointError("sealed DWH publication is not ACTIVE") @@ -355,6 +399,7 @@ class DwhPreprocessPipeline: seal_stage_artifacts(context, stage, required, spec) lease_fd = _acquire_generation_lock(self.workspace_root, exclusive=True) try: + self._assert_active_binding() self._publish(context.run_id, artifacts, required) if self.after_publish is not None: self.after_publish(context.run_id) @@ -392,6 +437,7 @@ class DwhPreprocessPipeline: "artifact_manifest_sha256": _digest( artifacts / "artifact-manifest.json" ), + **self.binding, } manifest_path = temporary / GENERATION_MANIFEST manifest_path.write_text( @@ -429,9 +475,10 @@ class DwhPreprocessPipeline: pointer_tmp.unlink(missing_ok=True) raise - @staticmethod - def _validate_published(target: Path, artifacts: Path, required: tuple[str, ...]) -> None: - manifest = validate_generation(target) + def _validate_published( + self, target: Path, artifacts: Path, required: tuple[str, ...], + ) -> None: + manifest = validate_generation(target, self.binding) if set(manifest["files"]) != set(required): raise CorruptCheckpointError("published DWH generation is invalid") if manifest["artifact_manifest_sha256"] != _digest( @@ -463,6 +510,8 @@ class DwhPreprocessPipeline: if not root.exists(): return active = active_generation_dir(self.workspace_root) + if active is not None: + validate_generation(active, self.binding) active_name = active.name if active else None protected = {active_name} if active_name else set() runs = self.workspace_root / ".tht-jobs" / "dwh" / "runs" @@ -487,7 +536,7 @@ class DwhPreprocessPipeline: continue try: info = os.fstat(candidate_fd) - validate_generation_fd(candidate_fd, name) + validate_generation_fd(candidate_fd, name, self.binding) except CorruptCheckpointError: continue finally: