mptyl/ThothII
1
0
Fork 0
Code Issues 25 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

fix(preprocess): bind DWH artifacts to workspace

Browse Source
This commit is contained in:
marcopan
2026-07-12 06:46:06 +02:00
parent 0a2421c513
commit b43075289e
4 changed files with 146 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
harness/tests/test_dwh_preprocess_job.py
+76 -10
View File
@@ -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()))
harness/tht/cli/preprocess_cmd.py
+7 -4
View File
@@ -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
harness/tht/config.py
+5 -1
View File
@@ -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
harness/tht/jobs/dwh_pipeline.py
+58 -9
View File
@@ -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: