From 2598beed09e42407c45da22eb4827936faf7b558 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Wed, 29 Jul 2026 23:10:58 +0100 Subject: [PATCH] feat(lattice): add whole config target --- apps/predbat/lattice_whole_config_target.py | 1144 ++++++++++ .../tests/test_lattice_whole_config_target.py | 1839 +++++++++++++++++ 2 files changed, 2983 insertions(+) create mode 100644 apps/predbat/lattice_whole_config_target.py create mode 100644 apps/predbat/tests/test_lattice_whole_config_target.py diff --git a/apps/predbat/lattice_whole_config_target.py b/apps/predbat/lattice_whole_config_target.py new file mode 100644 index 000000000..4c3a94e44 --- /dev/null +++ b/apps/predbat/lattice_whole_config_target.py @@ -0,0 +1,1144 @@ +# ----------------------------------------------------------------------------- +# Predbat Home Battery System - Lattice generated-overlay target +# Copyright Trefor Southwell 2026 - All Rights Reserved +# This application maybe used for personal use only and not for commercial use +# ----------------------------------------------------------------------------- +"""Pure, detached whole-config target for a Lattice-owned generated overlay. + +This module implements the target protocol consumed by the default-off atomic +materializer. It does not register runtime code, construct a compiler, or +read/write PredBat's ``apps.yaml`` or ``secrets.yaml``. The target owns one +explicit JSON overlay whose complete contents are replaced atomically. +""" + +# cspell:ignore autoconfig CLOEXEC CREAT EINVAL ENOTSUP EOPNOTSUPP fchmod +# cspell:ignore fspath fstat fsync geteuid NONBLOCK NOFOLLOW RDONLY RDWR +# cspell:ignore readback ROLLEDBACK RONLY WRONLY + +import errno +import fcntl +import hashlib +import json +import os +import secrets +import stat +import threading +from contextlib import contextmanager +from dataclasses import dataclass, replace +from types import MappingProxyType + +from lattice_atomic_materializer import ( + MAX_CONFIG_BYTES, + MAX_SEQUENCE, + TargetOperationKey, + TargetOperationPhase, + TargetOperationSnapshot, + TargetTerminalPhase, + WholeConfigTarget, + _canonical_config, + _canonical_json, + _config_digest, + _configs_equal, + _validate_handle, +) +from lattice_durable_runtime_stores import ( + MAX_OPAQUE_PAYLOAD_BYTES, + OpaqueTargetJournalEntry, +) + + +TARGET_FORMAT = "predbat-lattice-generated-overlay-target" +TARGET_VERSION = 1 +_PROTECTED_FILENAMES = frozenset( + ( + "apps.yaml", + "apps.yml", + "secrets.yaml", + "secrets.yml", + ) +) +_SYNC_UNSUPPORTED = frozenset( + ( + errno.EINVAL, + getattr(errno, "ENOTSUP", errno.EINVAL), + getattr(errno, "EOPNOTSUPP", errno.EINVAL), + ) +) + + +class WholeConfigTargetError(RuntimeError): + """Base error for unavailable, corrupt, or conflicting target state.""" + + +class WholeConfigTargetCorrupt(WholeConfigTargetError): + """The overlay or target-owned journal is malformed or unsafe.""" + + +class WholeConfigTargetConflict(WholeConfigTargetError): + """A target transition lost its exact journal compare-and-store.""" + + +@dataclass(frozen=True, repr=False) +class _OverlayState: + """Exact overlay visibility, including the distinction of file absence.""" + + present: bool + config: MappingProxyType + + def __post_init__(self): + """Validate one exact bounded overlay observation.""" + if type(self.present) is not bool: + raise ValueError("overlay presence flag is invalid") + canonical = _canonical_config( + self.config, + "generated overlay", + ) + if not self.present and canonical: + raise ValueError("absent generated overlay must be empty") + if not _configs_equal(canonical, self.config): + raise ValueError("generated overlay is not canonical") + object.__setattr__(self, "config", canonical) + + +@dataclass(frozen=True, repr=False) +class _TargetRecord: + """Target-private journal state; repr deliberately excludes configs.""" + + key: TargetOperationKey + phase: TargetOperationPhase + prepared: bool + target_binding: str + allowlist_binding: str + handle: object = None + preimage: object = None + candidate: object = None + terminal_phase: object = None + + def __post_init__(self): + """Validate phase and exact target-owned state.""" + if type(self.key) is not TargetOperationKey: + raise ValueError("target record key is invalid") + TargetOperationKey.__post_init__(self.key) + if not isinstance(self.phase, TargetOperationPhase): + raise ValueError("target record phase is invalid") + for binding in ( + self.target_binding, + self.allowlist_binding, + ): + if type(binding) is not str or len(binding) != 64 or any(character not in "0123456789abcdef" for character in binding): + raise ValueError("target record binding is invalid") + if type(self.prepared) is not bool: + raise ValueError("target record prepared flag is invalid") + + if not self.prepared: + if self.phase is not TargetOperationPhase.SEALED or self.terminal_phase is not TargetTerminalPhase.ROLLED_BACK or self.handle is not None or self.preimage is not None or self.candidate is not None: + raise ValueError("unprepared target record is invalid") + return + + _validate_handle(self.handle) + if type(self.preimage) is not _OverlayState or type(self.candidate) is not _OverlayState: + raise ValueError("target record state is invalid") + _OverlayState.__post_init__(self.preimage) + _OverlayState.__post_init__(self.candidate) + if self.key.projection_digest != _config_digest(self.candidate.config): + raise ValueError("target record candidate digest is invalid") + if self.phase is TargetOperationPhase.SEALED: + if not isinstance(self.terminal_phase, TargetTerminalPhase): + raise ValueError("sealed target record lacks terminal phase") + elif self.terminal_phase is not None: + raise ValueError("unsealed target record has terminal phase") + + +class GeneratedOverlayWholeConfigTarget(WholeConfigTarget): + """Durable fenced target for one complete allowlisted generated overlay.""" + + def __init__(self, overlay_path, journal, allowed_keys): + """Bind an explicit overlay path, opaque journal, and immutable key set.""" + path = os.fspath(overlay_path) + if type(path) is not str or not path or "\x00" in path: + raise ValueError("generated overlay path is invalid") + self._path = os.path.abspath(path) + self._directory_path = os.path.dirname(self._path) + self._name = os.path.basename(self._path) + if not self._name or self._name in (".", "..") or self._name.lower() in _PROTECTED_FILENAMES or len(self._name.encode("utf-8")) > 128: + raise ValueError("generated overlay filename is invalid") + if not self._name.endswith(".json"): + raise ValueError("generated overlay must be an explicit JSON file") + if not callable(getattr(journal, "load", None)) or not callable(getattr(journal, "compare_and_store", None)): + raise ValueError("target journal does not implement load/CAS") + self._journal = journal + self._allowed_keys = self._validate_allowlist(allowed_keys) + self._target_binding = hashlib.sha256(os.fsencode(self._path)).hexdigest() + self._allowlist_binding = hashlib.sha256(_canonical_json(self._allowed_keys).encode("utf-8")).hexdigest() + self._lock_name = self._name + ".lock" + self._thread_lock = threading.RLock() + + os.makedirs( + self._directory_path, + mode=0o700, + exist_ok=True, + ) + if not hasattr(os, "O_DIRECTORY") or not hasattr(os, "O_NOFOLLOW") or not hasattr(os, "O_NONBLOCK") or os.open not in os.supports_dir_fd or os.stat not in os.supports_dir_fd or os.unlink not in os.supports_dir_fd: + raise WholeConfigTargetError("generated overlay requires anchored POSIX dir_fd support") + descriptor = self._open_directory() + os.close(descriptor) + + @staticmethod + def _validate_allowlist(allowed_keys): + """Return a sorted exact tuple of target-owned config keys.""" + if type(allowed_keys) not in (tuple, frozenset): + raise ValueError("generated overlay allowlist must be a tuple or frozenset") + keys = tuple(allowed_keys) + if not keys or any(type(key) is not str for key in keys): + raise ValueError("generated overlay allowlist is invalid") + canonical = _canonical_config( + {key: None for key in keys}, + "generated overlay allowlist", + ) + if len(canonical) != len(keys): + raise ValueError("generated overlay allowlist contains duplicates") + return tuple(sorted(canonical)) + + def __repr__(self): + """Expose no path, projection, pre-image, or journal payload.""" + return "GeneratedOverlayWholeConfigTarget(allowlisted_keys={!r})".format(len(self._allowed_keys)) + + def _open_directory(self): + """Open and validate the private non-symlink directory anchor.""" + flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_NONBLOCK + if hasattr(os, "O_CLOEXEC"): + flags |= os.O_CLOEXEC + try: + descriptor = os.open(self._directory_path, flags) + except OSError as exc: + raise WholeConfigTargetError("generated overlay directory is unavailable or unsafe") from exc + try: + metadata = os.fstat(descriptor) + path_metadata = os.stat( + self._directory_path, + follow_symlinks=False, + ) + if not stat.S_ISDIR(metadata.st_mode) or not stat.S_ISDIR(path_metadata.st_mode) or metadata.st_dev != path_metadata.st_dev or metadata.st_ino != path_metadata.st_ino or metadata.st_uid != os.geteuid() or stat.S_IMODE(metadata.st_mode) & 0o022: + raise WholeConfigTargetError("generated overlay directory must be private and owned") + except Exception: + os.close(descriptor) + raise + return descriptor + + def _open_lock(self, directory_fd): + """Open the target-wide mode-0600 exclusion record.""" + flags = os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW | os.O_NONBLOCK + if hasattr(os, "O_CLOEXEC"): + flags |= os.O_CLOEXEC + try: + descriptor = os.open( + self._lock_name, + flags, + 0o600, + dir_fd=directory_fd, + ) + except (NotImplementedError, OSError, TypeError) as exc: + raise WholeConfigTargetError("generated overlay lock could not be opened safely") from exc + try: + metadata = os.fstat(descriptor) + if not stat.S_ISREG(metadata.st_mode) or stat.S_IMODE(metadata.st_mode) != 0o600 or metadata.st_uid != os.geteuid() or metadata.st_nlink != 1: + raise WholeConfigTargetError("generated overlay lock must be a private regular file") + except Exception: + os.close(descriptor) + raise + return descriptor + + @contextmanager + def _exclusive(self): + """Serialize every journal/overlay transition across target instances.""" + with self._thread_lock: + directory_fd = self._open_directory() + try: + descriptor = self._open_lock(directory_fd) + try: + fcntl.flock(descriptor, fcntl.LOCK_EX) + yield directory_fd + finally: + try: + fcntl.flock(descriptor, fcntl.LOCK_UN) + finally: + os.close(descriptor) + finally: + os.close(directory_fd) + + @staticmethod + def _sync_directory(directory_fd): + """Fsync an anchored directory where supported.""" + try: + os.fsync(directory_fd) + except OSError as exc: + if exc.errno not in _SYNC_UNSUPPORTED: + raise + + def _validate_owned_config(self, value, name): + """Canonicalize one config and reject keys outside target ownership.""" + config = _canonical_config(value, name) + if any(key not in self._allowed_keys for key in config): + raise ValueError("{} contains a non-owned key".format(name)) + return config + + def _encode_overlay(self, config): + """Return the exact canonical complete overlay bytes.""" + config = self._validate_owned_config( + config, + "generated overlay", + ) + encoded = _canonical_json(config).encode("ascii") + if len(encoded) > MAX_CONFIG_BYTES: + raise ValueError("generated overlay exceeds its byte bound") + return encoded + + @staticmethod + def _unique_object(pairs): + """Reject duplicate JSON keys while decoding an overlay.""" + result = {} + for key, value in pairs: + if key in result: + raise ValueError("generated overlay contains duplicate keys") + result[key] = value + return result + + def _decode_overlay(self, encoded): + """Decode an exact canonical allowlisted overlay.""" + try: + decoded = json.loads( + encoded.decode("ascii"), + object_pairs_hook=self._unique_object, + parse_constant=lambda _value: (_ for _ in ()).throw(ValueError("non-finite JSON value")), + ) + config = self._validate_owned_config( + decoded, + "generated overlay", + ) + if self._encode_overlay(config) != encoded: + raise ValueError("generated overlay is not canonical") + return config + except ( + AttributeError, + RecursionError, + TypeError, + UnicodeError, + ValueError, + ) as exc: + raise WholeConfigTargetCorrupt("generated overlay content is invalid") from exc + + def _read_overlay(self, directory_fd): + """Read one bounded private regular overlay without following links.""" + flags = os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK + if hasattr(os, "O_CLOEXEC"): + flags |= os.O_CLOEXEC + try: + descriptor = os.open( + self._name, + flags, + dir_fd=directory_fd, + ) + except FileNotFoundError: + return _OverlayState( + False, + MappingProxyType({}), + ) + except (NotImplementedError, OSError, TypeError) as exc: + raise WholeConfigTargetError("generated overlay could not be opened safely") from exc + try: + metadata = os.fstat(descriptor) + if not stat.S_ISREG(metadata.st_mode) or stat.S_IMODE(metadata.st_mode) != 0o600 or metadata.st_uid != os.geteuid() or metadata.st_nlink != 1 or metadata.st_size > MAX_CONFIG_BYTES: + raise WholeConfigTargetCorrupt("generated overlay metadata is unsafe") + chunks = [] + remaining = MAX_CONFIG_BYTES + 1 + while remaining: + chunk = os.read( + descriptor, + min(remaining, 65536), + ) + if not chunk: + break + chunks.append(chunk) + remaining -= len(chunk) + encoded = b"".join(chunks) + if len(encoded) > MAX_CONFIG_BYTES: + raise WholeConfigTargetCorrupt("generated overlay exceeds its byte bound") + return _OverlayState( + True, + self._decode_overlay(encoded), + ) + finally: + os.close(descriptor) + + def _write_overlay(self, directory_fd, config): + """Atomically replace, fsync, and exactly read back the full overlay.""" + encoded = self._encode_overlay(config) + descriptor = None + temporary_name = None + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW | os.O_NONBLOCK + if hasattr(os, "O_CLOEXEC"): + flags |= os.O_CLOEXEC + for _attempt in range(32): + candidate = ".{}.{}.tmp".format( + self._name, + secrets.token_hex(12), + ) + try: + descriptor = os.open( + candidate, + flags, + 0o600, + dir_fd=directory_fd, + ) + except FileExistsError: + continue + except (NotImplementedError, OSError, TypeError) as exc: + raise WholeConfigTargetError("generated overlay temporary could not be created") from exc + temporary_name = candidate + break + if descriptor is None or temporary_name is None: + raise WholeConfigTargetError("generated overlay temporary allocation is exhausted") + try: + os.fchmod(descriptor, 0o600) + offset = 0 + while offset < len(encoded): + written = os.write( + descriptor, + encoded[offset:], + ) + if written <= 0: + raise WholeConfigTargetError("generated overlay write made no progress") + offset += written + os.fsync(descriptor) + closing_descriptor = descriptor + descriptor = None + os.close(closing_descriptor) + try: + os.replace( + temporary_name, + self._name, + src_dir_fd=directory_fd, + dst_dir_fd=directory_fd, + ) + except (NotImplementedError, OSError, TypeError) as exc: + raise WholeConfigTargetError("generated overlay replace failed") from exc + temporary_name = None + self._sync_directory(directory_fd) + readback = self._read_overlay(directory_fd) + if not readback.present or self._encode_overlay(readback.config) != encoded: + raise WholeConfigTargetError("generated overlay exact readback failed") + finally: + if descriptor is not None: + os.close(descriptor) + if temporary_name is not None: + try: + os.unlink( + temporary_name, + dir_fd=directory_fd, + ) + except (FileNotFoundError, OSError): + pass + + def _remove_overlay(self, directory_fd): + """Remove an overlay to restore an exact absent pre-image.""" + try: + os.unlink( + self._name, + dir_fd=directory_fd, + ) + except FileNotFoundError: + pass + except (NotImplementedError, OSError, TypeError) as exc: + raise WholeConfigTargetError("generated overlay removal failed") from exc + self._sync_directory(directory_fd) + if self._read_overlay(directory_fd).present: + raise WholeConfigTargetError("generated overlay absence readback failed") + + @staticmethod + def _state_equal(left, right): + """Compare exact visibility and type-exact canonical config.""" + return left.present == right.present and _configs_equal(left.config, right.config) + + @staticmethod + def _key_document(key): + """Return one canonical journal key document.""" + return { + "fence": key.fence, + "projection_digest": key.projection_digest, + } + + @staticmethod + def _state_document(state): + """Return one canonical secret-bearing overlay state document.""" + return { + "present": state.present, + "config": json.loads(_canonical_json(state.config)), + } + + def _record_document(self, record): + """Build the target-owned journal document.""" + return { + "format": TARGET_FORMAT, + "version": TARGET_VERSION, + "key": self._key_document(record.key), + "phase": record.phase.value, + "prepared": record.prepared, + "target_binding": record.target_binding, + "allowlist_binding": record.allowlist_binding, + "handle": record.handle, + "preimage": (None if record.preimage is None else self._state_document(record.preimage)), + "candidate": (None if record.candidate is None else self._state_document(record.candidate)), + "terminal_phase": (None if record.terminal_phase is None else record.terminal_phase.value), + } + + def _encode_record(self, record): + """Encode one exact target-private record without repr exposure.""" + _TargetRecord.__post_init__(record) + encoded = json.dumps( + self._record_document(record), + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + allow_nan=False, + ).encode("ascii") + if len(encoded) > MAX_OPAQUE_PAYLOAD_BYTES: + raise ValueError("target journal payload exceeds its byte bound") + return encoded + + @staticmethod + def _require_exact_keys(value, expected, name): + """Require one plain mapping with exactly the named fields.""" + if type(value) is not dict or set(value) != set(expected): + raise ValueError("{} fields are invalid".format(name)) + return value + + def _decode_key(self, value): + """Decode and reconstruct one exact target key.""" + value = self._require_exact_keys( + value, + ("fence", "projection_digest"), + "target key", + ) + key = TargetOperationKey( + value["fence"], + value["projection_digest"], + ) + TargetOperationKey.__post_init__(key) + return key + + def _decode_state(self, value, name): + """Decode and bind one exact allowlisted overlay state.""" + value = self._require_exact_keys( + value, + ("present", "config"), + name, + ) + state = _OverlayState( + value["present"], + self._validate_owned_config( + value["config"], + name, + ), + ) + _OverlayState.__post_init__(state) + return state + + def _decode_record(self, entry): + """Decode, reconstruct, and exactly re-encode one journal head.""" + if type(entry) is not OpaqueTargetJournalEntry: + raise WholeConfigTargetCorrupt("target journal entry type is invalid") + try: + document = json.loads( + entry.payload.decode("ascii"), + object_pairs_hook=self._unique_object, + parse_constant=lambda _value: (_ for _ in ()).throw(ValueError("non-finite JSON value")), + ) + document = self._require_exact_keys( + document, + ( + "format", + "version", + "key", + "phase", + "prepared", + "target_binding", + "allowlist_binding", + "handle", + "preimage", + "candidate", + "terminal_phase", + ), + "target journal", + ) + if document["format"] != TARGET_FORMAT or type(document["version"]) is not int or document["version"] != TARGET_VERSION: + raise ValueError("target journal format is invalid") + key = self._decode_key(document["key"]) + if key != entry.key: + raise ValueError("target journal key binding is invalid") + phase = TargetOperationPhase(document["phase"]) + terminal_phase = None if document["terminal_phase"] is None else TargetTerminalPhase(document["terminal_phase"]) + prepared = document["prepared"] + preimage = ( + None + if document["preimage"] is None + else self._decode_state( + document["preimage"], + "target preimage", + ) + ) + candidate = ( + None + if document["candidate"] is None + else self._decode_state( + document["candidate"], + "target candidate", + ) + ) + record = _TargetRecord( + key=key, + phase=phase, + prepared=prepared, + target_binding=document["target_binding"], + allowlist_binding=document["allowlist_binding"], + handle=document["handle"], + preimage=preimage, + candidate=candidate, + terminal_phase=terminal_phase, + ) + _TargetRecord.__post_init__(record) + if record.target_binding != self._target_binding or record.allowlist_binding != self._allowlist_binding or self._encode_record(record) != entry.payload: + raise ValueError("target journal binding is invalid") + return record + except ( + AttributeError, + KeyError, + RecursionError, + TypeError, + UnicodeError, + ValueError, + ) as exc: + raise WholeConfigTargetCorrupt("target journal payload is invalid") from exc + + def _load_head(self): + """Load and validate the current target-private journal head.""" + try: + entry = self._journal.load() + except Exception as exc: + raise WholeConfigTargetError("target journal load failed ({})".format(type(exc).__name__)) + if entry is None: + return None, None + return entry, self._decode_record(entry) + + def _store_record(self, expected, record): + """CAS exactly one next opaque revision.""" + self._require_revision_headroom(expected, 1) + revision = 1 if expected is None else expected.revision + 1 + try: + replacement = OpaqueTargetJournalEntry( + revision, + record.key, + self._encode_record(record), + ) + stored = self._journal.compare_and_store( + expected, + replacement, + ) + except Exception as exc: + raise WholeConfigTargetError("target journal CAS failed ({})".format(type(exc).__name__)) + if type(stored) is not bool: + raise WholeConfigTargetError("target journal CAS acknowledgement is invalid") + if not stored: + raise WholeConfigTargetConflict("target journal CAS lost an exact transition") + return replacement, record + + @staticmethod + def _require_revision_headroom(entry, steps): + """Require enough journal revisions for a complete remaining phase.""" + if type(steps) is not int or steps < 1: + raise ValueError("target journal headroom step count is invalid") + revision = 0 if entry is None else entry.revision + if type(revision) is not int or revision < 0 or revision > MAX_SEQUENCE - steps: + raise WholeConfigTargetError("target journal revision has insufficient lifecycle headroom") + + @staticmethod + def _snapshot(record): + """Return the secret-free protocol view of one private record.""" + if not record.prepared: + return TargetOperationSnapshot( + key=record.key, + phase=TargetOperationPhase.SEALED, + terminal_phase=TargetTerminalPhase.ROLLED_BACK, + ) + return TargetOperationSnapshot( + key=record.key, + phase=record.phase, + handle=record.handle, + preimage_digest=_config_digest(record.preimage.config), + candidate_digest=_config_digest(record.candidate.config), + changed=not _configs_equal( + record.preimage.config, + record.candidate.config, + ), + terminal_phase=record.terminal_phase, + ) + + @staticmethod + def _validate_operation(operation): + """Revalidate an exact secret-free snapshot from a caller.""" + if type(operation) is not TargetOperationSnapshot: + raise ValueError("target operation type is invalid") + TargetOperationSnapshot.__post_init__(operation) + return operation + + @staticmethod + def _same_identity(operation, record): + """Bind an opaque caller snapshot to the exact private operation.""" + current = GeneratedOverlayWholeConfigTarget._snapshot(record) + return ( + operation.key, + operation.handle, + operation.preimage_digest, + operation.candidate_digest, + operation.changed, + ) == ( + current.key, + current.handle, + current.preimage_digest, + current.candidate_digest, + current.changed, + ) + + def _transition(self, entry, record, phase, terminal_phase=None): + """Persist one exact lifecycle transition.""" + replacement = replace( + record, + phase=phase, + terminal_phase=terminal_phase, + ) + return self._store_record( + entry, + replacement, + ) + + def _reconcile_sealed(self, directory_fd, entry, record): + """Re-prove a sealed terminal result against exact current visibility.""" + if record.phase is not TargetOperationPhase.SEALED: + raise ValueError("target record is not sealed") + if not record.prepared or record.terminal_phase is TargetTerminalPhase.UNKNOWN: + return entry, record + try: + current = self._read_overlay(directory_fd) + expected = record.candidate if record.terminal_phase is TargetTerminalPhase.APPLIED else record.preimage + matches_terminal = self._state_equal(current, expected) + except Exception: + matches_terminal = False + if matches_terminal: + return entry, record + self._require_revision_headroom(entry, 1) + return self._transition( + entry, + record, + TargetOperationPhase.SEALED, + TargetTerminalPhase.UNKNOWN, + ) + + def _reconcile_stable(self, directory_fd, entry, record): + """Re-prove any stable phase immediately before acknowledging it.""" + if record.phase is TargetOperationPhase.SEALED: + return self._reconcile_sealed( + directory_fd, + entry, + record, + ) + if record.phase not in ( + TargetOperationPhase.PREPARED, + TargetOperationPhase.APPLIED, + TargetOperationPhase.ROLLED_BACK, + ): + return entry, record + expected = record.candidate if record.phase is TargetOperationPhase.APPLIED else record.preimage + try: + current = self._read_overlay(directory_fd) + matches_stable = self._state_equal(current, expected) + except Exception: + matches_stable = False + if matches_stable: + return entry, record + self._require_revision_headroom(entry, 1) + return self._transition( + entry, + record, + TargetOperationPhase.UNKNOWN, + ) + + def prepare(self, fence, projection_digest, projection): + """Capture an exact pre-image after a winning materializer reservation.""" + key = TargetOperationKey( + fence, + projection_digest, + ) + TargetOperationKey.__post_init__(key) + candidate_config = self._validate_owned_config( + projection, + "target projection", + ) + if _config_digest(candidate_config) != projection_digest: + raise ValueError("target projection digest is invalid") + + with self._exclusive() as directory_fd: + entry, current = self._load_head() + if current is not None: + if key.fence <= current.key.fence: + raise WholeConfigTargetConflict("target prepare fence is stale or replayed") + if current.phase is not TargetOperationPhase.SEALED: + raise WholeConfigTargetConflict("target prepare cannot replace an unsettled operation") + entry, current = self._reconcile_sealed( + directory_fd, + entry, + current, + ) + if current.terminal_phase is TargetTerminalPhase.UNKNOWN: + raise WholeConfigTargetConflict("target prepare cannot replace an unknown operation") + self._require_revision_headroom(entry, 5) + preimage = self._read_overlay(directory_fd) + if _configs_equal(preimage.config, candidate_config): + candidate = _OverlayState( + preimage.present, + candidate_config, + ) + else: + candidate = _OverlayState( + True, + candidate_config, + ) + record = _TargetRecord( + key=key, + phase=TargetOperationPhase.PREPARED, + prepared=True, + target_binding=self._target_binding, + allowlist_binding=self._allowlist_binding, + handle="lattice-target-{}-{}".format( + key.fence, + key.projection_digest[:24], + ), + preimage=preimage, + candidate=candidate, + ) + entry, record = self._store_record( + entry, + record, + ) + _entry, record = self._reconcile_stable( + directory_fd, + entry, + record, + ) + return self._snapshot(record) + + def lookup(self, fence, projection_digest): + """Recover only the current exact reservation key.""" + key = TargetOperationKey( + fence, + projection_digest, + ) + TargetOperationKey.__post_init__(key) + with self._exclusive() as directory_fd: + entry, record = self._load_head() + if record is None or record.key != key: + return None + _entry, record = self._reconcile_stable( + directory_fd, + entry, + record, + ) + return self._snapshot(record) + + def apply(self, operation): + """Atomically replace the complete overlay behind a durable apply phase.""" + operation = self._validate_operation(operation) + with self._exclusive() as directory_fd: + entry, record = self._load_head() + if record is None or record.key != operation.key: + return operation + if not self._same_identity(operation, record): + raise WholeConfigTargetConflict("target apply operation identity is invalid") + if record.phase is not TargetOperationPhase.PREPARED: + _entry, record = self._reconcile_stable( + directory_fd, + entry, + record, + ) + return self._snapshot(record) + self._require_revision_headroom(entry, 4) + entry, record = self._transition( + entry, + record, + TargetOperationPhase.APPLYING, + ) + current = self._read_overlay(directory_fd) + if not self._state_equal(current, record.preimage): + _entry, record = self._transition( + entry, + record, + TargetOperationPhase.UNKNOWN, + ) + return self._snapshot(record) + if not self._state_equal(record.preimage, record.candidate): + self._write_overlay( + directory_fd, + record.candidate.config, + ) + readback = self._read_overlay(directory_fd) + if not self._state_equal(readback, record.candidate): + _entry, record = self._transition( + entry, + record, + TargetOperationPhase.UNKNOWN, + ) + return self._snapshot(record) + entry, record = self._transition( + entry, + record, + TargetOperationPhase.APPLIED, + ) + _entry, record = self._reconcile_stable( + directory_fd, + entry, + record, + ) + return self._snapshot(record) + + def rollback(self, operation): + """Restore the exact pre-image of the current sealed APPLIED operation.""" + operation = self._validate_operation(operation) + with self._exclusive() as directory_fd: + entry, record = self._load_head() + if record is None or record.key != operation.key: + return operation + if not self._same_identity(operation, record): + raise WholeConfigTargetConflict("target rollback operation identity is invalid") + if record.phase is not TargetOperationPhase.SEALED or record.terminal_phase is not TargetTerminalPhase.APPLIED: + _entry, record = self._reconcile_stable( + directory_fd, + entry, + record, + ) + return self._snapshot(record) + entry, record = self._reconcile_sealed( + directory_fd, + entry, + record, + ) + if record.terminal_phase is not TargetTerminalPhase.APPLIED: + return self._snapshot(record) + self._require_revision_headroom(entry, 4) + entry, record = self._transition( + entry, + record, + TargetOperationPhase.ROLLING_BACK, + ) + current = self._read_overlay(directory_fd) + if not self._state_equal(current, record.candidate): + _entry, record = self._transition( + entry, + record, + TargetOperationPhase.UNKNOWN, + ) + return self._snapshot(record) + if not self._state_equal(record.preimage, record.candidate): + if record.preimage.present: + self._write_overlay( + directory_fd, + record.preimage.config, + ) + else: + self._remove_overlay(directory_fd) + readback = self._read_overlay(directory_fd) + if not self._state_equal(readback, record.preimage): + _entry, record = self._transition( + entry, + record, + TargetOperationPhase.UNKNOWN, + ) + return self._snapshot(record) + entry, record = self._transition( + entry, + record, + TargetOperationPhase.ROLLED_BACK, + ) + _entry, record = self._reconcile_stable( + directory_fd, + entry, + record, + ) + return self._snapshot(record) + + def seal(self, operation): + """Seal one stable exact operation and reject later apply replay.""" + operation = self._validate_operation(operation) + with self._exclusive() as directory_fd: + entry, record = self._load_head() + if record is None or record.key != operation.key: + raise WholeConfigTargetConflict("target seal operation is not current") + if not self._same_identity(operation, record): + raise WholeConfigTargetConflict("target seal operation identity is invalid") + if record.phase is TargetOperationPhase.SEALED: + _entry, record = self._reconcile_sealed( + directory_fd, + entry, + record, + ) + return self._snapshot(record) + if record.phase is TargetOperationPhase.APPLIED: + expected = record.candidate + success_terminal = TargetTerminalPhase.APPLIED + elif record.phase in ( + TargetOperationPhase.PREPARED, + TargetOperationPhase.ROLLED_BACK, + ): + expected = record.preimage + success_terminal = TargetTerminalPhase.ROLLED_BACK + else: + raise WholeConfigTargetConflict("target seal operation is unstable") + try: + current = self._read_overlay(directory_fd) + terminal = success_terminal if self._state_equal(current, expected) else TargetTerminalPhase.UNKNOWN + except Exception: + terminal = TargetTerminalPhase.UNKNOWN + self._require_revision_headroom(entry, 2) + entry, record = self._transition( + entry, + record, + TargetOperationPhase.SEALED, + terminal, + ) + _entry, record = self._reconcile_sealed( + directory_fd, + entry, + record, + ) + return self._snapshot(record) + + def _terminal_visibility(self, directory_fd, record): + """Classify one unsettled record from exact durable visibility.""" + try: + current = self._read_overlay(directory_fd) + except Exception: + return TargetTerminalPhase.UNKNOWN + matches_preimage = self._state_equal( + current, + record.preimage, + ) + matches_candidate = self._state_equal( + current, + record.candidate, + ) + + if record.phase is TargetOperationPhase.PREPARED: + return TargetTerminalPhase.ROLLED_BACK if matches_preimage else TargetTerminalPhase.UNKNOWN + if record.phase is TargetOperationPhase.UNKNOWN: + return TargetTerminalPhase.UNKNOWN + if record.phase is TargetOperationPhase.APPLYING: + if matches_candidate and (not matches_preimage or record.phase is TargetOperationPhase.APPLYING): + return TargetTerminalPhase.APPLIED + if matches_preimage: + return TargetTerminalPhase.ROLLED_BACK + return TargetTerminalPhase.UNKNOWN + if record.phase is TargetOperationPhase.APPLIED: + return TargetTerminalPhase.APPLIED if matches_candidate else TargetTerminalPhase.UNKNOWN + if record.phase is TargetOperationPhase.ROLLING_BACK: + if matches_preimage: + return TargetTerminalPhase.ROLLED_BACK + if matches_candidate: + return TargetTerminalPhase.APPLIED + return TargetTerminalPhase.UNKNOWN + if record.phase is TargetOperationPhase.ROLLED_BACK: + return TargetTerminalPhase.ROLLED_BACK if matches_preimage else TargetTerminalPhase.UNKNOWN + return TargetTerminalPhase.UNKNOWN + + def _invalidate_current(self, directory_fd, entry, record): + """Fence and seal the current record from exact overlay visibility.""" + if record.phase is TargetOperationPhase.SEALED: + _entry, record = self._reconcile_sealed( + directory_fd, + entry, + record, + ) + return self._snapshot(record) + if not record.prepared: + return self._snapshot(record) + terminal = self._terminal_visibility( + directory_fd, + record, + ) + terminal_headroom = 1 if terminal is TargetTerminalPhase.UNKNOWN else 2 + self._require_revision_headroom(entry, terminal_headroom) + entry, record = self._transition( + entry, + record, + TargetOperationPhase.SEALED, + terminal, + ) + _entry, record = self._reconcile_sealed( + directory_fd, + entry, + record, + ) + return self._snapshot(record) + + def _tombstone(self, key): + """Build a durable unprepared ROLLED_BACK fence.""" + return _TargetRecord( + key=key, + phase=TargetOperationPhase.SEALED, + prepared=False, + target_binding=self._target_binding, + allowlist_binding=self._allowlist_binding, + terminal_phase=TargetTerminalPhase.ROLLED_BACK, + ) + + def abort(self, fence, projection_digest): + """Tombstone an unprepared reservation or invalidate its current record.""" + key = TargetOperationKey( + fence, + projection_digest, + ) + TargetOperationKey.__post_init__(key) + with self._exclusive() as directory_fd: + entry, record = self._load_head() + if record is not None and record.key == key: + return self._invalidate_current( + directory_fd, + entry, + record, + ) + if record is not None and key.fence < record.key.fence: + raise WholeConfigTargetConflict("target abort key is no longer current") + tombstone = self._tombstone(key) + self._store_record( + entry, + tombstone, + ) + return self._snapshot(tombstone) + + def invalidate(self, operation_key): + """Fence one current operation and classify only exact durable visibility.""" + if type(operation_key) is not TargetOperationKey: + raise ValueError("target invalidation key type is invalid") + TargetOperationKey.__post_init__(operation_key) + with self._exclusive() as directory_fd: + entry, record = self._load_head() + if record is not None and record.key == operation_key: + return self._invalidate_current( + directory_fd, + entry, + record, + ) + if record is not None and operation_key.fence < record.key.fence: + raise WholeConfigTargetConflict("target invalidation key is no longer current") + tombstone = self._tombstone(operation_key) + self._store_record( + entry, + tombstone, + ) + return self._snapshot(tombstone) diff --git a/apps/predbat/tests/test_lattice_whole_config_target.py b/apps/predbat/tests/test_lattice_whole_config_target.py new file mode 100644 index 000000000..faaf9030a --- /dev/null +++ b/apps/predbat/tests/test_lattice_whole_config_target.py @@ -0,0 +1,1839 @@ +"""Tests for the detached Lattice generated-overlay whole-config target.""" + +# cspell:ignore autoconfig mispaired Noncanonical readback recoverably + +import json +import os +import stat +import sys +import tempfile +import threading +import unittest +from dataclasses import replace +from unittest import mock + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from lattice_atomic_materializer import ( # noqa: E402 + MAX_SEQUENCE, + TargetOperationKey, + TargetOperationPhase, + TargetTerminalPhase, + _canonical_config, + _config_digest, +) +from lattice_durable_runtime_stores import ( # noqa: E402 + FileOpaqueTargetJournal, + OpaqueTargetJournalEntry, +) +from lattice_whole_config_target import ( # noqa: E402 + GeneratedOverlayWholeConfigTarget, + WholeConfigTargetConflict, + WholeConfigTargetCorrupt, + WholeConfigTargetError, +) + + +ALLOWED_KEYS = ( + "bar", + "enabled", + "foo", + "list_value", + "none_value", +) + + +def digest(config): + """Return the materializer's exact canonical projection digest.""" + return _config_digest( + _canonical_config( + config, + "test projection", + ) + ) + + +def exact_overlay(path, config): + """Create one private exact canonical JSON overlay.""" + with open(path, "wb") as overlay: + overlay.write( + json.dumps( + config, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + allow_nan=False, + ).encode("ascii") + ) + os.chmod(path, 0o600) + + +def read_overlay(path): + """Read a test overlay as plain JSON.""" + with open(path, "rb") as overlay: + encoded = overlay.read() + return encoded, json.loads(encoded.decode("ascii")) + + +class FailingWriteTarget(GeneratedOverlayWholeConfigTarget): + """Target with deterministic failures around the atomic replace.""" + + def __init__(self, *args, **kwargs): + """Create a target whose write fault defaults to disabled.""" + super().__init__(*args, **kwargs) + self.fail_before_write = False + self.fail_after_write = False + self.fail_before_remove = False + self.fail_after_remove = False + + def _write_overlay(self, directory_fd, config): + """Cut before or after the complete replace.""" + if self.fail_before_write: + raise OSError("secret-before-write") + super()._write_overlay(directory_fd, config) + if self.fail_after_write: + raise OSError("secret-after-write") + + def _remove_overlay(self, directory_fd): + """Cut before or after restoring exact absence.""" + if self.fail_before_remove: + raise OSError("secret-before-remove") + super()._remove_overlay(directory_fd) + if self.fail_after_remove: + raise OSError("secret-after-remove") + + +class BlockingWriteTarget(GeneratedOverlayWholeConfigTarget): + """Target that pauses while holding its target-wide outer lock.""" + + def __init__(self, *args, **kwargs): + """Create write coordination events.""" + super().__init__(*args, **kwargs) + self.entered = threading.Event() + self.release = threading.Event() + + def _write_overlay(self, directory_fd, config): + """Block immediately before the atomic full-file write.""" + self.entered.set() + if not self.release.wait(5): + raise RuntimeError("test write release timed out") + super()._write_overlay(directory_fd, config) + + +class RejectingJournal: + """Journal wrapper that rejects one exact compare-and-store call.""" + + def __init__(self, delegate, reject_call): + """Wrap a real durable journal.""" + self.delegate = delegate + self.reject_call = reject_call + self.calls = 0 + + def load(self): + """Delegate exact load.""" + return self.delegate.load() + + def compare_and_store(self, expected, replacement): + """Reject the selected CAS before any replacement.""" + self.calls += 1 + if self.calls == self.reject_call: + return False + return self.delegate.compare_and_store( + expected, + replacement, + ) + + +class BoundaryJournal: + """Exact in-memory journal whose seed revision can sit near exhaustion.""" + + def __init__(self, entry): + """Create a journal at one exact current entry.""" + self.entry = entry + self.writes = 0 + + def load(self): + """Return the exact current entry.""" + return self.entry + + def compare_and_store(self, expected, replacement): + """CAS exact identity and one next bounded revision.""" + if self.entry != expected: + return False + if replacement.revision != expected.revision + 1: + raise ValueError("boundary journal revision is not next") + self.entry = replacement + self.writes += 1 + return True + + +class MutatingJournal: + """Journal wrapper that mutates the overlay after selected successful CAS calls.""" + + def __init__(self, delegate, mutations): + """Wrap a journal with call-numbered post-store callbacks.""" + self.delegate = delegate + self.mutations = mutations + self.calls = 0 + + def load(self): + """Delegate exact load.""" + return self.delegate.load() + + def compare_and_store(self, expected, replacement): + """Run the selected mutation only after the replacement is durable.""" + stored = self.delegate.compare_and_store( + expected, + replacement, + ) + self.calls += 1 + mutation = self.mutations.get(self.calls) + if stored and mutation is not None: + mutation() + return stored + + +class GeneratedOverlayTargetTestCase(unittest.TestCase): + """Exercise complete target lifecycle, recovery, and filesystem safety.""" + + def setUp(self): + """Create an isolated target with a real filesystem journal.""" + self.temporary = tempfile.TemporaryDirectory() + self.addCleanup(self.temporary.cleanup) + self.overlay_path = os.path.join( + self.temporary.name, + "lattice-generated-overlay.json", + ) + self.journal_path = os.path.join( + self.temporary.name, + "lattice-target.journal", + ) + self.journal = FileOpaqueTargetJournal( + self.journal_path, + ) + self.target = GeneratedOverlayWholeConfigTarget( + self.overlay_path, + self.journal, + ALLOWED_KEYS, + ) + + def prepare(self, config, fence=1, target=None): + """Prepare one exact projection.""" + target = self.target if target is None else target + return target.prepare( + fence, + digest(config), + config, + ) + + def boundary_target(self, revision): + """Rebind the current exact payload to a chosen journal revision.""" + entry = self.journal.load() + boundary = BoundaryJournal( + OpaqueTargetJournalEntry( + revision, + entry.key, + entry.payload, + ) + ) + target = GeneratedOverlayWholeConfigTarget( + self.overlay_path, + boundary, + ALLOWED_KEYS, + ) + return target, boundary + + def test_full_overlay_apply_restart_and_exact_absent_rollback(self): + """Apply replaces the complete file and rollback restores absence.""" + projection = { + "enabled": True, + "foo": 7, + "list_value": ["sensor.one", None], + } + prepared = self.prepare(projection) + + self.assertFalse(os.path.exists(self.overlay_path)) + self.assertEqual( + prepared.phase, + TargetOperationPhase.PREPARED, + ) + self.assertEqual( + prepared.preimage_digest, + digest({}), + ) + self.assertEqual( + prepared.candidate_digest, + digest(projection), + ) + self.assertTrue(prepared.changed) + self.assertNotIn("sensor.one", repr(self.journal.load())) + self.assertNotIn("sensor.one", repr(self.target)) + + applied = self.target.apply(prepared) + self.assertEqual( + applied.phase, + TargetOperationPhase.APPLIED, + ) + encoded, observed = read_overlay(self.overlay_path) + self.assertEqual(observed, projection) + self.assertEqual( + encoded, + b'{"enabled":true,"foo":7,"list_value":["sensor.one",null]}', + ) + self.assertEqual( + stat.S_IMODE(os.stat(self.overlay_path).st_mode), + 0o600, + ) + + sealed = self.target.seal(applied) + self.assertEqual( + sealed.terminal_phase, + TargetTerminalPhase.APPLIED, + ) + restarted = GeneratedOverlayWholeConfigTarget( + self.overlay_path, + FileOpaqueTargetJournal(self.journal_path), + tuple(reversed(ALLOWED_KEYS)), + ) + self.assertEqual( + restarted.lookup(1, digest(projection)), + sealed, + ) + + rolled_back = restarted.rollback(sealed) + self.assertEqual( + rolled_back.phase, + TargetOperationPhase.ROLLED_BACK, + ) + self.assertFalse(os.path.exists(self.overlay_path)) + resealed = restarted.seal(rolled_back) + self.assertEqual( + resealed.terminal_phase, + TargetTerminalPhase.ROLLED_BACK, + ) + + def test_apply_is_complete_replacement_and_rollback_is_exact_preimage(self): + """Keys absent from the candidate are removed, then restored.""" + preimage = { + "bar": 2, + "foo": 1, + "none_value": None, + } + exact_overlay( + self.overlay_path, + preimage, + ) + before = read_overlay(self.overlay_path)[0] + prepared = self.prepare( + { + "foo": 9, + } + ) + applied = self.target.apply(prepared) + + self.assertEqual( + read_overlay(self.overlay_path)[1], + {"foo": 9}, + ) + sealed = self.target.seal(applied) + rolled_back = self.target.rollback(sealed) + self.assertEqual( + rolled_back.phase, + TargetOperationPhase.ROLLED_BACK, + ) + self.assertEqual( + read_overlay(self.overlay_path)[0], + before, + ) + + def test_unchanged_candidate_preserves_exact_file_identity(self): + """A type-exact unchanged full config performs no overlay write.""" + projection = { + "enabled": False, + "foo": 4, + } + exact_overlay( + self.overlay_path, + projection, + ) + before = os.stat(self.overlay_path) + prepared = self.prepare(projection) + + self.assertFalse(prepared.changed) + applied = self.target.apply(prepared) + after = os.stat(self.overlay_path) + self.assertEqual( + (after.st_dev, after.st_ino, after.st_mtime_ns), + (before.st_dev, before.st_ino, before.st_mtime_ns), + ) + self.assertEqual( + self.target.seal(applied).terminal_phase, + TargetTerminalPhase.APPLIED, + ) + + def test_absent_and_empty_projection_is_idempotent_without_file_creation(self): + """Semantic empty-over-empty keeps the exact absent pre-image.""" + prepared = self.prepare({}) + self.assertFalse(prepared.changed) + + applied = self.target.apply(prepared) + self.assertFalse(os.path.exists(self.overlay_path)) + sealed = self.target.seal(applied) + rolled_back = self.target.rollback(sealed) + self.assertFalse(os.path.exists(self.overlay_path)) + self.assertEqual( + rolled_back.phase, + TargetOperationPhase.ROLLED_BACK, + ) + + def test_external_mutation_after_prepare_fails_unknown_without_overwrite(self): + """A fenced pre-image mismatch is quarantined, never blindly replaced.""" + exact_overlay( + self.overlay_path, + {"foo": 1}, + ) + prepared = self.prepare({"foo": 2}) + exact_overlay( + self.overlay_path, + {"foo": 99}, + ) + + observed = self.target.apply(prepared) + self.assertEqual( + observed.phase, + TargetOperationPhase.UNKNOWN, + ) + self.assertEqual( + read_overlay(self.overlay_path)[1], + {"foo": 99}, + ) + sealed = self.target.invalidate(prepared.key) + self.assertEqual( + sealed.terminal_phase, + TargetTerminalPhase.UNKNOWN, + ) + self.assertEqual( + read_overlay(self.overlay_path)[1], + {"foo": 99}, + ) + + def test_external_mutation_after_apply_blocks_rollback(self): + """Rollback never overwrites a candidate mismatch.""" + exact_overlay( + self.overlay_path, + {"foo": 1}, + ) + prepared = self.prepare({"foo": 2}) + sealed = self.target.seal(self.target.apply(prepared)) + exact_overlay( + self.overlay_path, + {"foo": 77}, + ) + + observed = self.target.rollback(sealed) + self.assertEqual( + observed.phase, + TargetOperationPhase.SEALED, + ) + self.assertEqual( + observed.terminal_phase, + TargetTerminalPhase.UNKNOWN, + ) + self.assertEqual( + read_overlay(self.overlay_path)[1], + {"foo": 77}, + ) + quarantined = self.target.invalidate(sealed.key) + self.assertEqual( + quarantined.terminal_phase, + TargetTerminalPhase.UNKNOWN, + ) + + def test_seal_rechecks_applied_visibility_after_mutation(self): + """APPLIED cannot be sealed after its exact candidate disappears.""" + exact_overlay( + self.overlay_path, + {"foo": 1}, + ) + prepared = self.prepare({"foo": 2}) + applied = self.target.apply(prepared) + exact_overlay( + self.overlay_path, + {"foo": 88}, + ) + + sealed = self.target.seal(applied) + self.assertEqual( + sealed.phase, + TargetOperationPhase.SEALED, + ) + self.assertEqual( + sealed.terminal_phase, + TargetTerminalPhase.UNKNOWN, + ) + self.assertEqual( + read_overlay(self.overlay_path)[1], + {"foo": 88}, + ) + + def test_seal_rechecks_prepared_and_rolled_back_visibility(self): + """PREPARED/ROLLED_BACK seal only while the exact pre-image remains.""" + exact_overlay( + self.overlay_path, + {"foo": 1}, + ) + prepared = self.prepare({"foo": 2}) + exact_overlay( + self.overlay_path, + {"foo": 77}, + ) + sealed_prepared = self.target.seal(prepared) + self.assertEqual( + sealed_prepared.terminal_phase, + TargetTerminalPhase.UNKNOWN, + ) + + # Exercise the ROLLED_BACK boundary in an independent journal. + with tempfile.TemporaryDirectory() as directory: + overlay_path = os.path.join(directory, "generated.json") + journal_path = os.path.join(directory, "target.journal") + exact_overlay(overlay_path, {"foo": 1}) + target = GeneratedOverlayWholeConfigTarget( + overlay_path, + FileOpaqueTargetJournal(journal_path), + ALLOWED_KEYS, + ) + projection = {"foo": 2} + operation = target.prepare( + 1, + digest(projection), + projection, + ) + applied = target.apply(operation) + sealed = target.seal(applied) + rolled_back = target.rollback(sealed) + exact_overlay(overlay_path, {"foo": 99}) + sealed_rollback = target.seal(rolled_back) + self.assertEqual( + sealed_rollback.terminal_phase, + TargetTerminalPhase.UNKNOWN, + ) + self.assertEqual( + read_overlay(overlay_path)[1], + {"foo": 99}, + ) + + def test_apply_rechecks_visibility_after_applied_cas(self): + """A mutation after the APPLIED CAS is durably downgraded to UNKNOWN.""" + exact_overlay( + self.overlay_path, + {"foo": 1}, + ) + prepared = self.prepare({"foo": 2}) + mutating = MutatingJournal( + FileOpaqueTargetJournal(self.journal_path), + { + 2: lambda: exact_overlay( + self.overlay_path, + {"foo": 77}, + ) + }, + ) + target = GeneratedOverlayWholeConfigTarget( + self.overlay_path, + mutating, + ALLOWED_KEYS, + ) + + observed = target.apply(prepared) + + self.assertEqual( + observed.phase, + TargetOperationPhase.UNKNOWN, + ) + self.assertEqual(mutating.calls, 3) + self.assertEqual( + read_overlay(self.overlay_path)[1], + {"foo": 77}, + ) + + def test_prepare_rechecks_visibility_after_prepared_cas_mutation(self): + """A mutation after PREPARED CAS is durably downgraded to UNKNOWN.""" + exact_overlay( + self.overlay_path, + {"foo": 1}, + ) + mutating = MutatingJournal( + FileOpaqueTargetJournal(self.journal_path), + { + 1: lambda: exact_overlay( + self.overlay_path, + {"foo": 77}, + ) + }, + ) + target = GeneratedOverlayWholeConfigTarget( + self.overlay_path, + mutating, + ALLOWED_KEYS, + ) + + observed = self.prepare( + {"foo": 2}, + target=target, + ) + + self.assertEqual( + observed.phase, + TargetOperationPhase.UNKNOWN, + ) + self.assertEqual(mutating.calls, 2) + self.assertEqual( + read_overlay(self.overlay_path)[1], + {"foo": 77}, + ) + + def test_prepare_rechecks_read_failure_after_prepared_cas(self): + """A read failure after PREPARED CAS is durably UNKNOWN, not acknowledged.""" + exact_overlay( + self.overlay_path, + {"foo": 1}, + ) + original_read = self.target._read_overlay + reads = [0] + + def fail_second_read(directory_fd): + reads[0] += 1 + if reads[0] == 2: + raise WholeConfigTargetCorrupt("injected-post-prepare-read") + return original_read(directory_fd) + + with mock.patch.object( + self.target, + "_read_overlay", + side_effect=fail_second_read, + ): + observed = self.prepare({"foo": 2}) + + self.assertEqual( + observed.phase, + TargetOperationPhase.UNKNOWN, + ) + self.assertEqual(self.journal.load().revision, 2) + + def test_rollback_rechecks_visibility_after_rolled_back_cas(self): + """A mutation after ROLLED_BACK CAS is durably downgraded to UNKNOWN.""" + exact_overlay( + self.overlay_path, + {"foo": 1}, + ) + prepared = self.prepare({"foo": 2}) + sealed = self.target.seal(self.target.apply(prepared)) + mutating = MutatingJournal( + FileOpaqueTargetJournal(self.journal_path), + { + 2: lambda: exact_overlay( + self.overlay_path, + {"foo": 88}, + ) + }, + ) + target = GeneratedOverlayWholeConfigTarget( + self.overlay_path, + mutating, + ALLOWED_KEYS, + ) + + observed = target.rollback(sealed) + + self.assertEqual( + observed.phase, + TargetOperationPhase.UNKNOWN, + ) + self.assertEqual(mutating.calls, 3) + self.assertEqual( + read_overlay(self.overlay_path)[1], + {"foo": 88}, + ) + + def test_seal_rechecks_visibility_after_sealed_cas(self): + """A mutation after SEALED CAS consumes the reserved UNKNOWN revision.""" + exact_overlay( + self.overlay_path, + {"foo": 1}, + ) + prepared = self.prepare({"foo": 2}) + applied = self.target.apply(prepared) + mutating = MutatingJournal( + FileOpaqueTargetJournal(self.journal_path), + { + 1: lambda: exact_overlay( + self.overlay_path, + {"foo": 99}, + ) + }, + ) + target = GeneratedOverlayWholeConfigTarget( + self.overlay_path, + mutating, + ALLOWED_KEYS, + ) + + observed = target.seal(applied) + + self.assertEqual( + observed.terminal_phase, + TargetTerminalPhase.UNKNOWN, + ) + self.assertEqual(mutating.calls, 2) + self.assertEqual( + read_overlay(self.overlay_path)[1], + {"foo": 99}, + ) + + def test_invalidate_rechecks_visibility_after_sealed_cas(self): + """A mutation after invalidate's SEALED CAS is durably UNKNOWN.""" + exact_overlay( + self.overlay_path, + {"foo": 1}, + ) + prepared = self.prepare({"foo": 2}) + mutating = MutatingJournal( + FileOpaqueTargetJournal(self.journal_path), + { + 1: lambda: exact_overlay( + self.overlay_path, + {"foo": 66}, + ) + }, + ) + target = GeneratedOverlayWholeConfigTarget( + self.overlay_path, + mutating, + ALLOWED_KEYS, + ) + + observed = target.invalidate(prepared.key) + + self.assertEqual( + observed.terminal_phase, + TargetTerminalPhase.UNKNOWN, + ) + self.assertEqual(mutating.calls, 2) + self.assertEqual( + read_overlay(self.overlay_path)[1], + {"foo": 66}, + ) + + def test_post_stable_cas_read_errors_are_durably_unknown(self): + """APPLIED and ROLLED_BACK post-CAS read errors cannot be acknowledged.""" + exact_overlay( + self.overlay_path, + {"foo": 1}, + ) + prepared = self.prepare({"foo": 2}) + original_read = self.target._read_overlay + apply_reads = [0] + + def fail_fourth_apply_read(directory_fd): + apply_reads[0] += 1 + if apply_reads[0] == 4: + raise WholeConfigTargetCorrupt("injected-post-applied-read") + return original_read(directory_fd) + + with mock.patch.object( + self.target, + "_read_overlay", + side_effect=fail_fourth_apply_read, + ): + applied_unknown = self.target.apply(prepared) + self.assertEqual( + applied_unknown.phase, + TargetOperationPhase.UNKNOWN, + ) + + with tempfile.TemporaryDirectory() as directory: + overlay_path = os.path.join(directory, "generated.json") + journal_path = os.path.join(directory, "target.journal") + exact_overlay(overlay_path, {"foo": 1}) + target = GeneratedOverlayWholeConfigTarget( + overlay_path, + FileOpaqueTargetJournal(journal_path), + ALLOWED_KEYS, + ) + projection = {"foo": 2} + operation = target.prepare( + 1, + digest(projection), + projection, + ) + sealed = target.seal(target.apply(operation)) + original_rollback_read = target._read_overlay + rollback_reads = [0] + + def fail_fifth_rollback_read(directory_fd): + rollback_reads[0] += 1 + if rollback_reads[0] == 5: + raise WholeConfigTargetCorrupt("injected-post-rollback-read") + return original_rollback_read(directory_fd) + + with mock.patch.object( + target, + "_read_overlay", + side_effect=fail_fifth_rollback_read, + ): + rolled_back_unknown = target.rollback(sealed) + self.assertEqual( + rolled_back_unknown.phase, + TargetOperationPhase.UNKNOWN, + ) + + def test_restart_lookup_downgrades_stale_sealed_applied_to_unknown(self): + """A restart never blindly acknowledges a stale sealed APPLIED result.""" + prepared = self.prepare({"foo": 2}) + sealed = self.target.seal(self.target.apply(prepared)) + self.assertEqual( + sealed.terminal_phase, + TargetTerminalPhase.APPLIED, + ) + os.unlink(self.overlay_path) + + restarted = GeneratedOverlayWholeConfigTarget( + self.overlay_path, + FileOpaqueTargetJournal(self.journal_path), + ALLOWED_KEYS, + ) + observed = restarted.lookup( + sealed.key.fence, + sealed.key.projection_digest, + ) + self.assertEqual( + observed.terminal_phase, + TargetTerminalPhase.UNKNOWN, + ) + recovered = restarted.invalidate(sealed.key) + self.assertEqual( + recovered.terminal_phase, + TargetTerminalPhase.UNKNOWN, + ) + + def test_restart_invalidate_downgrades_mutated_sealed_applied(self): + """Invalidate also re-proves sealed APPLIED visibility after restart.""" + prepared = self.prepare({"foo": 2}) + sealed = self.target.seal(self.target.apply(prepared)) + exact_overlay( + self.overlay_path, + {"foo": 55}, + ) + restarted = GeneratedOverlayWholeConfigTarget( + self.overlay_path, + FileOpaqueTargetJournal(self.journal_path), + ALLOWED_KEYS, + ) + + observed = restarted.invalidate(sealed.key) + self.assertEqual( + observed.terminal_phase, + TargetTerminalPhase.UNKNOWN, + ) + self.assertEqual( + read_overlay(self.overlay_path)[1], + {"foo": 55}, + ) + + def test_restart_lookup_downgrades_stale_sealed_rollback_to_unknown(self): + """A sealed rollback is acknowledged only while its pre-image is exact.""" + exact_overlay( + self.overlay_path, + {"foo": 1}, + ) + prepared = self.prepare({"foo": 2}) + sealed = self.target.seal(self.target.apply(prepared)) + rolled_back = self.target.rollback(sealed) + sealed_rollback = self.target.seal(rolled_back) + exact_overlay( + self.overlay_path, + {"foo": 44}, + ) + + restarted = GeneratedOverlayWholeConfigTarget( + self.overlay_path, + FileOpaqueTargetJournal(self.journal_path), + ALLOWED_KEYS, + ) + observed = restarted.lookup( + sealed_rollback.key.fence, + sealed_rollback.key.projection_digest, + ) + self.assertEqual( + observed.terminal_phase, + TargetTerminalPhase.UNKNOWN, + ) + + def test_apply_crash_before_write_recovers_rolled_back(self): + """APPLYING plus the exact pre-image cannot be mistaken for commit.""" + target = FailingWriteTarget( + self.overlay_path, + self.journal, + ALLOWED_KEYS, + ) + exact_overlay( + self.overlay_path, + {"foo": 1}, + ) + prepared = self.prepare( + {"foo": 2}, + target=target, + ) + target.fail_before_write = True + with self.assertRaisesRegex(OSError, "secret-before-write"): + target.apply(prepared) + + restarted = GeneratedOverlayWholeConfigTarget( + self.overlay_path, + FileOpaqueTargetJournal(self.journal_path), + ALLOWED_KEYS, + ) + self.assertEqual( + restarted.lookup( + prepared.key.fence, + prepared.key.projection_digest, + ).phase, + TargetOperationPhase.APPLYING, + ) + recovered = restarted.invalidate(prepared.key) + self.assertEqual( + recovered.terminal_phase, + TargetTerminalPhase.ROLLED_BACK, + ) + self.assertEqual( + read_overlay(self.overlay_path)[1], + {"foo": 1}, + ) + + def test_apply_crash_after_exact_replace_recovers_applied(self): + """APPLYING plus the exact candidate is a durable committed outcome.""" + target = FailingWriteTarget( + self.overlay_path, + self.journal, + ALLOWED_KEYS, + ) + exact_overlay( + self.overlay_path, + {"foo": 1}, + ) + prepared = self.prepare( + {"foo": 2}, + target=target, + ) + target.fail_after_write = True + with self.assertRaisesRegex(OSError, "secret-after-write"): + target.apply(prepared) + + restarted = GeneratedOverlayWholeConfigTarget( + self.overlay_path, + FileOpaqueTargetJournal(self.journal_path), + ALLOWED_KEYS, + ) + recovered = restarted.invalidate(prepared.key) + self.assertEqual( + recovered.terminal_phase, + TargetTerminalPhase.APPLIED, + ) + self.assertEqual( + read_overlay(self.overlay_path)[1], + {"foo": 2}, + ) + + def test_rollback_crash_before_and_after_absence_restore_are_exact(self): + """ROLLING_BACK classifies candidate and absent pre-image distinctly.""" + target = FailingWriteTarget( + self.overlay_path, + self.journal, + ALLOWED_KEYS, + ) + prepared = self.prepare( + {"foo": 2}, + target=target, + ) + sealed = target.seal(target.apply(prepared)) + target.fail_before_remove = True + with self.assertRaisesRegex(OSError, "secret-before-remove"): + target.rollback(sealed) + + restarted = GeneratedOverlayWholeConfigTarget( + self.overlay_path, + FileOpaqueTargetJournal(self.journal_path), + ALLOWED_KEYS, + ) + still_applied = restarted.invalidate(sealed.key) + self.assertEqual( + still_applied.terminal_phase, + TargetTerminalPhase.APPLIED, + ) + + restored_first = restarted.rollback(still_applied) + restarted.seal(restored_first) + # Use a new fenced operation whose pre-image is again absent. + second = self.prepare( + {"foo": 3}, + fence=2, + ) + second_sealed = self.target.seal(self.target.apply(second)) + second_target = FailingWriteTarget( + self.overlay_path, + FileOpaqueTargetJournal(self.journal_path), + ALLOWED_KEYS, + ) + second_target.fail_after_remove = True + with self.assertRaisesRegex(OSError, "secret-after-remove"): + second_target.rollback(second_sealed) + final_restart = GeneratedOverlayWholeConfigTarget( + self.overlay_path, + FileOpaqueTargetJournal(self.journal_path), + ALLOWED_KEYS, + ) + restored = final_restart.invalidate(second.key) + self.assertEqual( + restored.terminal_phase, + TargetTerminalPhase.ROLLED_BACK, + ) + self.assertFalse(os.path.exists(self.overlay_path)) + + def test_applying_journal_cas_failure_precedes_overlay_write(self): + """No config mutation occurs unless APPLYING is durably acknowledged.""" + exact_overlay( + self.overlay_path, + {"foo": 1}, + ) + rejecting = RejectingJournal( + self.journal, + reject_call=2, + ) + target = GeneratedOverlayWholeConfigTarget( + self.overlay_path, + rejecting, + ALLOWED_KEYS, + ) + prepared = self.prepare( + {"foo": 2}, + target=target, + ) + + with self.assertRaises(WholeConfigTargetConflict): + target.apply(prepared) + self.assertEqual( + read_overlay(self.overlay_path)[1], + {"foo": 1}, + ) + self.assertEqual( + FileOpaqueTargetJournal(self.journal_path).load().revision, + 1, + ) + + def test_apply_reserves_complete_revision_headroom_before_file_effect(self): + """APPLYING reserves APPLIED, SEALED, and later UNKNOWN revisions.""" + exact_overlay( + self.overlay_path, + {"foo": 1}, + ) + prepared = self.prepare({"foo": 2}) + boundary_target, boundary = self.boundary_target(MAX_SEQUENCE - 3) + before = read_overlay(self.overlay_path)[0] + + with self.assertRaisesRegex( + WholeConfigTargetError, + "insufficient lifecycle headroom", + ): + boundary_target.apply(prepared) + self.assertEqual(boundary.writes, 0) + self.assertEqual( + read_overlay(self.overlay_path)[0], + before, + ) + self.assertEqual( + boundary_target.lookup( + prepared.key.fence, + prepared.key.projection_digest, + ).phase, + TargetOperationPhase.PREPARED, + ) + + boundary_target, boundary = self.boundary_target(MAX_SEQUENCE - 4) + applied = boundary_target.apply(prepared) + self.assertEqual( + applied.phase, + TargetOperationPhase.APPLIED, + ) + sealed = boundary_target.seal(applied) + self.assertEqual( + sealed.terminal_phase, + TargetTerminalPhase.APPLIED, + ) + self.assertEqual( + boundary.entry.revision, + MAX_SEQUENCE - 1, + ) + exact_overlay( + self.overlay_path, + {"foo": 9}, + ) + unknown = boundary_target.lookup( + prepared.key.fence, + prepared.key.projection_digest, + ) + self.assertEqual( + unknown.terminal_phase, + TargetTerminalPhase.UNKNOWN, + ) + self.assertEqual( + boundary.entry.revision, + MAX_SEQUENCE, + ) + + def test_rollback_reserves_complete_revision_headroom_before_file_effect(self): + """ROLLING_BACK reserves rollback, seal, and later UNKNOWN revisions.""" + exact_overlay( + self.overlay_path, + {"foo": 1}, + ) + prepared = self.prepare({"foo": 2}) + sealed = self.target.seal(self.target.apply(prepared)) + boundary_target, boundary = self.boundary_target(MAX_SEQUENCE - 3) + before = read_overlay(self.overlay_path)[0] + + with self.assertRaisesRegex( + WholeConfigTargetError, + "insufficient lifecycle headroom", + ): + boundary_target.rollback(sealed) + self.assertEqual(boundary.writes, 0) + self.assertEqual( + read_overlay(self.overlay_path)[0], + before, + ) + self.assertEqual( + boundary_target.lookup( + sealed.key.fence, + sealed.key.projection_digest, + ).terminal_phase, + TargetTerminalPhase.APPLIED, + ) + + boundary_target, boundary = self.boundary_target(MAX_SEQUENCE - 4) + rolled_back = boundary_target.rollback(sealed) + self.assertEqual( + rolled_back.phase, + TargetOperationPhase.ROLLED_BACK, + ) + resealed = boundary_target.seal(rolled_back) + self.assertEqual( + resealed.terminal_phase, + TargetTerminalPhase.ROLLED_BACK, + ) + self.assertEqual( + boundary.entry.revision, + MAX_SEQUENCE - 1, + ) + self.assertEqual( + read_overlay(self.overlay_path)[1], + {"foo": 1}, + ) + exact_overlay( + self.overlay_path, + {"foo": 9}, + ) + unknown = boundary_target.lookup( + sealed.key.fence, + sealed.key.projection_digest, + ) + self.assertEqual( + unknown.terminal_phase, + TargetTerminalPhase.UNKNOWN, + ) + self.assertEqual( + boundary.entry.revision, + MAX_SEQUENCE, + ) + + def test_post_applied_cas_mutation_reaches_sealed_unknown_at_max(self): + """MAX-4 covers apply, post-CAS UNKNOWN, and exact UNKNOWN sealing.""" + exact_overlay( + self.overlay_path, + {"foo": 1}, + ) + prepared = self.prepare({"foo": 2}) + _boundary_target, boundary = self.boundary_target(MAX_SEQUENCE - 4) + mutating = MutatingJournal( + boundary, + { + 2: lambda: exact_overlay( + self.overlay_path, + {"foo": 77}, + ) + }, + ) + target = GeneratedOverlayWholeConfigTarget( + self.overlay_path, + mutating, + ALLOWED_KEYS, + ) + + unknown = target.apply(prepared) + self.assertEqual( + unknown.phase, + TargetOperationPhase.UNKNOWN, + ) + self.assertEqual(boundary.entry.revision, MAX_SEQUENCE - 1) + sealed = target.invalidate(prepared.key) + self.assertEqual( + sealed.terminal_phase, + TargetTerminalPhase.UNKNOWN, + ) + self.assertEqual(boundary.entry.revision, MAX_SEQUENCE) + + def test_post_rolled_back_cas_mutation_reaches_sealed_unknown_at_max(self): + """MAX-4 covers rollback, post-CAS UNKNOWN, and exact UNKNOWN sealing.""" + exact_overlay( + self.overlay_path, + {"foo": 1}, + ) + prepared = self.prepare({"foo": 2}) + sealed_applied = self.target.seal(self.target.apply(prepared)) + _boundary_target, boundary = self.boundary_target(MAX_SEQUENCE - 4) + mutating = MutatingJournal( + boundary, + { + 2: lambda: exact_overlay( + self.overlay_path, + {"foo": 88}, + ) + }, + ) + target = GeneratedOverlayWholeConfigTarget( + self.overlay_path, + mutating, + ALLOWED_KEYS, + ) + + unknown = target.rollback(sealed_applied) + self.assertEqual( + unknown.phase, + TargetOperationPhase.UNKNOWN, + ) + self.assertEqual(boundary.entry.revision, MAX_SEQUENCE - 1) + sealed = target.invalidate(prepared.key) + self.assertEqual( + sealed.terminal_phase, + TargetTerminalPhase.UNKNOWN, + ) + self.assertEqual(boundary.entry.revision, MAX_SEQUENCE) + + def test_prepare_requires_full_apply_and_seal_revision_budget(self): + """A new PREPARED entry cannot consume the final lifecycle revisions.""" + exact_overlay( + self.overlay_path, + {"foo": 1}, + ) + first = self.prepare({"foo": 1}) + sealed = self.target.seal(first) + self.assertEqual( + sealed.terminal_phase, + TargetTerminalPhase.ROLLED_BACK, + ) + boundary_target, boundary = self.boundary_target(MAX_SEQUENCE - 4) + before = read_overlay(self.overlay_path)[0] + + with self.assertRaisesRegex( + WholeConfigTargetError, + "insufficient lifecycle headroom", + ): + boundary_target.prepare( + 2, + digest({"foo": 2}), + {"foo": 2}, + ) + self.assertEqual(boundary.writes, 0) + self.assertEqual( + read_overlay(self.overlay_path)[0], + before, + ) + + def test_prepare_at_max_minus_five_reaches_post_apply_unknown_and_seals(self): + """The complete prepared lifecycle safely consumes its five revisions.""" + exact_overlay( + self.overlay_path, + {"foo": 1}, + ) + first = self.prepare({"foo": 1}) + self.target.seal(first) + _boundary_target, boundary = self.boundary_target(MAX_SEQUENCE - 5) + target = GeneratedOverlayWholeConfigTarget( + self.overlay_path, + boundary, + ALLOWED_KEYS, + ) + projection = {"foo": 2} + prepared = target.prepare( + 2, + digest(projection), + projection, + ) + self.assertEqual(boundary.entry.revision, MAX_SEQUENCE - 4) + + mutating = MutatingJournal( + boundary, + { + 2: lambda: exact_overlay( + self.overlay_path, + {"foo": 99}, + ) + }, + ) + target = GeneratedOverlayWholeConfigTarget( + self.overlay_path, + mutating, + ALLOWED_KEYS, + ) + unknown = target.apply(prepared) + self.assertEqual( + unknown.phase, + TargetOperationPhase.UNKNOWN, + ) + self.assertEqual(boundary.entry.revision, MAX_SEQUENCE - 1) + sealed = target.invalidate(prepared.key) + self.assertEqual( + sealed.terminal_phase, + TargetTerminalPhase.UNKNOWN, + ) + self.assertEqual(boundary.entry.revision, MAX_SEQUENCE) + + def test_seal_reserves_unknown_revision_before_terminal_cas(self): + """Seal performs no CAS unless a later UNKNOWN downgrade remains possible.""" + exact_overlay( + self.overlay_path, + {"foo": 1}, + ) + prepared = self.prepare({"foo": 2}) + boundary_target, boundary = self.boundary_target(MAX_SEQUENCE - 1) + + with self.assertRaisesRegex( + WholeConfigTargetError, + "insufficient lifecycle headroom", + ): + boundary_target.seal(prepared) + self.assertEqual(boundary.writes, 0) + self.assertEqual( + read_overlay(self.overlay_path)[1], + {"foo": 1}, + ) + + _boundary_target, boundary = self.boundary_target(MAX_SEQUENCE - 2) + mutating = MutatingJournal( + boundary, + { + 1: lambda: exact_overlay( + self.overlay_path, + {"foo": 55}, + ) + }, + ) + target = GeneratedOverlayWholeConfigTarget( + self.overlay_path, + mutating, + ALLOWED_KEYS, + ) + sealed = target.seal(prepared) + self.assertEqual( + sealed.terminal_phase, + TargetTerminalPhase.UNKNOWN, + ) + self.assertEqual(boundary.entry.revision, MAX_SEQUENCE) + self.assertEqual(boundary.writes, 2) + + def test_invalidate_reserves_unknown_revision_before_terminal_cas(self): + """Invalidate performs no CAS unless post-CAS recheck can be recorded.""" + exact_overlay( + self.overlay_path, + {"foo": 1}, + ) + prepared = self.prepare({"foo": 2}) + boundary_target, boundary = self.boundary_target(MAX_SEQUENCE - 1) + + with self.assertRaisesRegex( + WholeConfigTargetError, + "insufficient lifecycle headroom", + ): + boundary_target.invalidate(prepared.key) + self.assertEqual(boundary.writes, 0) + self.assertEqual( + read_overlay(self.overlay_path)[1], + {"foo": 1}, + ) + + _boundary_target, boundary = self.boundary_target(MAX_SEQUENCE - 2) + mutating = MutatingJournal( + boundary, + { + 1: lambda: exact_overlay( + self.overlay_path, + {"foo": 66}, + ) + }, + ) + target = GeneratedOverlayWholeConfigTarget( + self.overlay_path, + mutating, + ALLOWED_KEYS, + ) + sealed = target.invalidate(prepared.key) + self.assertEqual( + sealed.terminal_phase, + TargetTerminalPhase.UNKNOWN, + ) + self.assertEqual(boundary.entry.revision, MAX_SEQUENCE) + self.assertEqual(boundary.writes, 2) + + def test_unknown_recovery_has_reserved_revision_headroom(self): + """Pre-image mismatch can still durably seal UNKNOWN at the boundary.""" + exact_overlay( + self.overlay_path, + {"foo": 1}, + ) + prepared = self.prepare({"foo": 2}) + exact_overlay( + self.overlay_path, + {"foo": 9}, + ) + boundary_target, boundary = self.boundary_target(MAX_SEQUENCE - 4) + + unknown = boundary_target.apply(prepared) + self.assertEqual( + unknown.phase, + TargetOperationPhase.UNKNOWN, + ) + sealed = boundary_target.invalidate(prepared.key) + self.assertEqual( + sealed.terminal_phase, + TargetTerminalPhase.UNKNOWN, + ) + self.assertEqual( + boundary.entry.revision, + MAX_SEQUENCE - 1, + ) + self.assertEqual( + read_overlay(self.overlay_path)[1], + {"foo": 9}, + ) + + def test_outer_lock_orders_apply_and_invalidate_across_instances(self): + """Invalidate cannot classify visibility while a commit is in flight.""" + exact_overlay( + self.overlay_path, + {"foo": 1}, + ) + writer = BlockingWriteTarget( + self.overlay_path, + self.journal, + ALLOWED_KEYS, + ) + observer = GeneratedOverlayWholeConfigTarget( + self.overlay_path, + FileOpaqueTargetJournal(self.journal_path), + ALLOWED_KEYS, + ) + prepared = self.prepare( + {"foo": 2}, + target=writer, + ) + results = {} + + apply_thread = threading.Thread( + target=lambda: results.setdefault( + "apply", + writer.apply(prepared), + ) + ) + invalidate_thread = threading.Thread( + target=lambda: results.setdefault( + "invalidate", + observer.invalidate(prepared.key), + ) + ) + apply_thread.start() + self.assertTrue(writer.entered.wait(2)) + invalidate_thread.start() + invalidate_thread.join(0.1) + self.assertTrue(invalidate_thread.is_alive()) + writer.release.set() + apply_thread.join(2) + invalidate_thread.join(2) + + self.assertFalse(apply_thread.is_alive()) + self.assertFalse(invalidate_thread.is_alive()) + self.assertEqual( + results["apply"].phase, + TargetOperationPhase.APPLIED, + ) + self.assertEqual( + results["invalidate"].terminal_phase, + TargetTerminalPhase.APPLIED, + ) + + def test_replay_stale_fence_and_unsettled_replacement_fail_closed(self): + """Prepare binds idempotency to exact monotonically fenced journal keys.""" + first = self.prepare( + {"foo": 1}, + fence=3, + ) + for fence, config in ( + (3, {"foo": 1}), + (3, {"foo": 2}), + (2, {"foo": 3}), + (4, {"foo": 4}), + ): + with self.assertRaises(WholeConfigTargetConflict): + self.prepare( + config, + fence=fence, + ) + + self.target.seal(first) + second = self.prepare( + {"foo": 5}, + fence=4, + ) + self.assertEqual( + second.key, + TargetOperationKey(4, digest({"foo": 5})), + ) + with self.assertRaises(WholeConfigTargetConflict): + self.target.invalidate(first.key) + + def test_abort_is_durable_and_cannot_regress_current_fence(self): + """An unprepared reservation is tombstoned with no overlay mutation.""" + key = TargetOperationKey( + 5, + digest({"foo": 1}), + ) + aborted = self.target.abort( + key.fence, + key.projection_digest, + ) + self.assertEqual( + aborted.phase, + TargetOperationPhase.SEALED, + ) + self.assertEqual( + aborted.terminal_phase, + TargetTerminalPhase.ROLLED_BACK, + ) + self.assertEqual( + self.target.lookup( + key.fence, + key.projection_digest, + ), + aborted, + ) + with self.assertRaises(WholeConfigTargetConflict): + self.target.abort( + 4, + digest({"foo": 2}), + ) + + def test_allowlist_and_protected_filename_checks_precede_target_effects(self): + """Only explicit owned keys can enter the journal or overlay.""" + before = self.journal.load() + for hostile in ( + {"not_owned": "secret-value"}, + {"apps": {"module": "predbat"}}, + ): + with self.assertRaises(ValueError): + self.prepare(hostile) + self.assertEqual( + self.journal.load(), + before, + ) + self.assertFalse(os.path.exists(self.overlay_path)) + + for protected in ( + "apps.yaml", + "APPS.YML", + "secrets.yaml", + "Secrets.yml", + ): + path = os.path.join( + self.temporary.name, + protected, + ) + with self.assertRaises(ValueError): + GeneratedOverlayWholeConfigTarget( + path, + self.journal, + ALLOWED_KEYS, + ) + self.assertFalse(os.path.exists(path)) + + def test_apps_and_secrets_markers_are_never_read_or_written(self): + """A complete target lifecycle leaves live config marker files exact.""" + markers = {} + for name in ( + "apps.yaml", + "secrets.yaml", + ): + path = os.path.join( + self.temporary.name, + name, + ) + payload = "{}-marker-secret".format(name).encode("ascii") + with open(path, "wb") as marker: + marker.write(payload) + markers[path] = payload + + prepared = self.prepare({"foo": 2}) + applied = self.target.apply(prepared) + sealed = self.target.seal(applied) + rolled_back = self.target.rollback(sealed) + self.target.seal(rolled_back) + + for path, payload in markers.items(): + with open(path, "rb") as marker: + self.assertEqual(marker.read(), payload) + + def test_overlay_requires_canonical_private_regular_allowlisted_json(self): + """Noncanonical, unsafe, and non-owned existing overlays fail closed.""" + hostile_payloads = ( + b'{"foo":1, "bar":2}', + b'{"foo":1,"foo":2}', + b'{"not_owned":1}', + b'{"foo":NaN}', + b"[]", + ) + for payload in hostile_payloads: + with self.subTest(payload=payload): + with open(self.overlay_path, "wb") as overlay: + overlay.write(payload) + os.chmod(self.overlay_path, 0o600) + with self.assertRaises(WholeConfigTargetCorrupt): + self.prepare({"foo": 2}) + os.unlink(self.overlay_path) + + exact_overlay( + self.overlay_path, + {"foo": 1}, + ) + os.chmod(self.overlay_path, 0o644) + with self.assertRaises(WholeConfigTargetCorrupt): + self.prepare({"foo": 2}) + os.unlink(self.overlay_path) + + real = os.path.join( + self.temporary.name, + "real.json", + ) + exact_overlay(real, {"foo": 1}) + os.symlink(real, self.overlay_path) + with self.assertRaises(WholeConfigTargetError): + self.prepare({"foo": 2}) + + def test_target_binding_and_allowlist_binding_reject_mispaired_restart(self): + """An opaque journal cannot be replayed against another target identity.""" + prepared = self.prepare({"foo": 1}) + self.assertIsNotNone(prepared) + + other_path = os.path.join( + self.temporary.name, + "other-generated-overlay.json", + ) + wrong_path = GeneratedOverlayWholeConfigTarget( + other_path, + FileOpaqueTargetJournal(self.journal_path), + ALLOWED_KEYS, + ) + with self.assertRaises(WholeConfigTargetCorrupt): + wrong_path.lookup( + prepared.key.fence, + prepared.key.projection_digest, + ) + + wrong_allowlist = GeneratedOverlayWholeConfigTarget( + self.overlay_path, + FileOpaqueTargetJournal(self.journal_path), + ("foo",), + ) + with self.assertRaises(WholeConfigTargetCorrupt): + wrong_allowlist.lookup( + prepared.key.fence, + prepared.key.projection_digest, + ) + + def test_journal_decode_rejects_absent_nonempty_overlay_state(self): + """File absence can only encode the exact canonical empty config.""" + prepared = self.prepare({"foo": 1}) + entry = self.journal.load() + document = json.loads(entry.payload.decode("ascii")) + document["preimage"] = { + "present": False, + "config": {"foo": 99}, + } + hostile = OpaqueTargetJournalEntry( + entry.revision, + entry.key, + json.dumps( + document, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + allow_nan=False, + ).encode("ascii"), + ) + target = GeneratedOverlayWholeConfigTarget( + self.overlay_path, + BoundaryJournal(hostile), + ALLOWED_KEYS, + ) + + with self.assertRaises(WholeConfigTargetCorrupt): + target.lookup( + prepared.key.fence, + prepared.key.projection_digest, + ) + + def test_hostile_snapshot_identity_cannot_apply_or_rollback(self): + """Opaque handle/digest forgery never reaches an overlay write.""" + exact_overlay( + self.overlay_path, + {"foo": 1}, + ) + prepared = self.prepare({"foo": 2}) + hostile = replace( + prepared, + handle="lattice-target-forged", + ) + with self.assertRaises(WholeConfigTargetConflict): + self.target.apply(hostile) + self.assertEqual( + read_overlay(self.overlay_path)[1], + {"foo": 1}, + ) + + sealed = self.target.seal(prepared) + with self.assertRaises(WholeConfigTargetConflict): + self.target.rollback( + replace( + sealed, + candidate_digest="0" * 64, + changed=True, + ) + ) + self.assertEqual( + read_overlay(self.overlay_path)[1], + {"foo": 1}, + ) + + def test_fault_messages_and_repr_do_not_expose_projection_values(self): + """Diagnostics retain only value-free failure classes and counts.""" + secret_value = "raw-target-secret-value" + target = FailingWriteTarget( + self.overlay_path, + self.journal, + ALLOWED_KEYS, + ) + prepared = self.prepare( + {"foo": secret_value}, + target=target, + ) + target.fail_before_write = True + with self.assertRaises(OSError) as raised: + target.apply(prepared) + self.assertNotIn(secret_value, repr(prepared)) + self.assertNotIn(secret_value, repr(self.journal.load())) + self.assertNotIn(secret_value, repr(target)) + # The injected test exception is deliberately hostile; production + # materializer wrapping exposes only its exception class. + self.assertNotIn(secret_value, type(raised.exception).__name__) + + def test_lock_and_directory_security_fail_closed(self): + """Unsafe lock metadata and a symlinked target directory are rejected.""" + lock_path = self.overlay_path + ".lock" + with open(lock_path, "wb") as lock: + lock.write(b"") + os.chmod(lock_path, 0o644) + with self.assertRaises(WholeConfigTargetError): + self.prepare({"foo": 1}) + + real_directory = os.path.join( + self.temporary.name, + "real", + ) + linked_directory = os.path.join( + self.temporary.name, + "linked", + ) + os.mkdir(real_directory, 0o700) + os.symlink(real_directory, linked_directory) + with self.assertRaises(WholeConfigTargetError): + GeneratedOverlayWholeConfigTarget( + os.path.join(linked_directory, "overlay.json"), + FileOpaqueTargetJournal(os.path.join(real_directory, "other.journal")), + ALLOWED_KEYS, + ) + + def test_bool_fence_wrong_digest_and_mutable_allowlist_fail_before_journal(self): + """Type-exact boundary validation precedes durable side effects.""" + before = self.journal.load() + with self.assertRaises(ValueError): + self.target.prepare( + True, + digest({"foo": 1}), + {"foo": 1}, + ) + with self.assertRaises(ValueError): + self.target.prepare( + 1, + "0" * 64, + {"foo": 1}, + ) + with self.assertRaises(ValueError): + GeneratedOverlayWholeConfigTarget( + self.overlay_path, + self.journal, + ["foo"], + ) + self.assertEqual( + self.journal.load(), + before, + ) + + def test_failed_exact_readback_remains_recoverably_ambiguous(self): + """Readback mismatch cannot be acknowledged as an applied operation.""" + prepared = self.prepare({"foo": 2}) + directory_fd = self.target._open_directory() + try: + preimage = self.target._read_overlay(directory_fd) + finally: + os.close(directory_fd) + with mock.patch.object( + self.target, + "_read_overlay", + side_effect=( + preimage, + WholeConfigTargetCorrupt("injected-readback"), + ), + ): + with self.assertRaises(WholeConfigTargetCorrupt): + self.target.apply(prepared) + restarted = GeneratedOverlayWholeConfigTarget( + self.overlay_path, + FileOpaqueTargetJournal(self.journal_path), + ALLOWED_KEYS, + ) + recovered = restarted.invalidate(prepared.key) + self.assertEqual( + recovered.terminal_phase, + TargetTerminalPhase.APPLIED, + ) + + +if __name__ == "__main__": + unittest.main()