From 2f6daaaea6cb61d34f35d056acaafd9b22bf43fc Mon Sep 17 00:00:00 2001 From: mptyl Date: Sun, 12 Jul 2026 07:26:10 +0200 Subject: [PATCH] fix(dwh): anchor generation operations to lease fd --- .superpowers/sdd/evidence-task-6-report.md | 20 + harness/tests/test_dwh_preprocess_job.py | 132 ++++- harness/tht/jobs/dwh_pipeline.py | 585 ++++++++++++--------- 3 files changed, 495 insertions(+), 242 deletions(-) create mode 100644 .superpowers/sdd/evidence-task-6-report.md diff --git a/.superpowers/sdd/evidence-task-6-report.md b/.superpowers/sdd/evidence-task-6-report.md new file mode 100644 index 00000000..91ccc3b9 --- /dev/null +++ b/.superpowers/sdd/evidence-task-6-report.md @@ -0,0 +1,20 @@ +# Evidence Task 6 — final fd-anchored DWH correction + +All DWH generation state below `.tht-dwh` is now accessed relative to the directory descriptor +retained by the shared/exclusive generation lease. ACTIVE reads, atomic temp writes, replacement, +fsync, and rollback use `openat`/`replaceat` operations. Generation staging, validation, +reconciliation, resume checks, retention classification, and recursive deletion likewise use owned +root/generations/candidate descriptors with `O_NOFOLLOW`; locked operations no longer reopen +generation paths through `workspace_root`. + +Portable reader snapshots are copied from validated generation file descriptors into private 0700 +process-owned temporary directories while the shared lease is held. This avoids Linux-only +`/proc/self/fd` paths and prevents a renamed/replaced `.tht-dwh` pathname from redirecting later +schema or LSH reads. Lease-scoped copies are removed on exit and standalone snapshots are removed +at process exit. + +Deterministic adversarial tests rename the DWH root after lease acquisition during ACTIVE reads, +ACTIVE publication, and retention cleanup. Each test proves the replacement tree is never read, +written, or deleted; the descriptor-pinned original either completes consistently or fails closed. +Existing owner binding, legacy rejection, crash reconciliation, resume, atomic rollback, retention, +and reader/writer exclusion behavior remains covered. diff --git a/harness/tests/test_dwh_preprocess_job.py b/harness/tests/test_dwh_preprocess_job.py index 1c5b2a13..fe1aa8e9 100644 --- a/harness/tests/test_dwh_preprocess_job.py +++ b/harness/tests/test_dwh_preprocess_job.py @@ -445,6 +445,8 @@ def test_unsafe_lsh_filename_is_rejected(tmp_path): def test_active_fsync_failure_restores_previous_pointer(monkeypatch, tmp_path): + import os + import tht.jobs.dwh_pipeline as module def build(physical, output): for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json"): (output / name).write_text(name) @@ -455,27 +457,147 @@ def test_active_fsync_failure_restores_previous_pointer(monkeypatch, tmp_path): introspect=lambda output: output.write_text("old"), build_lsh=build, ) first = first_pipeline.run() - original_fsync = first_pipeline._fsync + root = tmp_path / ".tht-dwh" + root_identity = (root.stat().st_dev, root.stat().st_ino) + original_fsync = module.os.fsync failed_once = False - def fail_active_once(path): + def fail_active_once(fd): nonlocal failed_once - if path.name == ".tht-dwh" and not failed_once: + info = os.fstat(fd) + if ( + (info.st_dev, info.st_ino) == root_identity + and "ACTIVE" in os.listdir(fd) + and not failed_once + ): failed_once = True raise OSError("injected directory fsync failure") - original_fsync(path) + original_fsync(fd) second = DwhPreprocessPipeline( workspace_id="demo", workspace_root=tmp_path, config_fingerprint=FP, input_fingerprint=FP, introspect=lambda output: output.write_text("new"), build_lsh=build, ) - monkeypatch.setattr(second, "_fsync", fail_active_once) + monkeypatch.setattr(module.os, "fsync", fail_active_once) failed = second.run() assert failed.status == "failed" assert (tmp_path / ".tht-dwh" / "ACTIVE").read_text().strip() == first.run_id +def test_snapshot_root_swap_after_lease_never_reads_replacement(monkeypatch, tmp_path): + import tht.jobs.dwh_pipeline as module + + pipeline = DwhPreprocessPipeline( + workspace_id="demo", workspace_root=tmp_path, + config_fingerprint=FP, input_fingerprint=FP, + introspect=lambda output: output.write_text("trusted"), + build_lsh=lambda physical, output: [ + (output / name).write_text("trusted") + for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json") + ], + ) + first = pipeline.run() + assert first.status == "succeeded" + root = tmp_path / ".tht-dwh" + moved = tmp_path / "moved-read-root" + replacement = root + real_read = module._read_owned_at + swapped = False + + def swapping_read(directory_fd, name, *, readonly): + nonlocal swapped + if name == "ACTIVE" and not swapped: + swapped = True + replacement.rename(moved) + replacement.mkdir(mode=0o700) + (replacement / "sentinel").write_text("replacement-secret") + return real_read(directory_fd, name, readonly=readonly) + + monkeypatch.setattr(module, "_read_owned_at", swapping_read) + try: + with lease_dwh_snapshot(snapshot_config(tmp_path)) as snapshot: + assert snapshot.physical.read_text() == "trusted" + except Exception as error: + assert "ACTIVE" in str(error) or "root" in str(error) + assert swapped + assert (replacement / "sentinel").read_text() == "replacement-secret" + + +def test_publish_root_swap_after_lease_never_writes_replacement(monkeypatch, tmp_path): + import tht.jobs.dwh_pipeline as module + + def make(content): + return DwhPreprocessPipeline( + workspace_id="demo", workspace_root=tmp_path, + config_fingerprint=FP, input_fingerprint=FP, + introspect=lambda output: output.write_text(content), + build_lsh=lambda physical, output: [ + (output / name).write_text(content) + for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json") + ], + ) + + first = make("old").run() + assert first.status == "succeeded", first + root = tmp_path / ".tht-dwh" + moved = tmp_path / "moved-publish-root" + real_replace = module.os.replace + swapped = False + + def swapping_replace(source, destination, *args, **kwargs): + nonlocal swapped + if destination == "ACTIVE" and kwargs.get("dst_dir_fd") is not None and not swapped: + swapped = True + root.rename(moved) + root.mkdir(mode=0o700) + (root / "sentinel").write_text("replacement-safe") + return real_replace(source, destination, *args, **kwargs) + + monkeypatch.setattr(module.os, "replace", swapping_replace) + result = make("new").run() + assert result.status in {"succeeded", "failed"} + assert swapped + assert (root / "sentinel").read_text() == "replacement-safe" + moved_active = (moved / "ACTIVE").read_text().strip() + assert len(moved_active) == 32 + assert (moved / "generations" / moved_active).is_dir() + + +def test_cleanup_root_swap_after_lease_never_deletes_replacement(monkeypatch, tmp_path): + import tht.jobs.dwh_pipeline as module + + pipeline = DwhPreprocessPipeline( + workspace_id="demo", workspace_root=tmp_path, + config_fingerprint=FP, input_fingerprint=FP, + introspect=lambda output: output.write_text("trusted"), + build_lsh=lambda physical, output: [ + (output / name).write_text("trusted") + for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json") + ], + retain_generations=1, + ) + pipeline.run() + root = tmp_path / ".tht-dwh" + moved = tmp_path / "moved-cleanup-root" + real_open = module.os.open + swapped = False + + def swapping_open(path, flags, *args, **kwargs): + nonlocal swapped + if path == "generations" and kwargs.get("dir_fd") is not None and not swapped: + swapped = True + root.rename(moved) + root.mkdir(mode=0o700) + (root / "sentinel").write_text("replacement-safe") + return real_open(path, flags, *args, **kwargs) + + monkeypatch.setattr(module.os, "open", swapping_open) + pipeline._cleanup_generations() + assert swapped + assert (root / "sentinel").read_text() == "replacement-safe" + + def test_snapshot_stays_on_one_generation_across_publish(tmp_path): def pipeline(content): return DwhPreprocessPipeline( diff --git a/harness/tht/jobs/dwh_pipeline.py b/harness/tht/jobs/dwh_pipeline.py index a4d8722e..bda6eacf 100644 --- a/harness/tht/jobs/dwh_pipeline.py +++ b/harness/tht/jobs/dwh_pipeline.py @@ -5,10 +5,12 @@ from __future__ import annotations import hashlib import json import fcntl +import atexit import os import re import shutil import stat +import tempfile import uuid from collections.abc import Callable from dataclasses import dataclass @@ -29,6 +31,16 @@ _RUN_ID = re.compile(r"^[0-9a-f]{32}$") _SAFE_FILE = re.compile(r"^[A-Za-z0-9_-]+\.(?:pkl|json)$") GENERATION_MANIFEST = "generation-manifest.json" OWNER_MARKER = "OWNER.json" +_SNAPSHOT_DIRS: set[Path] = set() + + +def _cleanup_snapshot_dirs() -> None: + for path in tuple(_SNAPSHOT_DIRS): + shutil.rmtree(path, ignore_errors=True) + _SNAPSHOT_DIRS.discard(path) + + +atexit.register(_cleanup_snapshot_dirs) def config_dwh_binding(cfg) -> dict[str, str]: @@ -48,19 +60,6 @@ def _binding_digest(binding: dict[str, str]) -> str: return hashlib.sha256(payload.encode("utf-8")).hexdigest() -def _read_root_binding(workspace_root: Path) -> dict[str, str]: - try: - root_fd = os.open( - workspace_root / ".tht-dwh", os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW - ) - try: - return _read_root_binding_fd(root_fd) - finally: - os.close(root_fd) - except OSError as error: - raise CorruptCheckpointError("DWH workspace ownership marker is missing or invalid") from error - - def _read_root_binding_fd(root_fd: int) -> dict[str, str]: try: fd = os.open(OWNER_MARKER, os.O_RDONLY | os.O_NOFOLLOW, dir_fd=root_fd) @@ -92,19 +91,6 @@ def _read_root_binding_fd(root_fd: int) -> dict[str, str]: raise CorruptCheckpointError("DWH workspace ownership marker is missing or invalid") from error -def _validate_root_binding(workspace_root: Path, expected: dict[str, str]) -> None: - try: - root_fd = os.open( - workspace_root / ".tht-dwh", os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW - ) - except OSError as error: - raise CorruptCheckpointError("DWH workspace ownership marker is missing or invalid") from error - try: - _validate_root_binding_fd(root_fd, expected) - finally: - os.close(root_fd) - - def _validate_root_binding_fd(root_fd: int, expected: dict[str, str]) -> None: if _read_root_binding_fd(root_fd) != expected: raise CorruptCheckpointError("DWH artifacts belong to a different workspace configuration") @@ -195,6 +181,7 @@ class DwhArtifactSnapshot: generation: str | None physical: Path lsh_dir: Path + _holder: object | None = None @dataclass @@ -249,6 +236,9 @@ class DwhSnapshotLease: if self._lease is not None: lease, self._lease = self._lease, None lease.close() + if self.snapshot is not None and isinstance(self.snapshot._holder, Path): + shutil.rmtree(self.snapshot._holder, ignore_errors=True) + _SNAPSHOT_DIRS.discard(self.snapshot._holder) def lease_dwh_snapshot(cfg) -> DwhSnapshotLease: @@ -323,25 +313,6 @@ def _digest(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() -def _read_owned(path: Path, *, readonly: bool) -> bytes: - fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW) - try: - info = os.fstat(fd) - if ( - not stat.S_ISREG(info.st_mode) - or info.st_uid != os.getuid() - or info.st_nlink != 1 - or (readonly and bool(info.st_mode & 0o222)) - ): - raise OSError("unsafe DWH generation file") - chunks = [] - while chunk := os.read(fd, 1024 * 1024): - chunks.append(chunk) - return b"".join(chunks) - finally: - os.close(fd) - - def _read_owned_at(directory_fd: int, name: str, *, readonly: bool) -> bytes: fd = os.open(name, os.O_RDONLY | os.O_NOFOLLOW, dir_fd=directory_fd) try: @@ -361,6 +332,94 @@ def _read_owned_at(directory_fd: int, name: str, *, readonly: bool) -> bytes: os.close(fd) +def _open_generations_fd(root_fd: int, *, create: bool = False) -> int: + if create: + try: + os.mkdir("generations", 0o700, dir_fd=root_fd) + os.fsync(root_fd) + except FileExistsError: + pass + fd = os.open( + "generations", os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=root_fd + ) + info = os.fstat(fd) + if ( + not stat.S_ISDIR(info.st_mode) + or info.st_uid != os.getuid() + or stat.S_IMODE(info.st_mode) != 0o700 + ): + os.close(fd) + raise CorruptCheckpointError("DWH generations directory is invalid") + return fd + + +def _open_generation_fd(generations_fd: int, generation: str) -> int: + if not _RUN_ID.fullmatch(generation): + raise CorruptCheckpointError("DWH generation identity is invalid") + try: + return os.open( + generation, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, + dir_fd=generations_fd, + ) + except OSError as error: + raise CorruptCheckpointError("active DWH generation is missing") from error + + +def _active_generation_fd( + root_fd: int, expected_binding: dict[str, str] +) -> tuple[str, int] | None: + _validate_root_binding_fd(root_fd, expected_binding) + try: + generation = _read_owned_at(root_fd, "ACTIVE", readonly=False).decode().strip() + except FileNotFoundError: + try: + generations_fd = _open_generations_fd(root_fd) + except FileNotFoundError: + return None + try: + if os.listdir(generations_fd): + raise CorruptCheckpointError( + "DWH generations exist without a consistent ACTIVE pointer" + ) + finally: + os.close(generations_fd) + return None + except (OSError, UnicodeDecodeError) as error: + raise CorruptCheckpointError("DWH ACTIVE pointer is invalid") from error + if not _RUN_ID.fullmatch(generation): + raise CorruptCheckpointError("DWH ACTIVE pointer is invalid") + generations_fd = _open_generations_fd(root_fd) + try: + generation_fd = _open_generation_fd(generations_fd, generation) + finally: + os.close(generations_fd) + try: + validate_generation_fd(generation_fd, generation, expected_binding) + return generation, generation_fd + except BaseException: + os.close(generation_fd) + raise + + +def _materialize_generation_fd( + generation_fd: int, generation: str, binding: dict[str, str] +) -> tuple[Path, Path]: + holder = Path(tempfile.mkdtemp(prefix="tht-dwh-snapshot-")) + _SNAPSHOT_DIRS.add(holder) + snapshot_root = holder + try: + manifest = validate_generation_fd(generation_fd, generation, binding) + for name in manifest["files"]: + destination = snapshot_root / name + destination.write_bytes(_read_owned_at(generation_fd, name, readonly=True)) + destination.chmod(0o400) + return holder, snapshot_root + except BaseException: + shutil.rmtree(holder, ignore_errors=True) + _SNAPSHOT_DIRS.discard(holder) + raise + + def validate_generation_fd( directory_fd: int, generation: str, expected_binding: dict[str, str] | None = None, ) -> dict: @@ -405,19 +464,6 @@ def validate_generation_fd( raise CorruptCheckpointError("published DWH generation is invalid") from error -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, expected_binding) - finally: - os.close(fd) - - def resolve_dwh_snapshot(cfg) -> DwhArtifactSnapshot: binding = config_dwh_binding(cfg) lease = _acquire_existing_generation_lock( @@ -432,16 +478,21 @@ def resolve_dwh_snapshot(cfg) -> DwhArtifactSnapshot: def _resolve_dwh_snapshot_locked( cfg, binding: dict[str, str], root_fd: int, ) -> DwhArtifactSnapshot: - _validate_root_binding_fd(root_fd, binding) - target = _active_generation_dir_locked( - cfg.paths.artifacts.parent, binding, root_fd=root_fd - ) - if target is None: + active = _active_generation_fd(root_fd, binding) + if active is None: return DwhArtifactSnapshot( None, cfg.paths.artifacts / "mschema" / "physical.yaml", cfg.paths.indexes / "lsh" ) - validate_generation(target, binding) - return DwhArtifactSnapshot(target.name, target / "physical.yaml", target) + generation, generation_fd = active + try: + holder, snapshot_root = _materialize_generation_fd( + generation_fd, generation, binding + ) + finally: + os.close(generation_fd) + return DwhArtifactSnapshot( + generation, snapshot_root / "physical.yaml", snapshot_root, holder + ) def active_generation_dir( @@ -449,36 +500,17 @@ def active_generation_dir( ) -> Path | None: lease = _acquire_existing_generation_lock(workspace_root, exclusive=False) try: - return _active_generation_dir_locked( - workspace_root, expected_binding, root_fd=lease.root_fd - ) + active = _active_generation_fd(lease.root_fd, expected_binding) + if active is None: + return None + generation, generation_fd = active + os.close(generation_fd) + lease.assert_root_identity(workspace_root / ".tht-dwh") + return workspace_root / ".tht-dwh" / "generations" / generation finally: lease.close() -def _active_generation_dir_locked( - workspace_root: Path, expected_binding: dict[str, str], *, root_fd: int | None = None -) -> Path | None: - if root_fd is None: - _validate_root_binding(workspace_root, expected_binding) - else: - _validate_root_binding_fd(root_fd, expected_binding) - pointer = workspace_root / ".tht-dwh" / "ACTIVE" - try: - generation = _read_owned(pointer, readonly=False).decode("utf-8").strip() - except FileNotFoundError: - return None - except (OSError, UnicodeDecodeError) as error: - raise CorruptCheckpointError("DWH ACTIVE pointer is invalid") from error - if not _RUN_ID.fullmatch(generation): - 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") - validate_generation(target, expected_binding) - return target - - class DwhPreprocessPipeline: """Stage a complete artifact bundle, then publish it through one atomic pointer.""" @@ -509,6 +541,7 @@ class DwhPreprocessPipeline: ) self.current_physical = current_physical self.current_lsh_dir = current_lsh_dir + self._snapshot_holder = None self.after_publish = after_publish if isinstance(retain_generations, bool) or retain_generations < 1: raise ValueError("retain_generations must be positive") @@ -528,11 +561,10 @@ class DwhPreprocessPipeline: def _assert_active_binding(self, root_fd: int) -> None: _validate_root_binding_fd(root_fd, self.binding) - active = _active_generation_dir_locked( - self.workspace_root, self.binding, root_fd=root_fd - ) + active = _active_generation_fd(root_fd, self.binding) if active is not None: - validate_generation(active, self.binding) + _, generation_fd = active + os.close(generation_fd) def run( self, steps: tuple[str, ...] = DWH_STAGE_IDS, *, resume_run_id: str | None = None @@ -544,12 +576,17 @@ class DwhPreprocessPipeline: _claim_or_validate_root_binding(lease.root_fd, self.binding) lease.assert_root_identity(self.workspace_root / ".tht-dwh") self._assert_active_binding(lease.root_fd) - active = _active_generation_dir_locked( - self.workspace_root, self.binding, root_fd=lease.root_fd - ) + active = _active_generation_fd(lease.root_fd, self.binding) if active is not None: - self.current_physical = active / "physical.yaml" - self.current_lsh_dir = active + generation, generation_fd = active + try: + self._snapshot_holder, snapshot_root = _materialize_generation_fd( + generation_fd, generation, self.binding + ) + finally: + os.close(generation_fd) + self.current_physical = snapshot_root / "physical.yaml" + self.current_lsh_dir = snapshot_root finally: lease.close() if resume_run_id is not None: @@ -608,12 +645,22 @@ class DwhPreprocessPipeline: ) if running is None: return set() - target = self.workspace_root / ".tht-dwh" / "generations" / source.run_id - validate_generation(target, self.binding) - active = active_generation_dir(self.workspace_root, self.binding) - if active != target: - raise CorruptCheckpointError("sealed DWH publication is not ACTIVE") - self._validate_published(target, run_dir / "artifacts", running.artifact_files) + lease = _acquire_existing_generation_lock(self.workspace_root, exclusive=True) + try: + generations_fd = _open_generations_fd(lease.root_fd) + try: + self._validate_published_fd( + generations_fd, source.run_id, + run_dir / "artifacts", running.artifact_files, + ) + active = _active_generation_fd(lease.root_fd, self.binding) + if active is None or active[0] != source.run_id: + raise CorruptCheckpointError("sealed DWH publication is not ACTIVE") + os.close(active[1]) + finally: + os.close(generations_fd) + finally: + lease.close() return {running.name} def _validate_resume_publication(self, run_id: str) -> None: @@ -622,23 +669,41 @@ class DwhPreprocessPipeline: 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 + lease = _acquire_existing_generation_lock(self.workspace_root, exclusive=True) 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) + try: + generations_fd = _open_generations_fd(lease.root_fd) + except FileNotFoundError: + if checkpoint.get("status") == "succeeded": + raise CorruptCheckpointError("published DWH generation is missing") + return + try: + if run_id not in os.listdir(generations_fd): + 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_fd( + generations_fd, run_id, run_dir / "artifacts", required + ) + finally: + os.close(generations_fd) + finally: + lease.close() def _assert_no_legacy_artifacts(self, root_fd: int) -> None: if OWNER_MARKER in os.listdir(root_fd): @@ -692,119 +757,165 @@ class DwhPreprocessPipeline: try: _validate_root_binding_fd(lease.root_fd, self.binding) self._assert_active_binding(lease.root_fd) - self._publish(context.run_id, artifacts, required) + self._publish(context.run_id, artifacts, required, lease.root_fd) if self.after_publish is not None: self.after_publish(context.run_id) - self._cleanup_generations() + self._cleanup_generations_fd(lease.root_fd) finally: lease.close() 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()) - destination.chmod(0o400) - manifest = { - "schema_version": 1, - "generation": generation, - "files": {name: _digest(artifacts / name) for name in required}, - "job_spec_fingerprint": json.loads( - (artifacts / "artifact-manifest.json").read_text(encoding="utf-8") - )["spec_fingerprint"], - "artifact_manifest_sha256": _digest( - artifacts / "artifact-manifest.json" - ), - **self.binding, - } - manifest_path = temporary / GENERATION_MANIFEST - manifest_path.write_text( - json.dumps(manifest, sort_keys=True, separators=(",", ":")) + "\n", - encoding="utf-8", - ) - manifest_path.chmod(0o400) - with manifest_path.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 = root / "ACTIVE" - try: - previous = pointer.read_text(encoding="utf-8") - except FileNotFoundError: - previous = None - 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, pointer) - try: - self._fsync(root) - except BaseException: - self._restore_pointer(root, pointer, previous) - raise - except BaseException: - pointer_tmp.unlink(missing_ok=True) - raise - - def _validate_published( - self, target: Path, artifacts: Path, required: tuple[str, ...], + def _publish( + self, generation: str, artifacts: Path, required: tuple[str, ...], root_fd: int ) -> 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( - artifacts / "artifact-manifest.json" - ): - raise CorruptCheckpointError("published DWH job manifest digest mismatch") - 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") + generations_fd = _open_generations_fd(root_fd, create=True) + temporary = f".{generation}.{uuid.uuid4().hex}.tmp" + try: + if generation in os.listdir(generations_fd): + self._validate_published_fd(generations_fd, generation, artifacts, required) + else: + os.mkdir(temporary, 0o700, dir_fd=generations_fd) + temporary_fd = os.open( + temporary, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, + dir_fd=generations_fd, + ) + try: + for name in required: + self._write_readonly_at( + temporary_fd, name, (artifacts / name).read_bytes() + ) + artifact_manifest = artifacts / "artifact-manifest.json" + manifest = { + "schema_version": 1, + "generation": generation, + "files": {name: _digest(artifacts / name) for name in required}, + "job_spec_fingerprint": json.loads( + artifact_manifest.read_text(encoding="utf-8") + )["spec_fingerprint"], + "artifact_manifest_sha256": _digest(artifact_manifest), + **self.binding, + } + self._write_readonly_at( + temporary_fd, GENERATION_MANIFEST, + (json.dumps(manifest, sort_keys=True, separators=(",", ":")) + "\n").encode(), + ) + os.fsync(temporary_fd) + finally: + os.close(temporary_fd) + os.rename( + temporary, generation, + src_dir_fd=generations_fd, dst_dir_fd=generations_fd, + ) + temporary = "" + os.fsync(generations_fd) + previous = None + try: + previous = _read_owned_at(root_fd, "ACTIVE", readonly=False) + except FileNotFoundError: + pass + self._replace_active_at(root_fd, (generation + "\n").encode(), previous) + finally: + if temporary: + self._safe_delete_generation(generations_fd, temporary, allow_temporary=True) + os.close(generations_fd) - def _restore_pointer(self, root: Path, pointer: Path, previous: str | None) -> None: + @staticmethod + def _write_readonly_at(directory_fd: int, name: str, payload: bytes) -> None: + fd = os.open( + name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, + 0o600, dir_fd=directory_fd, + ) + try: + offset = 0 + while offset < len(payload): + offset += os.write(fd, payload[offset:]) + os.fsync(fd) + os.fchmod(fd, 0o400) + os.fsync(fd) + finally: + os.close(fd) + + def _replace_active_at(self, root_fd: int, payload: bytes, previous: bytes | None) -> None: + temporary = f".ACTIVE.{uuid.uuid4().hex}.tmp" + fd = os.open( + temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, + 0o600, dir_fd=root_fd, + ) + try: + offset = 0 + while offset < len(payload): + offset += os.write(fd, payload[offset:]) + os.fsync(fd) + finally: + os.close(fd) + try: + os.replace(temporary, "ACTIVE", src_dir_fd=root_fd, dst_dir_fd=root_fd) + temporary = "" + try: + os.fsync(root_fd) + except BaseException: + self._restore_pointer_at(root_fd, previous) + raise + finally: + if temporary: + try: + os.unlink(temporary, dir_fd=root_fd) + except FileNotFoundError: + pass + + def _validate_published_fd( + self, generations_fd: int, generation: str, + artifacts: Path, required: tuple[str, ...], + ) -> None: + generation_fd = _open_generation_fd(generations_fd, generation) + try: + manifest = validate_generation_fd(generation_fd, generation, self.binding) + if set(manifest["files"]) != set(required): + raise CorruptCheckpointError("published DWH generation is invalid") + artifact_digest = _digest(artifacts / "artifact-manifest.json") + if manifest["artifact_manifest_sha256"] != artifact_digest: + raise CorruptCheckpointError("published DWH job manifest digest mismatch") + for name in required: + if hashlib.sha256((artifacts / name).read_bytes()).hexdigest() != manifest["files"][name]: + raise CorruptCheckpointError("published DWH artifact digest mismatch") + finally: + os.close(generation_fd) + + def _restore_pointer_at(self, root_fd: int, previous: bytes | None) -> None: if previous is None: - pointer.unlink(missing_ok=True) + try: + os.unlink("ACTIVE", dir_fd=root_fd) + except FileNotFoundError: + pass else: - restore = root / f".ACTIVE.restore.{uuid.uuid4().hex}.tmp" - restore.write_text(previous, encoding="utf-8") - with restore.open("rb") as stream: - os.fsync(stream.fileno()) - os.replace(restore, pointer) - self._fsync(root) + restore = f".ACTIVE.restore.{uuid.uuid4().hex}.tmp" + fd = os.open( + restore, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, + 0o600, dir_fd=root_fd, + ) + try: + offset = 0 + while offset < len(previous): + offset += os.write(fd, previous[offset:]) + os.fsync(fd) + finally: + os.close(fd) + os.replace(restore, "ACTIVE", src_dir_fd=root_fd, dst_dir_fd=root_fd) + os.fsync(root_fd) def _cleanup_generations(self) -> None: - root = self.workspace_root / ".tht-dwh" / "generations" - if not root.exists(): - return - _validate_root_binding(self.workspace_root, self.binding) - active = _active_generation_dir_locked(self.workspace_root, self.binding) + lease = _acquire_existing_generation_lock(self.workspace_root, exclusive=True) + try: + self._cleanup_generations_fd(lease.root_fd) + finally: + lease.close() + + def _cleanup_generations_fd(self, root_fd: int) -> None: + _validate_root_binding_fd(root_fd, self.binding) + active = _active_generation_fd(root_fd, self.binding) + active_name = active[0] if active else None if active is not None: - validate_generation(active, self.binding) - active_name = active.name if active else None + os.close(active[1]) protected = {active_name} if active_name else set() runs = self.workspace_root / ".tht-jobs" / "dwh" / "runs" for checkpoint in runs.glob("*/checkpoint.json") if runs.exists() else (): @@ -815,14 +926,18 @@ class DwhPreprocessPipeline: except (OSError, KeyError, ValueError): continue generations = [] - root_fd = os.open(root, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) try: - for name in os.listdir(root_fd): + generations_fd = _open_generations_fd(root_fd) + except FileNotFoundError: + return + try: + for name in os.listdir(generations_fd): if not _RUN_ID.fullmatch(name): continue try: candidate_fd = os.open( - name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=root_fd + name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, + dir_fd=generations_fd, ) except OSError: continue @@ -843,13 +958,17 @@ class DwhPreprocessPipeline: for _, name in generations: if name in protected | keep_recent: continue - self._safe_delete_generation(root_fd, name) - os.fsync(root_fd) + self._safe_delete_generation(generations_fd, name) + os.fsync(generations_fd) finally: - os.close(root_fd) + os.close(generations_fd) @staticmethod - def _safe_delete_generation(root_fd: int, name: str) -> None: + def _safe_delete_generation( + root_fd: int, name: str, *, allow_temporary: bool = False + ) -> None: + if not (_RUN_ID.fullmatch(name) or (allow_temporary and name.startswith("."))): + return try: generation_fd = os.open( name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=root_fd @@ -905,14 +1024,6 @@ class DwhPreprocessPipeline: 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(