diff --git a/harness/tests/test_dwh_preprocess_job.py b/harness/tests/test_dwh_preprocess_job.py index 9a650c5c..d612e750 100644 --- a/harness/tests/test_dwh_preprocess_job.py +++ b/harness/tests/test_dwh_preprocess_job.py @@ -1,5 +1,6 @@ import hashlib import json +from pathlib import Path from typer.testing import CliRunner @@ -22,8 +23,8 @@ def test_selected_dwh_stages_run_in_declared_order(tmp_path): workspace_root=tmp_path, config_fingerprint=FP, input_fingerprint=FP, - introspect=lambda: calls.append("introspect"), - build_lsh=lambda: calls.append("lsh"), + introspect=lambda output: (calls.append("introspect"), output.write_text("catalog")), + build_lsh=lambda physical, output: _write_lsh(calls, physical, output), ) report = pipeline.run(("introspect", "lsh")) @@ -31,6 +32,19 @@ def test_selected_dwh_stages_run_in_declared_order(tmp_path): assert report.status == "succeeded" assert calls == ["introspect", "lsh"] assert [stage.name for stage in report.stages] == ["introspect", "lsh"] + active = (tmp_path / ".tht-dwh" / "ACTIVE").read_text().strip() + published = tmp_path / ".tht-dwh" / "generations" / active + assert (published / "physical.yaml").read_text() == "catalog" + assert sorted(path.name for path in published.iterdir()) == [ + "demo_lsh.pkl", "demo_meta.json", "demo_minhashes.pkl", "physical.yaml" + ] + + +def _write_lsh(calls, physical: Path, output: Path): + calls.append("lsh") + assert physical.read_text() == "catalog" + for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json"): + (output / name).write_text(name) def test_preprocess_dwh_json_is_pristine(monkeypatch, tmp_path): @@ -81,3 +95,47 @@ def test_preprocess_dwh_rejects_unknown_or_duplicate_steps(monkeypatch, tmp_path assert response.exit_code == 2 assert json.loads(response.output)["status"] == "failed" assert called is False + + +def test_failed_multi_file_build_never_replaces_active_generation(tmp_path): + def catalog(output): + output.write_text("old-catalog") + + first = DwhPreprocessPipeline( + workspace_id="demo", workspace_root=tmp_path, + config_fingerprint=FP, input_fingerprint=FP, + introspect=catalog, + build_lsh=lambda physical, output: [ + (output / name).write_text(name) + for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json") + ], + ).run(("introspect", "lsh")) + assert first.status == "succeeded" + old_active = (tmp_path / ".tht-dwh" / "ACTIVE").read_text() + + def partial_lsh(physical, output): + (output / "demo_lsh.pkl").write_text("new-but-partial") + raise RuntimeError("crash between LSH files") + + failed = DwhPreprocessPipeline( + workspace_id="demo", workspace_root=tmp_path, + config_fingerprint=FP, input_fingerprint=FP, + introspect=lambda output: output.write_text("new-catalog"), + build_lsh=partial_lsh, + ).run(("introspect", "lsh")) + + assert failed.status == "failed" + assert (tmp_path / ".tht-dwh" / "ACTIVE").read_text() == old_active + + resumed = DwhPreprocessPipeline( + workspace_id="demo", workspace_root=tmp_path, + config_fingerprint=FP, input_fingerprint=FP, + introspect=lambda output: (_ for _ in ()).throw( + AssertionError("completed introspection must not repeat") + ), + build_lsh=lambda physical, output: [ + (output / name).write_text("recovered") + for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json") + ], + ).run(("introspect", "lsh"), resume_run_id=failed.run_id) + assert resumed.status == "succeeded" diff --git a/harness/tests/test_lsh_job_resume.py b/harness/tests/test_lsh_job_resume.py index 78a776e3..8b3f3368 100644 --- a/harness/tests/test_lsh_job_resume.py +++ b/harness/tests/test_lsh_job_resume.py @@ -1,5 +1,7 @@ import hashlib +import pytest + from tht.jobs.dwh_pipeline import DwhPreprocessPipeline @@ -9,10 +11,11 @@ FP = "sha256:" + hashlib.sha256(b"test").hexdigest() def test_lsh_failure_resumes_exact_run_without_repeating_introspection(tmp_path): calls = [] - def introspect(): + def introspect(output): calls.append("introspect") + output.write_text("catalog") - def fail_lsh(): + def fail_lsh(physical, output): calls.append("lsh-failed") raise RuntimeError("database detail that must not leak") @@ -32,7 +35,7 @@ def test_lsh_failure_resumes_exact_run_without_repeating_introspection(tmp_path) config_fingerprint=FP, input_fingerprint=FP, introspect=introspect, - build_lsh=lambda: calls.append("lsh-recovered"), + build_lsh=lambda physical, output: _recover_lsh(calls, output), ).run(("introspect", "lsh"), resume_run_id=failed.run_id) assert resumed.status == "succeeded" @@ -40,14 +43,20 @@ def test_lsh_failure_resumes_exact_run_without_repeating_introspection(tmp_path) assert calls == ["introspect", "lsh-failed", "lsh-recovered"] +def _recover_lsh(calls, output): + calls.append("lsh-recovered") + for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json"): + (output / name).write_text(name) + + def test_resume_rejects_a_different_stage_selection(tmp_path): failed = DwhPreprocessPipeline( workspace_id="demo", workspace_root=tmp_path, config_fingerprint=FP, input_fingerprint=FP, - introspect=lambda: None, - build_lsh=lambda: (_ for _ in ()).throw(RuntimeError()), + introspect=lambda output: output.write_text("catalog"), + build_lsh=lambda physical, output: (_ for _ in ()).throw(RuntimeError()), ).run(("introspect", "lsh")) pipeline = DwhPreprocessPipeline( @@ -55,8 +64,8 @@ def test_resume_rejects_a_different_stage_selection(tmp_path): workspace_root=tmp_path, config_fingerprint=FP, input_fingerprint=FP, - introspect=lambda: None, - build_lsh=lambda: None, + introspect=lambda output: output.write_text("catalog"), + build_lsh=lambda physical, output: None, ) try: pipeline.run(("lsh",), resume_run_id=failed.run_id) @@ -64,3 +73,71 @@ def test_resume_rejects_a_different_stage_selection(tmp_path): assert "incompatible" in str(error) else: raise AssertionError("resume with different stages must fail") + + +def test_resume_rejects_tampered_succeeded_stage_artifact(tmp_path): + failed = 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: (_ for _ in ()).throw(RuntimeError()), + ).run(("introspect", "lsh")) + artifact = ( + tmp_path / ".tht-jobs" / "dwh" / "runs" / failed.run_id + / "artifacts" / "physical.yaml" + ) + artifact.write_text("tampered") + + pipeline = 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: _recover_lsh([], output), + ) + with pytest.raises(Exception, match="artifact manifest is invalid"): + pipeline.run(("introspect", "lsh"), resume_run_id=failed.run_id) + + +def test_post_publish_crash_reconciles_same_generation_on_resume(tmp_path): + crashed = False + + def crash_once(_generation): + nonlocal crashed + if not crashed: + crashed = True + raise KeyboardInterrupt("simulated process death") + + pipeline = 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: _recover_lsh([], output), + after_publish=crash_once, + ) + with pytest.raises(KeyboardInterrupt): + pipeline.run(("introspect", "lsh")) + active = (tmp_path / ".tht-dwh" / "ACTIVE").read_text().strip() + checkpoint = next((tmp_path / ".tht-jobs" / "dwh" / "runs").glob("*/checkpoint.json")) + source_run_id = checkpoint.parent.name + assert active == source_run_id + + resumed = pipeline.run(("introspect", "lsh"), resume_run_id=source_run_id) + assert resumed.status == "succeeded" + assert (tmp_path / ".tht-dwh" / "ACTIVE").read_text().strip() == resumed.run_id + + +def test_resume_of_succeeded_run_detects_tampered_published_file(tmp_path): + pipeline = 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: _recover_lsh([], output), + ) + succeeded = pipeline.run(("introspect", "lsh")) + published = ( + tmp_path / ".tht-dwh" / "generations" / succeeded.run_id / "demo_meta.json" + ) + published.write_text("tampered") + + with pytest.raises(Exception, match="digest mismatch"): + pipeline.run(("introspect", "lsh"), resume_run_id=succeeded.run_id) diff --git a/harness/tht/cli/lsh_cmd.py b/harness/tht/cli/lsh_cmd.py index 16210ae0..e341d1df 100644 --- a/harness/tht/cli/lsh_cmd.py +++ b/harness/tht/cli/lsh_cmd.py @@ -9,6 +9,11 @@ lsh_app = typer.Typer(help="Indice LSH su valori dei campi (derivato, rigenerabi def _lsh_dir(cfg) -> Path: + from tht.jobs.dwh_pipeline import active_generation_dir + + active = active_generation_dir(cfg.paths.artifacts.parent) + if active is not None: + return active return cfg.paths.indexes / "lsh" @@ -36,14 +41,17 @@ def _extract_lsh_values(dwh, physical, annotations, limit): return values, skipped, truncated -def build_lsh_artifacts(cfg, *, dwh=None, verbose: bool = False): +def build_lsh_artifacts( + cfg, *, dwh=None, verbose: bool = False, physical_file: Path | None = None, + output_dir: Path | None = None, +): """Run the existing LSH extraction/build algorithm and persist its outputs.""" from tht.adapters.factory import build_dwh from tht.cli.schema_cmd import annotations_path from tht.lshindex import build_index, save_index from tht.mschema.models import Annotations, PhysicalSchema - phys_file = physical_path(cfg) + phys_file = physical_file or physical_path(cfg) if not phys_file.exists(): raise FileNotFoundError("physical catalog is missing; run schema introspect first") physical = PhysicalSchema.from_yaml(phys_file) @@ -53,7 +61,10 @@ def build_lsh_artifacts(cfg, *, dwh=None, verbose: bool = False): target, physical, annotations, cfg.lsh.max_values_per_column ) lsh, minhashes = build_index(values, cfg.lsh, verbose=verbose) - save_index(lsh, minhashes, cfg.lsh, _lsh_dir(cfg), name=cfg.database.db_schema) + save_index( + lsh, minhashes, cfg.lsh, output_dir or (cfg.paths.indexes / "lsh"), + name=cfg.database.db_schema, + ) return minhashes, skipped, truncated, values @@ -69,9 +80,25 @@ def build_cmd(config: Path = CONFIG_OPT) -> None: ) raise typer.Exit(code=1) typer.echo("Estrazione valori (i più frequenti) dalle colonne testuali eligible...") - minhashes, skipped, truncated, values = build_lsh_artifacts(cfg, verbose=True) - n_values = sum(len(v) for t in values.values() for v in t.values()) - typer.echo(f" {n_values} valori da {sum(len(t) for t in values.values())} colonne") + from tht.jobs.dwh_pipeline import active_generation_dir + + if active_generation_dir(cfg.paths.artifacts.parent) is None: + minhashes, skipped, truncated, values = build_lsh_artifacts(cfg, verbose=True) + n_values = sum(len(v) for table in values.values() for v in table.values()) + n_columns = sum(len(table) for table in values.values()) + else: + from tht.cli.preprocess_cmd import run_dwh_from_config + from tht.lshindex import load_index + + report = run_dwh_from_config(config, steps=("lsh",)) + if report.status != "succeeded": + typer.secho("ERRORE: DWH preprocessing failed", fg=typer.colors.RED, err=True) + raise typer.Exit(code=1) + _, minhashes, _ = load_index(_lsh_dir(cfg), name=cfg.database.db_schema) + skipped, truncated = [], [] + n_values = len(minhashes) + n_columns = len({(entry[1], entry[2]) for entry in minhashes.values()}) + typer.echo(f" {n_values} valori da {n_columns} colonne") for s in skipped: typer.secho(f" saltata {s.table}.{s.column}: {s.reason}", fg=typer.colors.YELLOW) for t in truncated: diff --git a/harness/tht/cli/preprocess_cmd.py b/harness/tht/cli/preprocess_cmd.py index 0e627a42..5b102a98 100644 --- a/harness/tht/cli/preprocess_cmd.py +++ b/harness/tht/cli/preprocess_cmd.py @@ -19,18 +19,29 @@ def run_dwh_from_config( config: Path, *, steps: tuple[str, ...], resume: str | None = None, ): from tht.cli.lsh_cmd import build_lsh_artifacts - from tht.cli.schema_cmd import _load_config_or_exit, refresh_catalog - from tht.jobs.dwh_pipeline import DwhPreprocessPipeline, fingerprint + 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 cfg = _load_config_or_exit(config) workspace_root = cfg.paths.artifacts.parent + active = active_generation_dir(workspace_root) + lsh_names = ( + f"{cfg.database.db_schema}_lsh.pkl", + f"{cfg.database.db_schema}_minhashes.pkl", + f"{cfg.database.db_schema}_meta.json", + ) pipeline = DwhPreprocessPipeline( workspace_id=config.stem.lower().replace(".", "-").replace("_", "-"), workspace_root=workspace_root, config_fingerprint=fingerprint(cfg.model_dump_json()), input_fingerprint=fingerprint(config.resolve().as_posix()), - introspect=lambda: refresh_catalog(cfg), - build_lsh=lambda: build_lsh_artifacts(cfg), + introspect=lambda output: refresh_catalog(cfg, output_path=output), + build_lsh=lambda physical, output: build_lsh_artifacts( + cfg, physical_file=physical, output_dir=output + ), + lsh_filenames=lsh_names, + current_physical=physical_path(cfg), + current_lsh_dir=active if active is not None else cfg.paths.indexes / "lsh", ) return pipeline.run(steps, resume_run_id=resume) diff --git a/harness/tht/cli/schema_cmd.py b/harness/tht/cli/schema_cmd.py index eaaa6eb8..92471422 100644 --- a/harness/tht/cli/schema_cmd.py +++ b/harness/tht/cli/schema_cmd.py @@ -37,6 +37,11 @@ def _load_config_or_exit(config: Path): def physical_path(cfg) -> Path: + from tht.jobs.dwh_pipeline import active_generation_dir + + active = active_generation_dir(cfg.paths.artifacts.parent) + if active is not None: + return active / "physical.yaml" return cfg.paths.artifacts / "mschema" / "physical.yaml" @@ -44,13 +49,13 @@ def annotations_path(cfg) -> Path: return cfg.paths.artifacts / "mschema" / "annotations.yaml" -def refresh_catalog(cfg, *, dwh=None): +def refresh_catalog(cfg, *, dwh=None, output_path: Path | None = None): """Run the existing catalog algorithm and persist its canonical output.""" target = dwh if dwh is not None else build_dwh(cfg) physical = target.introspect() _add_examples(target, physical, cfg.examples) classify_all(physical, cfg.eligibility) - physical.to_yaml(physical_path(cfg)) + physical.to_yaml(output_path or (cfg.paths.artifacts / "mschema" / "physical.yaml")) return physical @@ -90,7 +95,18 @@ def introspect_cmd( ) return try: - phys = refresh_catalog(cfg) + from tht.jobs.dwh_pipeline import active_generation_dir + + if active_generation_dir(cfg.paths.artifacts.parent) is None: + phys = refresh_catalog(cfg) + else: + from tht.cli.preprocess_cmd import run_dwh_from_config + from tht.mschema.models import PhysicalSchema + + report = run_dwh_from_config(config, steps=("introspect",)) + if report.status != "succeeded": + raise RuntimeError("DWH preprocessing failed") + phys = PhysicalSchema.from_yaml(physical_path(cfg)) except Exception as e: typer.secho(f"ERRORE: {e}", fg=typer.colors.RED, err=True) raise typer.Exit(code=1) diff --git a/harness/tht/jobs/dwh_pipeline.py b/harness/tht/jobs/dwh_pipeline.py index 72e32e55..d10f4c8f 100644 --- a/harness/tht/jobs/dwh_pipeline.py +++ b/harness/tht/jobs/dwh_pipeline.py @@ -1,20 +1,47 @@ -"""Resumable DWH catalog and LSH preprocessing stages.""" +"""Crash-safe, resumable DWH catalog and LSH preprocessing stages.""" from __future__ import annotations import hashlib +import json +import os +import re +import shutil +import stat +import uuid from collections.abc import Callable from pathlib import Path from tht.jobs.models import JobReport, JobSpec -from tht.jobs.runner import run_job +from tht.jobs.runner import ( + CorruptCheckpointError, + JobContext, + StageArtifacts, + run_job, + seal_stage_artifacts, +) DWH_STAGE_IDS = ("introspect", "lsh") +_RUN_ID = re.compile(r"^[0-9a-f]{32}$") + + +def active_generation_dir(workspace_root: Path) -> Path | None: + pointer = workspace_root / ".tht-dwh" / "ACTIVE" + try: + generation = pointer.read_text(encoding="utf-8").strip() + except FileNotFoundError: + return None + if not _RUN_ID.fullmatch(generation) or pointer.is_symlink(): + raise CorruptCheckpointError("DWH ACTIVE pointer is invalid") + target = pointer.parent / "generations" / generation + if not target.is_dir() or target.is_symlink(): + raise CorruptCheckpointError("active DWH generation is missing") + return target class DwhPreprocessPipeline: - """Adapt existing DWH preprocessing operations to the shared job envelope.""" + """Stage a complete artifact bundle, then publish it through one atomic pointer.""" def __init__( self, @@ -23,42 +50,218 @@ class DwhPreprocessPipeline: workspace_root: Path, config_fingerprint: str, input_fingerprint: str, - introspect: Callable[[], object], - build_lsh: Callable[[], object], + introspect: Callable[[Path], object], + build_lsh: Callable[[Path, Path], object], + lsh_filenames: tuple[str, str, str] | None = None, + current_physical: Path | None = None, + current_lsh_dir: Path | None = None, + after_publish: Callable[[str], object] | None = None, ) -> None: self.workspace_id = workspace_id self.workspace_root = workspace_root self.config_fingerprint = config_fingerprint self.input_fingerprint = input_fingerprint - self._operations = {"introspect": introspect, "lsh": build_lsh} + self.introspect = introspect + self.build_lsh = build_lsh + self.lsh_filenames = lsh_filenames or ( + f"{workspace_id}_lsh.pkl", f"{workspace_id}_minhashes.pkl", + f"{workspace_id}_meta.json", + ) + self.current_physical = current_physical + self.current_lsh_dir = current_lsh_dir + self.after_publish = after_publish def run( self, steps: tuple[str, ...] = DWH_STAGE_IDS, *, resume_run_id: str | None = None ) -> JobReport: + self._validate_steps(steps) + if resume_run_id is not None: + self._validate_resume_publication(resume_run_id) + spec = JobSpec( + workspace_id=self.workspace_id, + job_type="dwh", + workspace_root=self.workspace_root, + spec_version="jobs-v1", + pipeline_version="dwh-v2", + config_fingerprint=self.config_fingerprint, + input_fingerprint=self.input_fingerprint, + stage_ids=steps, + resume_run_id=resume_run_id, + ) + + def introspect_stage(context: JobContext): + artifacts = self._artifacts(context) + physical = artifacts / "physical.yaml" + try: + self.introspect(physical) + except BaseException: + physical.unlink(missing_ok=True) + raise + if steps[-1] == "introspect": + self._copy_current_lsh(artifacts) + return self._publish_stage(context, "introspect", spec) + return StageArtifacts(("physical.yaml",)) + + def lsh_stage(context: JobContext): + artifacts = self._artifacts(context) + physical = artifacts / "physical.yaml" + if not physical.exists(): + source = self.current_physical + if source is None or not source.is_file(): + raise FileNotFoundError("physical catalog is missing") + shutil.copyfile(source, physical) + try: + self.build_lsh(physical, artifacts) + except BaseException: + for name in self.lsh_filenames: + (artifacts / name).unlink(missing_ok=True) + raise + return self._publish_stage(context, "lsh", spec) + + implementations = {"introspect": introspect_stage, "lsh": lsh_stage} + return run_job(spec, tuple(implementations[step] for step in steps)) + + def _validate_resume_publication(self, run_id: str) -> None: + run_dir = self.workspace_root / ".tht-jobs" / "dwh" / "runs" / run_id + try: + checkpoint = json.loads((run_dir / "checkpoint.json").read_text(encoding="utf-8")) + except (OSError, ValueError, TypeError) as error: + raise CorruptCheckpointError("checkpoint is invalid and cannot be resumed") from error + target = self.workspace_root / ".tht-dwh" / "generations" / run_id + if not target.exists(): + if checkpoint.get("status") == "succeeded": + raise CorruptCheckpointError("published DWH generation is missing") + return + try: + manifest = json.loads( + (run_dir / "artifacts" / "artifact-manifest.json").read_text(encoding="utf-8") + ) + required = tuple( + name + for stage in manifest["stages"].values() + for name in stage["required"] + ) + except (OSError, KeyError, ValueError, TypeError) as error: + raise CorruptCheckpointError("resume artifact manifest is invalid") from error + self._validate_published(target, run_dir / "artifacts", required) + + @staticmethod + def _artifacts(context: JobContext) -> Path: + root = context.run_dir / "artifacts" + root.mkdir(exist_ok=True) + return root + + def _required(self, artifacts: Path) -> tuple[str, ...]: + required = ("physical.yaml",) + tuple( + name for name in self.lsh_filenames if (artifacts / name).is_file() + ) + if not (artifacts / "physical.yaml").is_file(): + raise CorruptCheckpointError("staged physical catalog is missing") + lsh_count = len(required) - 1 + if lsh_count not in (0, len(self.lsh_filenames)): + raise CorruptCheckpointError("staged LSH artifact set is incomplete") + return required + + def _copy_current_lsh(self, artifacts: Path) -> None: + if self.current_lsh_dir is None: + return + existing = [self.current_lsh_dir / name for name in self.lsh_filenames] + if not any(path.exists() for path in existing): + return + if not all(path.is_file() for path in existing): + raise CorruptCheckpointError("current LSH artifact set is incomplete") + for source in existing: + shutil.copyfile(source, artifacts / source.name) + + def _publish_stage(self, context: JobContext, stage: str, spec: JobSpec): + artifacts = self._artifacts(context) + required = self._required(artifacts) + seal_stage_artifacts(context, stage, required, spec) + self._publish(context.run_id, artifacts, required) + if self.after_publish is not None: + self.after_publish(context.run_id) + return StageArtifacts(required) + + def _publish(self, generation: str, artifacts: Path, required: tuple[str, ...]) -> None: + root = self.workspace_root / ".tht-dwh" + generations = root / "generations" + self._ensure_owned_dir(root) + self._ensure_owned_dir(generations) + target = generations / generation + if target.exists(): + self._validate_published(target, artifacts, required) + else: + temporary = generations / f".{generation}.{uuid.uuid4().hex}.tmp" + temporary.mkdir(mode=0o700) + try: + for name in required: + destination = temporary / name + shutil.copyfile(artifacts / name, destination) + with destination.open("rb") as stream: + os.fsync(stream.fileno()) + self._fsync(temporary) + os.replace(temporary, target) + self._fsync(generations) + except BaseException: + shutil.rmtree(temporary, ignore_errors=True) + raise + pointer_tmp = root / f".ACTIVE.{uuid.uuid4().hex}.tmp" + fd = os.open(pointer_tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + try: + with os.fdopen(fd, "w", encoding="utf-8") as stream: + stream.write(generation + "\n") + stream.flush() + os.fsync(stream.fileno()) + os.replace(pointer_tmp, root / "ACTIVE") + self._fsync(root) + except BaseException: + pointer_tmp.unlink(missing_ok=True) + raise + + @staticmethod + def _validate_published(target: Path, artifacts: Path, required: tuple[str, ...]) -> None: + if ( + target.is_symlink() + or not target.is_dir() + or {path.name for path in target.iterdir()} != set(required) + ): + raise CorruptCheckpointError("published DWH generation is invalid") + for name in required: + source, published = artifacts / name, target / name + if published.is_symlink() or not published.is_file(): + raise CorruptCheckpointError("published DWH artifact is invalid") + if hashlib.sha256(source.read_bytes()).digest() != hashlib.sha256( + published.read_bytes() + ).digest(): + raise CorruptCheckpointError("published DWH artifact digest mismatch") + + @staticmethod + def _ensure_owned_dir(path: Path) -> None: + try: + path.mkdir(mode=0o700) + except FileExistsError: + pass + info = path.lstat() + if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid(): + raise OSError("unsafe DWH publication directory") + path.chmod(0o700) + + @staticmethod + def _fsync(path: Path) -> None: + fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(fd) + finally: + os.close(fd) + + @staticmethod + def _validate_steps(steps: tuple[str, ...]) -> None: if not steps or len(steps) != len(set(steps)) or any( step not in DWH_STAGE_IDS for step in steps ): raise ValueError("DWH preprocessing steps must be unique introspect/lsh stages") if tuple(sorted(steps, key=DWH_STAGE_IDS.index)) != steps: raise ValueError("DWH preprocessing steps must follow introspect,lsh order") - spec = JobSpec( - workspace_id=self.workspace_id, - job_type="dwh", - workspace_root=self.workspace_root, - spec_version="jobs-v1", - pipeline_version="dwh-v1", - config_fingerprint=self.config_fingerprint, - input_fingerprint=self.input_fingerprint, - stage_ids=steps, - resume_run_id=resume_run_id, - ) - def stage(operation): - def execute(_context): - return operation() - - return execute - - return run_job(spec, tuple(stage(self._operations[step]) for step in steps)) def fingerprint(value: str) -> str: