From 870b383ccc19ca2c5f89754a57d866ea1a4a003c Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Tue, 28 Jul 2026 21:47:58 +0100 Subject: [PATCH] feat(lattice): add atomic autoconfig materializer --- apps/predbat/lattice_atomic_materializer.py | 1673 +++++++++++++++++ .../tests/test_lattice_atomic_materializer.py | 1340 +++++++++++++ 2 files changed, 3013 insertions(+) create mode 100644 apps/predbat/lattice_atomic_materializer.py create mode 100644 apps/predbat/tests/test_lattice_atomic_materializer.py diff --git a/apps/predbat/lattice_atomic_materializer.py b/apps/predbat/lattice_atomic_materializer.py new file mode 100644 index 000000000..bbaacd75c --- /dev/null +++ b/apps/predbat/lattice_atomic_materializer.py @@ -0,0 +1,1673 @@ +# ----------------------------------------------------------------------------- +# Predbat Home Battery System - atomic Lattice auto-config materializer +# Copyright Trefor Southwell 2026 - All Rights Reserved +# This application maybe used for personal use only and not for commercial use +# ----------------------------------------------------------------------------- +"""Default-off atomic materialization of immutable Lattice auto-config plans. + +The materializer first wins a durable secret-free reservation, then asks the +target to retain the secret-bearing preimage and candidate behind an opaque +handle. Every target mutation is fenced, recoverable by reservation key, and +sealed before the materializer reports a terminal result. +""" + +# cspell:ignore autoconfig dedupe preimage readback + +import hashlib +import json +import math +import re +import threading +from dataclasses import dataclass +from enum import Enum +from types import MappingProxyType + +from lattice_autoconfig import ( + AutoConfigPlan, + FieldProvenance, + MaterializationReadiness, + MaterializationRequest, + ProjectedConfigArgument, + _canonical_json as _compiler_canonical_json, + _semantic_topology, +) + + +MAX_SEQUENCE = (1 << 63) - 1 +MAX_CONFIG_INTEGER = MAX_SEQUENCE +MAX_CONFIG_ENTRIES = 256 +MAX_CONFIG_KEY_LENGTH = 128 +MAX_CONFIG_VALUE_BYTES = 16384 +MAX_CONFIG_BYTES = 262144 +MAX_LIST_ITEMS = 256 +MAX_VALUE_DEPTH = 4 +MAX_FEEDBACK_TOKEN_LENGTH = 256 +MAX_PROVENANCE_ENTRIES = 4096 +MAX_PROVENANCE_STRING_LENGTH = 1024 +MAX_PROVIDER_ID_LENGTH = 256 +MAX_OPAQUE_HANDLE_LENGTH = 256 +MAX_SEMANTIC_ITEMS = 4096 +MAX_SEMANTIC_DEPTH = 12 +MAX_SEMANTIC_NODES = 16384 +MAX_SEMANTIC_STRING_LENGTH = 16384 +MAX_SEMANTIC_BYTES = 1048576 +MAX_RESERVATION_RETRIES = 32 + +_CONFIG_KEY = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") +_DIGEST = re.compile(r"^[0-9a-f]{64}$") +_OPAQUE_HANDLE = re.compile(r"^[A-Za-z0-9_.:@-]+$") +_MAPPING_PROXY_TYPE = type(MappingProxyType({})) + + +class AtomicMaterializationError(RuntimeError): + """Base error for fail-closed materialization.""" + + +class MaterializationValidationError(AtomicMaterializationError): + """An immutable request or durable record is unsafe.""" + + +class MaterializationStorageError(AtomicMaterializationError): + """Durable reservation or CAS state could not be established exactly.""" + + +class MaterializationTargetError(AtomicMaterializationError): + """The target lifecycle could not be acknowledged exactly.""" + + +class MaterializationPhase(Enum): + """Secret-free durable coordinator phase.""" + + RESERVED = "reserved" + PREPARED = "prepared" + APPLYING = "applying" + APPLIED = "applied" + ROLLING_BACK = "rolling_back" + ROLLED_BACK = "rolled_back" + UNKNOWN = "unknown" + + +class TargetOperationPhase(Enum): + """Target-owned durable operation lifecycle.""" + + PREPARED = "prepared" + APPLYING = "applying" + APPLIED = "applied" + ROLLING_BACK = "rolling_back" + ROLLED_BACK = "rolled_back" + SEALED = "sealed" + UNKNOWN = "unknown" + + +class TargetTerminalPhase(Enum): + """Meaning of one acknowledged sealed target operation.""" + + APPLIED = "applied" + ROLLED_BACK = "rolled_back" + UNKNOWN = "unknown" + + +class MaterializationStatus(Enum): + """Caller-visible result of apply, recovery, or rollback.""" + + DISABLED = "disabled" + APPLIED = "applied" + UNCHANGED = "unchanged" + NOT_APPLIED = "not_applied" + UNKNOWN = "unknown" + SUPERSEDED = "superseded" + ROLLED_BACK = "rolled_back" + + +class AtomicMaterializationStateStore: + """Injected durable fencing and exact compare-and-store protocol.""" + + def allocate_fence(self): + """Durably allocate a globally increasing bounded positive fence.""" + raise NotImplementedError + + def load(self): + """Return the current exact state or ``None``.""" + raise NotImplementedError + + def compare_and_store(self, expected, replacement): + """Atomically replace exact ``expected`` state.""" + raise NotImplementedError + + +class WholeConfigTarget: + """Target-owned durable, fenced, secret-bearing transaction protocol.""" + + def prepare(self, fence, projection_digest, projection): + """Create one PREPARED operation for a winning reservation.""" + raise NotImplementedError + + def lookup(self, fence, projection_digest): + """Recover a durable operation by its reservation key.""" + raise NotImplementedError + + def apply(self, operation): + """Run PREPARED -> APPLYING -> APPLIED with a commit fence recheck.""" + raise NotImplementedError + + def rollback(self, operation): + """Run sealed APPLIED -> ROLLING_BACK -> ROLLED_BACK.""" + raise NotImplementedError + + def seal(self, operation): + """Seal a stable operation and reject later apply replay.""" + raise NotImplementedError + + def abort(self, fence, projection_digest): + """Seal an unprepared/prepared reservation as ROLLED_BACK.""" + raise NotImplementedError + + def invalidate(self, operation_key): + """Fence an operation and acknowledge one SEALED terminal result.""" + raise NotImplementedError + + +@dataclass(frozen=True, repr=False) +class PublishedMaterializationRequest: + """Materialization handoff plus its immutable publication cursor.""" + + request: MaterializationRequest + publication_digest: str + publication_version: int + + def __post_init__(self): + """Validate non-secret envelope fields without traversing the plan.""" + if type(self.request) is not MaterializationRequest: + raise ValueError("request type is invalid") + if type(self.request.plan) is not AutoConfigPlan: + raise ValueError("plan type is invalid") + _validate_digest(self.publication_digest, "publication_digest") + if self.publication_digest != self.request.plan.digest: + raise ValueError("publication digest binding is invalid") + _validate_sequence( + self.publication_version, + "publication_version", + minimum=1, + ) + _validate_feedback_token(self.request.feedback_token) + + def __repr__(self): + """Exclude projection values and feedback tokens from diagnostics.""" + return ("PublishedMaterializationRequest(" "plan_digest={!r}, publication_version={!r})").format(self.publication_digest, self.publication_version) + + +@dataclass(frozen=True) +class TargetOperationKey: + """Bounded durable key shared by reservation state and target journal.""" + + fence: int + projection_digest: str + + def __post_init__(self): + """Validate a reservation lookup key.""" + _validate_sequence(self.fence, "fence", minimum=1) + _validate_digest(self.projection_digest, "projection_digest") + + +@dataclass(frozen=True) +class TargetOperationSnapshot: + """Secret-free target journal observation.""" + + key: TargetOperationKey + phase: TargetOperationPhase + handle: object = None + preimage_digest: object = None + candidate_digest: object = None + changed: object = None + terminal_phase: object = None + + def __post_init__(self): + """Validate phase-specific opaque target metadata.""" + if type(self.key) is not TargetOperationKey: + raise ValueError("target operation key type is invalid") + TargetOperationKey.__post_init__(self.key) + if not isinstance(self.phase, TargetOperationPhase): + raise ValueError("target operation phase is invalid") + prepared_values = ( + self.handle, + self.preimage_digest, + self.candidate_digest, + self.changed, + ) + has_prepared_values = all(value is not None for value in prepared_values) + has_partial_values = any(value is not None for value in prepared_values) + if has_partial_values and not has_prepared_values: + raise ValueError("target operation metadata is partial") + if has_prepared_values: + _validate_handle(self.handle) + _validate_digest(self.preimage_digest, "preimage_digest") + _validate_digest(self.candidate_digest, "candidate_digest") + if type(self.changed) is not bool: + raise ValueError("target changed flag is invalid") + if self.changed != (self.preimage_digest != self.candidate_digest): + raise ValueError("target changed flag contradicts digests") + elif self.phase is not TargetOperationPhase.SEALED: + raise ValueError("unprepared target operation must be sealed") + + if self.phase is TargetOperationPhase.SEALED: + if not isinstance(self.terminal_phase, TargetTerminalPhase): + raise ValueError("sealed target operation lacks terminal phase") + if not has_prepared_values and self.terminal_phase is not TargetTerminalPhase.ROLLED_BACK: + raise ValueError("unprepared seal must be rolled back") + elif self.terminal_phase is not None: + raise ValueError("unsealed target operation has terminal phase") + + +@dataclass(frozen=True) +class AtomicMaterializationState: + """Bounded secret-free durable reservation and transaction checkpoint.""" + + revision: int + phase: MaterializationPhase + plan_digest: str + publication_digest: str + publication_version: int + projection_digest: str + feedback_digest: str + provenance_digest: str + provenance_count: int + reservation_key: TargetOperationKey + prior_operation_key: object = None + operation: object = None + applied_plan_digest: object = None + applied_publication_digest: object = None + applied_publication_version: object = None + integrity_digest: str = "" + + def __post_init__(self): + """Verify every checkpoint field and its complete fingerprint.""" + _validate_sequence(self.revision, "revision", minimum=1) + if not isinstance(self.phase, MaterializationPhase): + raise ValueError("state phase is invalid") + _validate_digest(self.plan_digest, "plan_digest") + _validate_digest(self.publication_digest, "publication_digest") + if self.plan_digest != self.publication_digest: + raise ValueError("state plan/publication binding is invalid") + _validate_sequence( + self.publication_version, + "publication_version", + minimum=1, + ) + _validate_digest(self.projection_digest, "projection_digest") + _validate_digest(self.feedback_digest, "feedback_digest") + _validate_digest(self.provenance_digest, "provenance_digest") + _validate_sequence( + self.provenance_count, + "provenance_count", + minimum=0, + maximum=MAX_PROVENANCE_ENTRIES, + ) + if type(self.reservation_key) is not TargetOperationKey: + raise ValueError("reservation key type is invalid") + TargetOperationKey.__post_init__(self.reservation_key) + if self.reservation_key.projection_digest != self.projection_digest: + raise ValueError("reservation projection binding is invalid") + if self.prior_operation_key is not None: + if type(self.prior_operation_key) is not TargetOperationKey: + raise ValueError("prior operation key type is invalid") + TargetOperationKey.__post_init__(self.prior_operation_key) + if self.prior_operation_key.fence >= self.reservation_key.fence: + raise ValueError("prior operation fence is not older") + + if self.phase is MaterializationPhase.RESERVED: + if self.operation is not None: + raise ValueError("reserved state cannot have target operation") + else: + if type(self.operation) is not TargetOperationSnapshot: + raise ValueError("transaction state lacks target operation") + TargetOperationSnapshot.__post_init__(self.operation) + if self.operation.key != self.reservation_key: + raise ValueError("target operation key binding is invalid") + if self.phase is MaterializationPhase.PREPARED and self.operation.phase is not TargetOperationPhase.PREPARED: + raise ValueError("PREPARED state target phase is invalid") + if self.phase is MaterializationPhase.APPLYING and ( + self.operation.phase + not in ( + TargetOperationPhase.PREPARED, + TargetOperationPhase.APPLYING, + ) + ): + raise ValueError("APPLYING state target phase is invalid") + if self.phase is MaterializationPhase.ROLLING_BACK and not (self.operation.phase is TargetOperationPhase.SEALED and self.operation.terminal_phase is TargetTerminalPhase.APPLIED): + raise ValueError("ROLLING_BACK state target phase is invalid") + + if self.phase is MaterializationPhase.APPLIED: + if self.operation.phase is not TargetOperationPhase.SEALED or self.operation.terminal_phase is not TargetTerminalPhase.APPLIED: + raise ValueError("APPLIED requires sealed target APPLIED") + expected_applied = ( + self.plan_digest, + self.publication_digest, + self.publication_version, + ) + else: + expected_applied = (None, None, None) + if self.phase is MaterializationPhase.ROLLED_BACK and (self.operation.phase is not TargetOperationPhase.SEALED or self.operation.terminal_phase is not TargetTerminalPhase.ROLLED_BACK): + raise ValueError("ROLLED_BACK requires sealed target ROLLED_BACK") + if ( + self.applied_plan_digest, + self.applied_publication_digest, + self.applied_publication_version, + ) != expected_applied: + raise ValueError("state applied cursor is invalid") + + if self.integrity_digest != _state_integrity(self): + raise ValueError("state integrity digest is invalid") + + @classmethod + def reserve( + cls, + revision, + envelope, + projection_digest, + feedback_digest, + provenance_digest, + provenance_count, + fence, + prior_operation_key, + ): + """Create one secret-free RESERVED checkpoint.""" + return _new_state( + revision=revision, + phase=MaterializationPhase.RESERVED, + plan_digest=envelope.request.plan.digest, + publication_digest=envelope.publication_digest, + publication_version=envelope.publication_version, + projection_digest=projection_digest, + feedback_digest=feedback_digest, + provenance_digest=provenance_digest, + provenance_count=provenance_count, + reservation_key=TargetOperationKey( + fence, + projection_digest, + ), + prior_operation_key=prior_operation_key, + operation=None, + applied_plan_digest=None, + applied_publication_digest=None, + applied_publication_version=None, + ) + + def with_operation(self, phase, operation): + """Advance one reservation using acknowledged target metadata.""" + applied = phase is MaterializationPhase.APPLIED + return _new_state( + revision=_next_sequence(self.revision, "revision"), + phase=phase, + plan_digest=self.plan_digest, + publication_digest=self.publication_digest, + publication_version=self.publication_version, + projection_digest=self.projection_digest, + feedback_digest=self.feedback_digest, + provenance_digest=self.provenance_digest, + provenance_count=self.provenance_count, + reservation_key=self.reservation_key, + prior_operation_key=self.prior_operation_key, + operation=operation, + applied_plan_digest=self.plan_digest if applied else None, + applied_publication_digest=(self.publication_digest if applied else None), + applied_publication_version=(self.publication_version if applied else None), + ) + + def advance_publication( + self, + envelope, + feedback_digest, + provenance_digest, + provenance_count, + ): + """Advance an unchanged sealed APPLIED publication cursor.""" + if self.phase is not MaterializationPhase.APPLIED: + raise ValueError("publication advance requires APPLIED state") + if envelope.request.plan.digest != self.plan_digest: + raise ValueError("publication advance changes plan digest") + return _new_state( + revision=_next_sequence(self.revision, "revision"), + phase=MaterializationPhase.APPLIED, + plan_digest=self.plan_digest, + publication_digest=envelope.publication_digest, + publication_version=envelope.publication_version, + projection_digest=self.projection_digest, + feedback_digest=feedback_digest, + provenance_digest=provenance_digest, + provenance_count=provenance_count, + reservation_key=self.reservation_key, + prior_operation_key=self.prior_operation_key, + operation=self.operation, + applied_plan_digest=self.plan_digest, + applied_publication_digest=envelope.publication_digest, + applied_publication_version=envelope.publication_version, + ) + + +@dataclass(frozen=True) +class AtomicMaterializationResult: + """Immutable outcome and durable state observation.""" + + status: MaterializationStatus + state: object + target_written: bool = False + + def __post_init__(self): + """Validate result shape.""" + if not isinstance(self.status, MaterializationStatus): + raise ValueError("result status is invalid") + if self.state is not None and type(self.state) is not AtomicMaterializationState: + raise ValueError("result state type is invalid") + if type(self.target_written) is not bool: + raise ValueError("target_written is invalid") + + +def _validate_sequence(value, name, minimum=0, maximum=MAX_SEQUENCE): + """Validate one bounded non-boolean integer before serialization.""" + if type(value) is not int or value < minimum or value > maximum: + raise ValueError("{} is outside bounded integer range".format(name)) + return value + + +def _next_sequence(value, name): + """Increment one bounded sequence without overflow.""" + _validate_sequence(value, name, minimum=0) + if value == MAX_SEQUENCE: + raise ValueError("{} exhausted".format(name)) + return value + 1 + + +def _validate_digest(value, name): + """Validate one lowercase SHA-256 digest.""" + if type(value) is not str or _DIGEST.fullmatch(value) is None: + raise ValueError("{} is not a SHA-256 digest".format(name)) + return value + + +def _validate_handle(value): + """Validate one bounded non-secret opaque handle.""" + if type(value) is not str or not value or len(value) > MAX_OPAQUE_HANDLE_LENGTH or _OPAQUE_HANDLE.fullmatch(value) is None: + raise ValueError("target handle is invalid") + return value + + +def _validate_feedback_token(value): + """Validate one bounded feedback token without retaining it.""" + if type(value) is not str or not value or value != value.strip() or len(value) > MAX_FEEDBACK_TOKEN_LENGTH: + raise ValueError("feedback token is invalid") + return value + + +def _canonical_value(value, depth=0): + """Detach one bounded JSON scalar or list value.""" + if depth > MAX_VALUE_DEPTH: + raise ValueError("config value depth exceeds limit") + if value is None or type(value) is bool: + canonical = value + elif type(value) is int: + _validate_sequence( + abs(value), + "config integer", + minimum=0, + maximum=MAX_CONFIG_INTEGER, + ) + canonical = value + elif type(value) is float: + if not math.isfinite(value): + raise ValueError("config float is not finite") + canonical = value + elif type(value) is str: + canonical = value + elif type(value) in (list, tuple): + if len(value) > MAX_LIST_ITEMS: + raise ValueError("config list exceeds item limit") + canonical = tuple(_canonical_value(item, depth + 1) for item in value) + else: + raise ValueError("config value type is invalid") + if len(_canonical_json(canonical).encode("utf-8")) > MAX_CONFIG_VALUE_BYTES: + raise ValueError("config value exceeds byte limit") + return canonical + + +def _plain_value(value): + """Convert immutable tuples into canonical JSON arrays.""" + if type(value) is tuple: + return [_plain_value(item) for item in value] + return value + + +def _canonical_json(value): + """Return deterministic strict JSON for already-bounded values.""" + if type(value) in (dict, _MAPPING_PROXY_TYPE): + value = {key: _plain_value(item) for key, item in value.items()} + else: + value = _plain_value(value) + return json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + allow_nan=False, + ) + + +def _bounded_mapping_items(value, name, maximum): + """Consume at most ``maximum + 1`` mapping items.""" + if type(value) not in (dict, _MAPPING_PROXY_TYPE): + raise ValueError("{} mapping type is invalid".format(name)) + iteration_failed = False + try: + iterator = iter(value.items()) + items = [] + for _index in range(maximum + 1): + try: + items.append(next(iterator)) + except StopIteration: + break + except Exception: + iteration_failed = True + items = [] + if iteration_failed: + raise ValueError("{} mapping iteration failed".format(name)) + if len(items) > maximum: + raise ValueError("{} mapping exceeds item limit".format(name)) + return items + + +def _canonical_config(value, name): + """Detach one bounded plain configuration without trusting Mapping.len.""" + canonical = {} + for key, item in _bounded_mapping_items( + value, + name, + MAX_CONFIG_ENTRIES, + ): + if type(key) is not str or not key or key != key.strip() or len(key) > MAX_CONFIG_KEY_LENGTH or _CONFIG_KEY.fullmatch(key) is None: + raise ValueError("{} contains invalid config key".format(name)) + if key in canonical: + raise ValueError("{} contains duplicate config key".format(name)) + canonical[key] = _canonical_value(item) + canonical = dict(sorted(canonical.items())) + if len(_canonical_json(canonical).encode("utf-8")) > MAX_CONFIG_BYTES: + raise ValueError("{} exceeds byte limit".format(name)) + return MappingProxyType(canonical) + + +def _configs_equal(left, right): + """Compare configs without bool/int/float equality ambiguity.""" + return _canonical_json(left) == _canonical_json(right) + + +def _config_digest(config): + """Digest one already-canonical complete config.""" + return hashlib.sha256(_canonical_json(config).encode("utf-8")).hexdigest() + + +def _feedback_digest(token): + """Retain only a one-way audit fingerprint of a feedback token.""" + return hashlib.sha256(token.encode("utf-8")).hexdigest() + + +def _require_tuple(value, name, maximum=MAX_SEMANTIC_ITEMS): + """Reject iterable coercion and bound an existing exact tuple.""" + if type(value) is not tuple: + raise ValueError("{} container type is invalid".format(name)) + if len(value) > maximum: + raise ValueError("{} exceeds item limit".format(name)) + return value + + +def _validate_provenance(value): + """Validate one bounded exact provenance tuple.""" + provenance = _require_tuple( + value, + "provenance", + MAX_PROVENANCE_ENTRIES, + ) + for source in provenance: + if type(source) is not FieldProvenance: + raise ValueError("provenance element type is invalid") + for item in ( + source.field_path, + source.provider_id, + source.source_path, + ): + if type(item) is not str or not item or len(item) > MAX_PROVENANCE_STRING_LENGTH: + raise ValueError("provenance string is invalid") + _validate_sequence( + source.generation, + "provenance generation", + minimum=0, + ) + return provenance + + +def _validate_provider_generations(value): + """Validate the full bounded provider generation cursor.""" + generations = _require_tuple( + value, + "provider_generations", + MAX_PROVENANCE_ENTRIES, + ) + seen = set() + for item in generations: + if type(item) is not tuple or len(item) != 2: + raise ValueError("provider generation item is invalid") + provider_id, generation = item + if type(provider_id) is not str or not provider_id or len(provider_id) > MAX_PROVIDER_ID_LENGTH: + raise ValueError("provider id is invalid") + _validate_sequence( + generation, + "provider generation", + minimum=0, + ) + if provider_id in seen: + raise ValueError("provider generation is duplicated") + seen.add(provider_id) + if generations != tuple(sorted(generations)): + raise ValueError("provider generations are not sorted") + return generations + + +def _provenance_payload(provenance): + """Return deterministic bounded audit data for provenance.""" + return [ + { + "field_path": source.field_path, + "provider_id": source.provider_id, + "generation": source.generation, + "source_path": source.source_path, + } + for source in provenance + ] + + +def _provenance_digest(provenance): + """Retain provenance accountability without raw durable paths.""" + return hashlib.sha256(_canonical_json(_provenance_payload(provenance)).encode("utf-8")).hexdigest() + + +def _sanitize_semantic(value, name, budget, depth=0): + """Recursively detach bounded compiler semantics before serialization.""" + if depth > MAX_SEMANTIC_DEPTH: + raise ValueError("{} depth exceeds limit".format(name)) + budget[0] += 1 + if budget[0] > MAX_SEMANTIC_NODES: + raise ValueError("{} node count exceeds limit".format(name)) + + if type(value) in (dict, _MAPPING_PROXY_TYPE): + sanitized = {} + for key, item in _bounded_mapping_items( + value, + name, + MAX_SEMANTIC_ITEMS, + ): + if type(key) is not str or not key or len(key) > MAX_SEMANTIC_STRING_LENGTH: + raise ValueError("{} key is invalid".format(name)) + if key in sanitized: + raise ValueError("{} key is duplicated".format(name)) + sanitized[key] = _sanitize_semantic( + item, + name, + budget, + depth + 1, + ) + return dict(sorted(sanitized.items())) + if type(value) in (tuple, list): + if len(value) > MAX_SEMANTIC_ITEMS: + raise ValueError("{} sequence exceeds item limit".format(name)) + return tuple(_sanitize_semantic(item, name, budget, depth + 1) for item in value) + if type(value) is frozenset: + if len(value) > MAX_SEMANTIC_ITEMS: + raise ValueError("{} set exceeds item limit".format(name)) + return tuple(sorted(_sanitize_semantic(item, name, budget, depth + 1) for item in value)) + if isinstance(value, Enum): + return _sanitize_semantic(value.value, name, budget, depth + 1) + if value is None or type(value) is bool: + return value + if type(value) is int: + _validate_sequence( + abs(value), + "{} integer".format(name), + minimum=0, + ) + return value + if type(value) is float: + if not math.isfinite(value): + raise ValueError("{} float is not finite".format(name)) + return value + if type(value) is str: + if len(value) > MAX_SEMANTIC_STRING_LENGTH: + raise ValueError("{} string exceeds limit".format(name)) + return value + raise ValueError("{} value type is invalid".format(name)) + + +def _plan_semantic_digest(plan, projection): + """Recompute the compiler digest from recursively bounded semantics.""" + aliases = _require_tuple(plan.aliases, "aliases") + identity_aliases = _require_tuple( + plan.identity_aliases, + "identity_aliases", + ) + role_assignments = _require_tuple( + plan.role_assignments, + "role_assignments", + ) + primary_targets = _require_tuple( + plan.primary_targets, + "primary_targets", + ) + control_targets = _require_tuple( + plan.control_targets, + "control_targets", + ) + arguments = _require_tuple( + plan.config_arguments, + "config_arguments", + MAX_CONFIG_ENTRIES, + ) + fields = _require_tuple(plan.fields, "fields") + topology = _sanitize_semantic( + plan.topology, + "topology", + [0], + ) + semantic = { + "topology": _semantic_topology(topology), + "aliases": [ + { + "qualified_name": binding.qualified_name, + "node_id": binding.node_id, + "roles": binding.roles, + } + for binding in aliases + ], + "identity_aliases": [ + { + "provider_id": binding.provider_id, + "kind": binding.kind, + "value": binding.value, + "local_node_id": binding.local_node_id, + "canonical_node_id": binding.canonical_node_id, + } + for binding in identity_aliases + ], + "role_assignments": [ + { + "provider_id": binding.provider_id, + "role": binding.role, + "group": binding.group, + "index": binding.index, + "local_node_id": binding.local_node_id, + "canonical_node_id": binding.canonical_node_id, + } + for binding in role_assignments + ], + "primary_targets": [ + { + "group": target.group, + "index": target.index, + "node_id": target.node_id, + } + for target in primary_targets + ], + "control_targets": [ + { + "group": target.group, + "index": target.index, + "node_id": target.node_id, + } + for target in control_targets + ], + "primary_target": plan.primary_target, + "control_target": plan.control_target, + "materialization_readiness": { + "ready": plan.materialization_readiness.ready, + "blockers": plan.materialization_readiness.blockers, + }, + "config_arguments": [ + { + "name": argument.name, + "value": projection[argument.name], + "cardinality": argument.cardinality, + "required": argument.required, + "transforms": argument.transforms, + } + for argument in arguments + ], + "projected_config": dict(projection), + "fields": [{"name": field.name, "value": field.value} for field in fields], + } + semantic = _sanitize_semantic( + semantic, + "plan semantics", + [0], + ) + encoded = _compiler_canonical_json(semantic).encode("utf-8") + if len(encoded) > MAX_SEMANTIC_BYTES: + raise ValueError("plan semantics exceed byte limit") + return hashlib.sha256(encoded).hexdigest() + + +def _operation_payload(operation): + """Return deterministic secret-free target operation data.""" + if operation is None: + return None + return { + "key": { + "fence": operation.key.fence, + "projection_digest": operation.key.projection_digest, + }, + "phase": operation.phase.value, + "handle": operation.handle, + "preimage_digest": operation.preimage_digest, + "candidate_digest": operation.candidate_digest, + "changed": operation.changed, + "terminal_phase": (operation.terminal_phase.value if operation.terminal_phase is not None else None), + } + + +def _key_payload(key): + """Return deterministic target key data.""" + if key is None: + return None + return { + "fence": key.fence, + "projection_digest": key.projection_digest, + } + + +def _state_integrity(state): + """Fingerprint every safety-relevant secret-free checkpoint field.""" + payload = { + "revision": state.revision, + "phase": state.phase.value, + "plan_digest": state.plan_digest, + "publication_digest": state.publication_digest, + "publication_version": state.publication_version, + "projection_digest": state.projection_digest, + "feedback_digest": state.feedback_digest, + "provenance_digest": state.provenance_digest, + "provenance_count": state.provenance_count, + "reservation_key": _key_payload(state.reservation_key), + "prior_operation_key": _key_payload(state.prior_operation_key), + "operation": _operation_payload(state.operation), + "applied_plan_digest": state.applied_plan_digest, + "applied_publication_digest": state.applied_publication_digest, + "applied_publication_version": state.applied_publication_version, + } + return hashlib.sha256(_canonical_json(payload).encode("utf-8")).hexdigest() + + +def _new_state(**values): + """Construct and validate one fingerprinted exact base state.""" + _validate_sequence(values["revision"], "revision", minimum=1) + if not isinstance(values["phase"], MaterializationPhase): + raise ValueError("state phase is invalid") + _validate_digest(values["plan_digest"], "plan_digest") + _validate_digest(values["publication_digest"], "publication_digest") + _validate_sequence( + values["publication_version"], + "publication_version", + minimum=1, + ) + _validate_digest(values["projection_digest"], "projection_digest") + _validate_digest(values["feedback_digest"], "feedback_digest") + _validate_digest(values["provenance_digest"], "provenance_digest") + _validate_sequence( + values["provenance_count"], + "provenance_count", + minimum=0, + maximum=MAX_PROVENANCE_ENTRIES, + ) + if type(values["reservation_key"]) is not TargetOperationKey: + raise ValueError("reservation key type is invalid") + TargetOperationKey.__post_init__(values["reservation_key"]) + if values["prior_operation_key"] is not None: + if type(values["prior_operation_key"]) is not TargetOperationKey: + raise ValueError("prior operation key type is invalid") + TargetOperationKey.__post_init__(values["prior_operation_key"]) + if values["operation"] is not None: + if type(values["operation"]) is not TargetOperationSnapshot: + raise ValueError("target operation type is invalid") + TargetOperationSnapshot.__post_init__(values["operation"]) + if values["applied_publication_version"] is not None: + _validate_sequence( + values["applied_publication_version"], + "applied_publication_version", + minimum=1, + ) + values["integrity_digest"] = "" + provisional = AtomicMaterializationState.__new__(AtomicMaterializationState) + for name, value in values.items(): + object.__setattr__(provisional, name, value) + values["integrity_digest"] = _state_integrity(provisional) + return AtomicMaterializationState(**values) + + +def _validate_envelope(envelope): + """Validate and fingerprint all input before any external seam call.""" + if type(envelope) is not PublishedMaterializationRequest: + raise MaterializationValidationError("publication envelope type is invalid") + failure_kind = None + try: + PublishedMaterializationRequest.__post_init__(envelope) + plan = envelope.request.plan + readiness = plan.materialization_readiness + if type(readiness) is not MaterializationReadiness or not isinstance(readiness.ready, bool) or type(readiness.blockers) is not tuple: + raise ValueError("readiness shape is invalid") + if readiness.ready or readiness.blockers != ("atomic_materializer_missing",): + raise ValueError("readiness blocker is invalid") + _validate_digest(plan.digest, "plan_digest") + projection = _canonical_config( + plan.projected_config, + "projected_config", + ) + arguments = _require_tuple( + plan.config_arguments, + "config_arguments", + MAX_CONFIG_ENTRIES, + ) + if any(type(argument) is not ProjectedConfigArgument for argument in arguments): + raise ValueError("config argument type is invalid") + names = tuple(argument.name for argument in arguments) + if names != tuple(sorted(set(names))): + raise ValueError("config argument names are invalid") + if names != tuple(projection): + raise ValueError("config projection names are invalid") + for argument in arguments: + if not _configs_equal( + {"value": argument.value}, + {"value": projection[argument.name]}, + ): + raise ValueError("config argument binding is invalid") + provenance = _validate_provenance(plan.provenance) + _validate_provider_generations(plan.provider_generations) + if _plan_semantic_digest(plan, projection) != plan.digest: + raise ValueError("semantic digest binding is invalid") + except Exception as exc: + failure_kind = type(exc).__name__ + projection = None + provenance = None + if failure_kind is not None: + raise MaterializationValidationError("publication envelope validation failed ({})".format(failure_kind)) + return ( + projection, + _config_digest(projection), + _feedback_digest(envelope.request.feedback_token), + _provenance_digest(provenance), + len(provenance), + ) + + +class AtomicConfigMaterializer: + """Fail-closed two-phase durable reservation coordinator.""" + + def __init__(self, state_store=None, target=None, enabled=False): + """Create a default-off, explicitly injected coordinator.""" + if type(enabled) is not bool: + raise ValueError("enabled flag is invalid") + self._enabled = enabled + self._state_store = state_store + self._target = target + self._lock = threading.RLock() + if enabled: + for owner, methods, name in ( + ( + state_store, + ("allocate_fence", "load", "compare_and_store"), + "state_store", + ), + ( + target, + ( + "prepare", + "lookup", + "apply", + "rollback", + "seal", + "abort", + "invalidate", + ), + "target", + ), + ): + if owner is None or any(not callable(getattr(owner, method, None)) for method in methods): + raise ValueError("{} atomic seam is invalid".format(name)) + + @property + def enabled(self): + """Return the explicit feature gate.""" + return self._enabled + + def _result(self, status, state, target_written=False): + """Build one immutable public result.""" + return AtomicMaterializationResult( + status, + state, + target_written, + ) + + def _load_state(self): + """Load and freshly validate one exact base checkpoint.""" + failure_kind = None + try: + state = self._state_store.load() + except Exception as exc: + failure_kind = type(exc).__name__ + state = None + if failure_kind is not None: + raise MaterializationStorageError("state load failed ({})".format(failure_kind)) + if state is None: + return None + if type(state) is not AtomicMaterializationState: + raise MaterializationStorageError("state checkpoint type is invalid") + validation_failed = False + try: + AtomicMaterializationState.__post_init__(state) + except Exception: + validation_failed = True + if validation_failed: + raise MaterializationStorageError("state checkpoint validation failed") + return state + + def _compare_store(self, expected, replacement): + """Return ``(stored, current)`` with exact ack-loss recovery.""" + failure_kind = None + try: + committed = self._state_store.compare_and_store( + expected, + replacement, + ) + except Exception as exc: + failure_kind = type(exc).__name__ + committed = None + current = self._load_state() + if current == replacement: + return True, replacement + if failure_kind is not None: + raise MaterializationStorageError("state CAS failed ambiguously ({})".format(failure_kind)) + if committed is True: + raise MaterializationStorageError("state CAS acknowledgement was not durable") + if current == expected: + raise MaterializationStorageError("state CAS rejected without competing checkpoint") + return False, current + + def _allocate_fence(self, previous): + """Allocate and validate a bounded globally monotonic fence.""" + failure_kind = None + try: + fence = self._state_store.allocate_fence() + except Exception as exc: + failure_kind = type(exc).__name__ + fence = None + if failure_kind is not None: + raise MaterializationStorageError("fence allocation failed ({})".format(failure_kind)) + try: + _validate_sequence(fence, "fence", minimum=1) + except ValueError: + raise MaterializationStorageError("allocated fence is invalid") + if previous is not None and fence <= previous.reservation_key.fence: + raise MaterializationStorageError("allocated fence is not monotonic") + return fence + + def _target_snapshot_call(self, name, *args, allow_none=False): + """Call a target seam and validate a value-free snapshot result.""" + failure_kind = None + try: + snapshot = getattr(self._target, name)(*args) + except Exception as exc: + failure_kind = type(exc).__name__ + snapshot = None + if failure_kind is not None: + raise MaterializationTargetError("target {} failed ({})".format(name, failure_kind)) + if snapshot is None and allow_none: + return None + if type(snapshot) is not TargetOperationSnapshot: + raise MaterializationTargetError("target {} returned invalid type".format(name)) + validation_failed = False + try: + TargetOperationSnapshot.__post_init__(snapshot) + except Exception: + validation_failed = True + if validation_failed: + raise MaterializationTargetError("target {} returned invalid snapshot".format(name)) + return snapshot + + def _lookup(self, key): + """Lookup one exact target reservation key.""" + snapshot = self._target_snapshot_call( + "lookup", + key.fence, + key.projection_digest, + allow_none=True, + ) + if snapshot is not None and snapshot.key != key: + raise MaterializationTargetError("target lookup key mismatch") + return snapshot + + def _require_sealed(self, snapshot, key, operation_name): + """Validate one acknowledged stable terminal target result.""" + if snapshot.key != key: + raise MaterializationTargetError("target {} key mismatch".format(operation_name)) + if snapshot.phase is not TargetOperationPhase.SEALED or not isinstance( + snapshot.terminal_phase, + TargetTerminalPhase, + ): + raise MaterializationTargetError("target {} was not sealed".format(operation_name)) + return snapshot + + def _same_operation_identity(self, left, right): + """Compare immutable target identity independently of lifecycle phase.""" + if left is None or right is None: + return left is right + return ( + left.key, + left.handle, + left.preimage_digest, + left.candidate_digest, + left.changed, + ) == ( + right.key, + right.handle, + right.preimage_digest, + right.candidate_digest, + right.changed, + ) + + def _invalidate(self, key): + """Fence and seal one operation key.""" + snapshot = self._target_snapshot_call( + "invalidate", + key, + ) + return self._require_sealed(snapshot, key, "invalidate") + + def _abort(self, key): + """Abort and seal one possibly unprepared reservation.""" + snapshot = self._target_snapshot_call( + "abort", + key.fence, + key.projection_digest, + ) + return self._require_sealed(snapshot, key, "abort") + + def _operation_key_for_state(self, state): + """Return the durable target key represented by a state.""" + if state is None: + return None + return state.reservation_key + + def _publication_order( + self, + envelope, + state, + feedback_digest, + provenance_digest, + provenance_count, + ): + """Compare an input publication with the durable observed cursor.""" + if envelope.publication_version < state.publication_version: + return -1 + if envelope.publication_version > state.publication_version: + return 1 + if envelope.publication_digest != state.publication_digest: + raise MaterializationValidationError("publication version identity conflict") + if feedback_digest != state.feedback_digest or provenance_digest != state.provenance_digest or provenance_count != state.provenance_count: + raise MaterializationValidationError("publication audit identity conflict") + return 0 + + def _terminal_result(self, state): + """Map one stable sealed durable state to its public result.""" + if state.phase is MaterializationPhase.APPLIED: + return self._result(MaterializationStatus.UNCHANGED, state) + if state.phase is MaterializationPhase.ROLLED_BACK: + return self._result(MaterializationStatus.NOT_APPLIED, state) + return self._result(MaterializationStatus.UNKNOWN, state) + + def _require_revision_headroom(self, state, steps=1): + """Fail before a target seam if durable finalization cannot advance.""" + _validate_sequence(state.revision, "revision", minimum=1) + if steps < 1 or state.revision > MAX_SEQUENCE - steps: + raise MaterializationStorageError("state revision has no transition headroom") + + def _with_operation(self, state, phase, operation): + """Build a value-free validated state transition.""" + transition_failed = False + try: + replacement = state.with_operation(phase, operation) + except Exception: + transition_failed = True + replacement = None + if transition_failed: + raise MaterializationStorageError("state transition validation failed") + return replacement + + def _advance_publication( + self, + state, + envelope, + feedback_digest, + provenance_digest, + provenance_count, + ): + """Build a value-free publication cursor transition.""" + transition_failed = False + try: + replacement = state.advance_publication( + envelope, + feedback_digest, + provenance_digest, + provenance_count, + ) + except Exception: + transition_failed = True + replacement = None + if transition_failed: + raise MaterializationStorageError("publication state transition failed") + return replacement + + def _persist_terminal(self, base_state, sealed): + """Persist a sealed result despite same-key transition races.""" + if base_state.operation is not None and not self._same_operation_identity( + base_state.operation, + sealed, + ): + return self._result( + MaterializationStatus.UNKNOWN, + self._load_state(), + ) + current = self._load_state() + for _attempt in range(MAX_RESERVATION_RETRIES): + if current is None: + raise MaterializationStorageError("terminal reservation disappeared") + if current.reservation_key != base_state.reservation_key: + return self._result( + MaterializationStatus.SUPERSEDED, + current, + ) + if sealed.terminal_phase is TargetTerminalPhase.APPLIED: + phase = MaterializationPhase.APPLIED + status = MaterializationStatus.APPLIED + elif sealed.terminal_phase is TargetTerminalPhase.ROLLED_BACK: + phase = MaterializationPhase.ROLLED_BACK + status = MaterializationStatus.NOT_APPLIED + else: + phase = MaterializationPhase.UNKNOWN + status = MaterializationStatus.UNKNOWN + replacement = self._with_operation( + current, + phase, + sealed, + ) + stored, observed = self._compare_store(current, replacement) + if stored: + return self._result( + status, + replacement, + target_written=(sealed.changed is True and sealed.terminal_phase is TargetTerminalPhase.APPLIED), + ) + current = observed + raise MaterializationStorageError("terminal state CAS retry limit exceeded") + + def _settle(self, state, snapshot): + """Seal/invalidate before any terminal materializer result.""" + self._require_revision_headroom(state) + key = state.reservation_key + if not self._same_operation_identity(state.operation, snapshot): + try: + self._invalidate(key) + except MaterializationTargetError: + pass + return self._result( + MaterializationStatus.UNKNOWN, + self._load_state(), + ) + try: + if snapshot.phase is TargetOperationPhase.SEALED: + sealed = self._require_sealed( + snapshot, + key, + "operation", + ) + elif snapshot.phase in ( + TargetOperationPhase.APPLIED, + TargetOperationPhase.ROLLED_BACK, + ): + sealed = self._target_snapshot_call( + "seal", + snapshot, + ) + sealed = self._require_sealed( + sealed, + key, + "seal", + ) + else: + sealed = self._invalidate(key) + except MaterializationTargetError: + return self._result( + MaterializationStatus.UNKNOWN, + self._load_state(), + ) + return self._persist_terminal(state, sealed) + + def _recover_state(self, state): + """Recover one durable phase without racing a late same-fence commit.""" + if state.phase in ( + MaterializationPhase.APPLIED, + MaterializationPhase.ROLLED_BACK, + ): + try: + snapshot = self._lookup(state.reservation_key) + except MaterializationTargetError: + return self._result(MaterializationStatus.UNKNOWN, state) + if snapshot is None or snapshot != state.operation or snapshot.phase is not TargetOperationPhase.SEALED: + return self._result(MaterializationStatus.UNKNOWN, state) + return self._terminal_result(state) + + self._require_revision_headroom(state) + try: + if state.prior_operation_key is not None: + self._invalidate(state.prior_operation_key) + snapshot = self._lookup(state.reservation_key) + if snapshot is None: + sealed = self._abort(state.reservation_key) + elif state.operation is not None and not self._same_operation_identity( + state.operation, + snapshot, + ): + try: + self._invalidate(state.reservation_key) + except MaterializationTargetError: + pass + return self._result( + MaterializationStatus.UNKNOWN, + state, + ) + elif snapshot.phase is TargetOperationPhase.SEALED: + sealed = self._require_sealed( + snapshot, + state.reservation_key, + "lookup", + ) + else: + sealed = self._invalidate(state.reservation_key) + except MaterializationTargetError: + return self._result(MaterializationStatus.UNKNOWN, state) + return self._persist_terminal(state, sealed) + + def recover(self): + """Recover a reservation through target invalidation and sealing.""" + if not self._enabled: + return self._result(MaterializationStatus.DISABLED, None) + with self._lock: + state = self._load_state() + if state is None: + return self._result( + MaterializationStatus.NOT_APPLIED, + None, + ) + return self._recover_state(state) + + def _reserve( + self, + envelope, + projection_digest, + feedback_digest, + provenance_digest, + provenance_count, + ): + """Win a durable reservation before any target prepare call.""" + current = self._load_state() + for _attempt in range(MAX_RESERVATION_RETRIES): + if current is not None: + order = self._publication_order( + envelope, + current, + feedback_digest, + provenance_digest, + provenance_count, + ) + if order < 0: + return None, self._result( + MaterializationStatus.SUPERSEDED, + current, + ) + same_plan = envelope.request.plan.digest == current.plan_digest and projection_digest == current.projection_digest + if current.phase is MaterializationPhase.APPLIED and same_plan: + if order == 0: + return None, self._result( + MaterializationStatus.UNCHANGED, + current, + ) + advanced = self._advance_publication( + current, + envelope, + feedback_digest, + provenance_digest, + provenance_count, + ) + stored, observed = self._compare_store( + current, + advanced, + ) + if stored: + return None, self._result( + MaterializationStatus.UNCHANGED, + advanced, + ) + current = observed + continue + if order == 0 and current.phase not in ( + MaterializationPhase.ROLLED_BACK, + MaterializationPhase.UNKNOWN, + ): + return None, self._result( + MaterializationStatus.UNKNOWN, + current, + ) + + fence = self._allocate_fence(current) + reserved = AtomicMaterializationState.reserve( + 1, + envelope, + projection_digest, + feedback_digest, + provenance_digest, + provenance_count, + fence, + self._operation_key_for_state(current), + ) + stored, observed = self._compare_store( + current, + reserved, + ) + if stored: + return reserved, None + current = observed + raise MaterializationStorageError("reservation CAS retry limit exceeded") + + def _attach_operation(self, reserved, operation): + """Attach target metadata only while the reservation remains current.""" + prepared = self._with_operation( + reserved, + MaterializationPhase.PREPARED, + operation, + ) + stored, current = self._compare_store(reserved, prepared) + if stored: + return prepared, None + try: + self._invalidate(reserved.reservation_key) + except MaterializationTargetError: + pass + if current is not None: + return None, self._result( + MaterializationStatus.SUPERSEDED, + current, + ) + raise MaterializationStorageError("prepared reservation disappeared") + + def _mark_applying(self, prepared): + """Durably announce APPLYING before invoking target.apply.""" + self._require_revision_headroom(prepared, steps=2) + applying = self._with_operation( + prepared, + MaterializationPhase.APPLYING, + prepared.operation, + ) + stored, current = self._compare_store(prepared, applying) + if stored: + return applying, None + try: + self._invalidate(prepared.reservation_key) + except MaterializationTargetError: + pass + return None, self._result( + MaterializationStatus.SUPERSEDED, + current, + ) + + def apply(self, envelope): + """Reserve, prepare, apply, seal, and checkpoint one publication.""" + if not self._enabled: + return self._result(MaterializationStatus.DISABLED, None) + ( + projection, + projection_digest, + feedback_digest, + provenance_digest, + provenance_count, + ) = _validate_envelope(envelope) + with self._lock: + reserved, terminal = self._reserve( + envelope, + projection_digest, + feedback_digest, + provenance_digest, + provenance_count, + ) + if terminal is not None: + return terminal + + self._require_revision_headroom(reserved, steps=3) + try: + if reserved.prior_operation_key is not None: + self._invalidate(reserved.prior_operation_key) + except MaterializationTargetError: + return self._result( + MaterializationStatus.UNKNOWN, + reserved, + ) + + try: + operation = self._target_snapshot_call( + "prepare", + reserved.reservation_key.fence, + reserved.projection_digest, + dict(projection), + ) + if operation.key != reserved.reservation_key or operation.phase is not TargetOperationPhase.PREPARED: + raise MaterializationTargetError("target prepare acknowledgement is invalid") + except MaterializationTargetError: + return self._result( + MaterializationStatus.UNKNOWN, + reserved, + ) + + prepared, terminal = self._attach_operation( + reserved, + operation, + ) + if terminal is not None: + return terminal + applying, terminal = self._mark_applying(prepared) + if terminal is not None: + return terminal + + self._require_revision_headroom(applying) + try: + snapshot = self._target_snapshot_call( + "apply", + operation, + ) + except MaterializationTargetError: + try: + snapshot = self._lookup(applying.reservation_key) + except MaterializationTargetError: + snapshot = None + if snapshot is None: + return self._result( + MaterializationStatus.UNKNOWN, + applying, + ) + return self._settle(applying, snapshot) + + def rollback(self): + """Rollback an exact sealed APPLIED transaction and reseal it.""" + if not self._enabled: + return self._result(MaterializationStatus.DISABLED, None) + with self._lock: + state = self._load_state() + if state is None: + return self._result( + MaterializationStatus.NOT_APPLIED, + None, + ) + if state.phase not in ( + MaterializationPhase.APPLIED, + MaterializationPhase.ROLLED_BACK, + ): + recovered = self._recover_state(state) + state = recovered.state + if ( + recovered.status + in ( + MaterializationStatus.UNKNOWN, + MaterializationStatus.SUPERSEDED, + ) + or state is None + ): + return recovered + if state.phase is MaterializationPhase.ROLLED_BACK: + return self._result( + MaterializationStatus.NOT_APPLIED, + state, + ) + self._require_revision_headroom(state, steps=2) + rolling = self._with_operation( + state, + MaterializationPhase.ROLLING_BACK, + state.operation, + ) + stored, current = self._compare_store(state, rolling) + if not stored: + return self._result( + MaterializationStatus.SUPERSEDED, + current, + ) + try: + snapshot = self._target_snapshot_call( + "rollback", + state.operation, + ) + except MaterializationTargetError: + try: + snapshot = self._lookup(rolling.reservation_key) + except MaterializationTargetError: + snapshot = None + if snapshot is None: + return self._result( + MaterializationStatus.UNKNOWN, + rolling, + ) + settled = self._settle(rolling, snapshot) + if settled.status is MaterializationStatus.NOT_APPLIED: + return self._result( + MaterializationStatus.ROLLED_BACK, + settled.state, + target_written=(settled.state.operation.changed is True), + ) + return settled diff --git a/apps/predbat/tests/test_lattice_atomic_materializer.py b/apps/predbat/tests/test_lattice_atomic_materializer.py new file mode 100644 index 000000000..aaaa425e2 --- /dev/null +++ b/apps/predbat/tests/test_lattice_atomic_materializer.py @@ -0,0 +1,1340 @@ +"""Tests for the two-phase atomic Lattice auto-config materializer.""" + +# cspell:ignore autoconfig dedupe preimage readback + +import copy +import itertools +import math +import os +import sys +import threading +import unittest +from collections.abc import Mapping +from dataclasses import replace +from types import MappingProxyType + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from lattice_atomic_materializer import ( # noqa: E402 + MAX_SEQUENCE, + AtomicConfigMaterializer, + AtomicMaterializationState, + MaterializationPhase, + MaterializationStatus, + MaterializationStorageError, + MaterializationValidationError, + PublishedMaterializationRequest, + TargetOperationKey, + TargetOperationPhase, + TargetOperationSnapshot, + TargetTerminalPhase, + _canonical_config, + _config_digest, + _configs_equal, + _validate_envelope, +) +from lattice_autoconfig import ( # noqa: E402 + MaterializationReadiness, + MaterializationRequest, + ProjectionCardinality, + ProjectionRouting, + ProjectionValueKind, + ProviderConfigProjection, + compile_auto_config, +) +from tests.test_lattice_autoconfig import ( # noqa: E402 + indexed_roles, + projection_snapshot, + projection_value, +) + + +class InMemoryAtomicStateStore: + """Thread-safe durable fence/CAS store with deterministic cuts.""" + + def __init__(self, state=None): + """Create a store with optional durable state.""" + self._lock = threading.RLock() + self.state = state + self.fence = 0 if state is None else state.reservation_key.fence + self.writes = 0 + self.calls = 0 + self.history = [] + self.fail_on_calls = {} + self.corrupt_load = None + self.raise_load = False + self.raise_fence = False + self.fence_override = None + self.block_cas_call = None + self.cas_entered = None + self.cas_release = None + + def allocate_fence(self): + """Allocate a globally increasing fence.""" + with self._lock: + if self.raise_fence: + raise OSError("secret-store-fence-error") + if self.fence_override is not None: + return self.fence_override + self.fence += 1 + return self.fence + + def load(self): + """Return current state or one injected failure.""" + with self._lock: + if self.raise_load: + raise OSError("secret-store-load-error") + if self.corrupt_load is not None: + return self.corrupt_load + return self.state + + def compare_and_store(self, expected, replacement): + """CAS with deterministic conflict and acknowledgement faults.""" + with self._lock: + self.calls += 1 + call = self.calls + if call == self.block_cas_call: + self.cas_entered.set() + self.cas_release.wait(5) + with self._lock: + mode = self.fail_on_calls.get(call) + if mode == "reject": + return False + if mode == "raise_before": + raise OSError("secret-store-cas-error") + if self.state != expected: + return False + self.state = replacement + self.writes += 1 + self.history.append(replacement) + if mode == "raise_after": + raise OSError("secret-store-cas-error") + return True + + +class InMemoryWholeConfigTarget: + """Durable target journal that alone retains raw configurations.""" + + def __init__(self, config=None): + """Create an exact fenced target with lifecycle fault injection.""" + self._lock = threading.RLock() + self.config = copy.deepcopy(config or {}) + self.records = {} + self.current_fence = 0 + self.prepare_calls = 0 + self.lookup_calls = 0 + self.apply_calls = 0 + self.rollback_calls = 0 + self.seal_calls = 0 + self.abort_calls = 0 + self.invalidate_calls = 0 + self.writes = 0 + self.history = [] + self.raise_prepare = False + self.raise_lookup = False + self.raise_apply_before = False + self.raise_apply_after = False + self.raise_rollback_before = False + self.raise_rollback_after = False + self.raise_seal = False + self.raise_abort = False + self.raise_invalidate = False + self.lookup_override = None + self.partial_apply = None + self.block_prepare = False + self.prepare_entered = None + self.prepare_release = None + self.block_apply = False + self.apply_entered = None + self.apply_release = None + self.block_rollback = False + self.rollback_entered = None + self.rollback_release = None + self.reservation_probe = None + + @staticmethod + def _tuple_key(key): + """Return a dictionary key for one exact operation key.""" + return (key.fence, key.projection_digest) + + def _key(self, fence, projection_digest): + """Build one exact operation key.""" + return TargetOperationKey(fence, projection_digest) + + def _snapshot(self, record): + """Return secret-free durable journal metadata.""" + if record["prepared"]: + return TargetOperationSnapshot( + key=record["key"], + phase=record["phase"], + handle=record["handle"], + preimage_digest=_config_digest(record["preimage"]), + candidate_digest=_config_digest(record["candidate"]), + changed=not _configs_equal( + record["preimage"], + record["candidate"], + ), + terminal_phase=record["terminal"], + ) + return TargetOperationSnapshot( + key=record["key"], + phase=TargetOperationPhase.SEALED, + terminal_phase=TargetTerminalPhase.ROLLED_BACK, + ) + + def _record_for_snapshot(self, snapshot): + """Resolve an exact snapshot without trusting only its handle.""" + record = self.records.get(self._tuple_key(snapshot.key)) + if record is None or not record["prepared"]: + return None + current = self._snapshot(record) + if snapshot.key != current.key or snapshot.handle != current.handle or snapshot.preimage_digest != current.preimage_digest or snapshot.candidate_digest != current.candidate_digest: + return None + return record + + def prepare(self, fence, projection_digest, projection): + """Create PREPARED only after observing a winning reservation.""" + if self.block_prepare: + self.prepare_entered.set() + self.prepare_release.wait(5) + with self._lock: + self.prepare_calls += 1 + if self.reservation_probe is not None: + self.reservation_probe(fence, projection_digest) + if self.raise_prepare: + raise OSError("secret-target-prepare-error") + key = self._key(fence, projection_digest) + tuple_key = self._tuple_key(key) + if fence < self.current_fence or tuple_key in self.records: + raise ValueError("stale or replayed prepare") + preimage = _canonical_config( + copy.deepcopy(self.config), + "target preimage", + ) + projection = _canonical_config( + copy.deepcopy(projection), + "target projection", + ) + if _config_digest(projection) != projection_digest: + raise ValueError("projection digest mismatch") + candidate = dict(preimage) + candidate.update(dict(projection)) + candidate = _canonical_config(candidate, "target candidate") + record = { + "key": key, + "prepared": True, + "handle": "txn-{}".format(fence), + "preimage": preimage, + "candidate": candidate, + "phase": TargetOperationPhase.PREPARED, + "terminal": None, + } + self.current_fence = fence + self.records[tuple_key] = record + return self._snapshot(record) + + def lookup(self, fence, projection_digest): + """Recover a journal record by reservation key.""" + with self._lock: + self.lookup_calls += 1 + if self.raise_lookup: + raise OSError("secret-target-lookup-error") + if self.lookup_override is not None: + return self.lookup_override + record = self.records.get((fence, projection_digest)) + return None if record is None else self._snapshot(record) + + def apply(self, operation): + """Commit only after rechecking current fence and operation phase.""" + with self._lock: + self.apply_calls += 1 + record = self._record_for_snapshot(operation) + if record is None or record["key"].fence != self.current_fence or record["phase"] is not TargetOperationPhase.PREPARED: + if record is None: + return operation + return self._snapshot(record) + record["phase"] = TargetOperationPhase.APPLYING + if self.raise_apply_before: + raise OSError("secret-target-apply-error") + if self.block_apply: + self.apply_entered.set() + self.apply_release.wait(5) + with self._lock: + if record["key"].fence != self.current_fence or record["phase"] is not TargetOperationPhase.APPLYING: + return self._snapshot(record) + if not _configs_equal(self.config, record["preimage"]): + record["phase"] = TargetOperationPhase.UNKNOWN + return self._snapshot(record) + if self.partial_apply is None: + self.config = copy.deepcopy(dict(record["candidate"])) + else: + self.config = copy.deepcopy(self.partial_apply) + self.writes += 1 + self.history.append(copy.deepcopy(self.config)) + if _configs_equal(self.config, record["candidate"]): + record["phase"] = TargetOperationPhase.APPLIED + else: + record["phase"] = TargetOperationPhase.UNKNOWN + if self.raise_apply_after: + raise OSError("secret-target-apply-error") + return self._snapshot(record) + + def rollback(self, operation): + """Rollback only the current sealed APPLIED handle.""" + with self._lock: + self.rollback_calls += 1 + record = self._record_for_snapshot(operation) + if record is None or record["key"].fence != self.current_fence or record["phase"] is not TargetOperationPhase.SEALED or record["terminal"] is not TargetTerminalPhase.APPLIED: + if record is None: + return operation + return self._snapshot(record) + record["phase"] = TargetOperationPhase.ROLLING_BACK + record["terminal"] = None + if self.raise_rollback_before: + raise OSError("secret-target-rollback-error") + if self.block_rollback: + self.rollback_entered.set() + self.rollback_release.wait(5) + with self._lock: + if record["key"].fence != self.current_fence or record["phase"] is not TargetOperationPhase.ROLLING_BACK: + return self._snapshot(record) + if not _configs_equal(self.config, record["candidate"]): + record["phase"] = TargetOperationPhase.UNKNOWN + return self._snapshot(record) + self.config = copy.deepcopy(dict(record["preimage"])) + self.writes += 1 + self.history.append(copy.deepcopy(self.config)) + record["phase"] = TargetOperationPhase.ROLLED_BACK + if self.raise_rollback_after: + raise OSError("secret-target-rollback-error") + return self._snapshot(record) + + def seal(self, operation): + """Seal a stable result and make apply replay impossible.""" + with self._lock: + self.seal_calls += 1 + if self.raise_seal: + raise OSError("secret-target-seal-error") + record = self._record_for_snapshot(operation) + if record is None: + raise ValueError("unknown operation") + if record["phase"] is TargetOperationPhase.SEALED: + return self._snapshot(record) + if record["phase"] is TargetOperationPhase.APPLIED: + terminal = TargetTerminalPhase.APPLIED + elif record["phase"] in ( + TargetOperationPhase.ROLLED_BACK, + TargetOperationPhase.PREPARED, + ): + terminal = TargetTerminalPhase.ROLLED_BACK + else: + raise ValueError("unstable operation") + record["phase"] = TargetOperationPhase.SEALED + record["terminal"] = terminal + return self._snapshot(record) + + def _invalidate_record(self, record): + """Atomically fence and seal one record from its exact visibility.""" + if record["phase"] is TargetOperationPhase.SEALED: + return self._snapshot(record) + if not record["prepared"]: + return self._snapshot(record) + if _configs_equal(self.config, record["candidate"]): + terminal = TargetTerminalPhase.APPLIED + elif _configs_equal(self.config, record["preimage"]): + terminal = TargetTerminalPhase.ROLLED_BACK + else: + terminal = TargetTerminalPhase.UNKNOWN + record["phase"] = TargetOperationPhase.SEALED + record["terminal"] = terminal + return self._snapshot(record) + + def abort(self, fence, projection_digest): + """Tombstone even an unprepared reservation.""" + with self._lock: + self.abort_calls += 1 + if self.raise_abort: + raise OSError("secret-target-abort-error") + key = self._key(fence, projection_digest) + tuple_key = self._tuple_key(key) + record = self.records.get(tuple_key) + if record is None: + record = { + "key": key, + "prepared": False, + "phase": TargetOperationPhase.SEALED, + "terminal": TargetTerminalPhase.ROLLED_BACK, + } + self.records[tuple_key] = record + self.current_fence = max(self.current_fence, fence) + return self._invalidate_record(record) + + def invalidate(self, operation_key): + """Fence a key so any late commit recheck fails.""" + with self._lock: + self.invalidate_calls += 1 + if self.raise_invalidate: + raise OSError("secret-target-invalidate-error") + tuple_key = self._tuple_key(operation_key) + record = self.records.get(tuple_key) + if record is None: + record = { + "key": operation_key, + "prepared": False, + "phase": TargetOperationPhase.SEALED, + "terminal": TargetTerminalPhase.ROLLED_BACK, + } + self.records[tuple_key] = record + self.current_fence = max( + self.current_fence, + operation_key.fence, + ) + return self._invalidate_record(record) + + +class ExplosiveSeam: + """Disabled-mode sentinel that must never be touched.""" + + def __getattr__(self, _name): + """Fail if any seam operation is inspected or called.""" + raise AssertionError("disabled materializer touched seam") + + +class InfiniteMapping(Mapping): + """Mapping whose items iterator never terminates.""" + + def __getitem__(self, key): + raise KeyError(key) + + def __iter__(self): + return itertools.count() + + def __len__(self): + return 0 + + def items(self): + return (("key_{}".format(index), index) for index in itertools.count()) + + +class FaultingMapping(Mapping): + """Mapping whose item traversal raises a secret-bearing error.""" + + def __getitem__(self, key): + raise KeyError(key) + + def __iter__(self): + raise OSError("secret-topology-error") + + def __len__(self): + raise OSError("secret-topology-error") + + def items(self): + raise OSError("secret-topology-error") + + +def config_plan(value=7): + """Compile one projection-complete scalar config plan.""" + node = "INV1" + projection = ProviderConfigProjection( + argument="battery_rate_max", + role=indexed_roles((node,))[0].role, + group="inverters", + routing=ProjectionRouting.LEAF, + cardinality=ProjectionCardinality.SCALAR, + values=( + projection_value( + node, + ProjectionValueKind.CONSTANT, + value, + ), + ), + ) + provider = projection_snapshot( + "gateway", + (node,), + indexed_roles((node,)), + (projection,), + ) + return compile_auto_config((provider,)) + + +def values_plan(): + """Compile explicit None, list, scalar, boolean, and string values.""" + nodes = ("INV1", "INV2") + roles = indexed_roles(nodes) + projections = ( + ProviderConfigProjection( + "battery_power", + roles[0].role, + "inverters", + ProjectionRouting.LEAF, + ProjectionCardinality.PER_INDEX, + ( + projection_value( + nodes[0], + ProjectionValueKind.CONSTANT, + 5, + ), + projection_value( + nodes[1], + ProjectionValueKind.NONE, + ), + ), + required=False, + ), + ProviderConfigProjection( + "givtcp_rest", + roles[0].role, + "inverters", + ProjectionRouting.LEAF, + ProjectionCardinality.SCALAR, + ( + projection_value( + nodes[0], + ProjectionValueKind.NONE, + ), + ), + required=False, + ), + ProviderConfigProjection( + "inverter_type", + roles[0].role, + "inverters", + ProjectionRouting.LEAF, + ProjectionCardinality.SCALAR, + ( + projection_value( + nodes[0], + ProjectionValueKind.CONSTANT, + "GEC", + ), + ), + ), + ProviderConfigProjection( + "set_read_only", + roles[0].role, + "inverters", + ProjectionRouting.LEAF, + ProjectionCardinality.SCALAR, + ( + projection_value( + nodes[0], + ProjectionValueKind.CONSTANT, + True, + ), + ), + ), + ) + provider = projection_snapshot( + "gateway", + nodes, + roles, + projections, + ) + return compile_auto_config((provider,)) + + +def envelope(plan=None, version=1, token="lattice-publication-1-test"): + """Bind one plan to its publication cursor.""" + plan = plan or config_plan() + return PublishedMaterializationRequest( + MaterializationRequest(plan, token), + plan.digest, + version, + ) + + +def enabled_materializer(config=None, store=None, target=None): + """Create one enabled coordinator and its reference seams.""" + store = store or InMemoryAtomicStateStore() + target = target or InMemoryWholeConfigTarget(config) + materializer = AtomicConfigMaterializer( + store, + target, + enabled=True, + ) + return materializer, store, target + + +def reserve_only(materializer, request): + """Create one winning RESERVED state without touching the target.""" + ( + _projection, + projection_digest, + feedback_digest, + provenance_digest, + provenance_count, + ) = _validate_envelope(request) + reserved, terminal = materializer._reserve( + request, + projection_digest, + feedback_digest, + provenance_digest, + provenance_count, + ) + if terminal is not None: + raise AssertionError("reservation unexpectedly returned terminal result") + return reserved + + +class TestAtomicConfigMaterializer(unittest.TestCase): + """Reservations and target operations remain atomic across every cut.""" + + def test_default_off_touches_nothing(self): + """The unregistered default cannot discover or mutate live config.""" + materializer = AtomicConfigMaterializer( + ExplosiveSeam(), + ExplosiveSeam(), + ) + + result = materializer.apply(object()) + + self.assertEqual(result.status, MaterializationStatus.DISABLED) + self.assertIsNone(result.state) + + def test_reservation_precedes_prepare_and_state_contains_no_secrets(self): + """The target sees durable RESERVED before any secret-bearing prepare.""" + secret = "do-not-persist-secret-value" + store = InMemoryAtomicStateStore() + target = InMemoryWholeConfigTarget({"manual_setting": "keep", "api_secret": secret}) + + def assert_reserved(fence, projection_digest): + state = store.state + self.assertEqual(state.phase, MaterializationPhase.RESERVED) + self.assertEqual(state.reservation_key.fence, fence) + self.assertEqual( + state.projection_digest, + projection_digest, + ) + + target.reservation_probe = assert_reserved + materializer, _store, _target = enabled_materializer( + store=store, + target=target, + ) + request = envelope() + + result = materializer.apply(request) + + self.assertEqual(result.status, MaterializationStatus.APPLIED) + self.assertEqual( + [state.phase for state in store.history], + [ + MaterializationPhase.RESERVED, + MaterializationPhase.PREPARED, + MaterializationPhase.APPLYING, + MaterializationPhase.APPLIED, + ], + ) + self.assertEqual(result.state.operation.phase, TargetOperationPhase.SEALED) + self.assertEqual( + result.state.operation.terminal_phase, + TargetTerminalPhase.APPLIED, + ) + durable_text = repr(result.state) + repr(result) + repr(request) + self.assertNotIn(secret, durable_text) + self.assertNotIn(request.request.feedback_token, durable_text) + for name in ( + "projection", + "preimage", + "candidate", + "feedback_token", + "provenance", + ): + self.assertFalse(hasattr(result.state, name)) + + def test_semantic_forgery_and_cursor_regression_fail_closed(self): + """Initial digest forgery and later publication regression cannot apply.""" + plan = config_plan(1) + forged_argument = replace(plan.config_arguments[0], value=99) + forged = replace( + plan, + config_arguments=(forged_argument,), + projected_config={"battery_rate_max": 99}, + ) + materializer, store, target = enabled_materializer() + + with self.assertRaises(MaterializationValidationError): + materializer.apply(envelope(forged, 1, "forged")) + self.assertEqual(store.writes, 0) + self.assertEqual(target.prepare_calls, 0) + + materializer.apply(envelope(plan, 1, "one")) + replay = materializer.apply(envelope(plan, 100, "hundred")) + self.assertEqual(replay.status, MaterializationStatus.UNCHANGED) + self.assertEqual(replay.state.publication_version, 100) + regressed = materializer.apply(envelope(config_plan(2), 2, "two")) + self.assertEqual(regressed.status, MaterializationStatus.SUPERSEDED) + self.assertEqual(target.config["battery_rate_max"], 1) + + def test_dual_reserve_cas_loser_never_prepares_target(self): + """Only the durable CAS winner may create a target transaction.""" + store = InMemoryAtomicStateStore() + store.block_cas_call = 1 + store.cas_entered = threading.Event() + store.cas_release = threading.Event() + target = InMemoryWholeConfigTarget() + target.block_prepare = True + target.prepare_entered = threading.Event() + target.prepare_release = threading.Event() + first, _store, _target = enabled_materializer( + store=store, + target=target, + ) + second, _store, _target = enabled_materializer( + store=store, + target=target, + ) + request = envelope() + results = {} + + first_thread = threading.Thread(target=lambda: results.setdefault("first", first.apply(request))) + second_thread = threading.Thread(target=lambda: results.setdefault("second", second.apply(request))) + first_thread.start() + self.assertTrue(store.cas_entered.wait(5)) + second_thread.start() + self.assertTrue(target.prepare_entered.wait(5)) + store.cas_release.set() + first_thread.join(5) + + self.assertEqual( + results["first"].status, + MaterializationStatus.UNKNOWN, + ) + self.assertEqual(target.prepare_calls, 0) + self.assertEqual(len(target.records), 0) + target.prepare_release.set() + second_thread.join(5) + self.assertEqual( + results["second"].status, + MaterializationStatus.APPLIED, + ) + self.assertEqual(target.prepare_calls, 1) + self.assertEqual(len(target.records), 1) + + def test_restart_overlap_invalidates_late_apply(self): + """Recovery fences APPLYING before a blocked old commit can resume.""" + store = InMemoryAtomicStateStore() + target = InMemoryWholeConfigTarget() + target.block_apply = True + target.apply_entered = threading.Event() + target.apply_release = threading.Event() + first, _store, _target = enabled_materializer( + store=store, + target=target, + ) + restarted, _store, _target = enabled_materializer( + store=store, + target=target, + ) + results = {} + thread = threading.Thread( + target=lambda: results.setdefault( + "apply", + first.apply(envelope()), + ) + ) + thread.start() + self.assertTrue(target.apply_entered.wait(5)) + + recovered = restarted.recover() + target.apply_release.set() + thread.join(5) + + self.assertEqual(recovered.status, MaterializationStatus.NOT_APPLIED) + self.assertEqual(results["apply"].status, MaterializationStatus.NOT_APPLIED) + self.assertEqual(target.writes, 0) + self.assertEqual(store.state.phase, MaterializationPhase.ROLLED_BACK) + + def test_restart_overlap_invalidates_late_rollback(self): + """Recovery seals visibility before a blocked rollback can resume.""" + store = InMemoryAtomicStateStore() + target = InMemoryWholeConfigTarget() + materializer, _store, _target = enabled_materializer( + store=store, + target=target, + ) + materializer.apply(envelope()) + target.block_rollback = True + target.rollback_entered = threading.Event() + target.rollback_release = threading.Event() + results = {} + thread = threading.Thread( + target=lambda: results.setdefault( + "rollback", + materializer.rollback(), + ) + ) + thread.start() + self.assertTrue(target.rollback_entered.wait(5)) + restarted, _store, _target = enabled_materializer( + store=store, + target=target, + ) + + recovered = restarted.recover() + target.rollback_release.set() + thread.join(5) + + self.assertEqual(recovered.status, MaterializationStatus.APPLIED) + self.assertEqual(results["rollback"].status, MaterializationStatus.APPLIED) + self.assertEqual(target.config["battery_rate_max"], 7) + self.assertEqual(store.state.phase, MaterializationPhase.APPLIED) + + def test_sealed_and_rolled_back_handles_reject_apply_replay(self): + """A captured old PREPARED handle can never write after sealing.""" + materializer, _store, target = enabled_materializer() + applied = materializer.apply(envelope()) + record = target.records[target._tuple_key(applied.state.reservation_key)] + captured = TargetOperationSnapshot( + key=record["key"], + phase=TargetOperationPhase.PREPARED, + handle=record["handle"], + preimage_digest=_config_digest(record["preimage"]), + candidate_digest=_config_digest(record["candidate"]), + changed=True, + ) + writes = target.writes + + replay_after_apply = target.apply(captured) + materializer.rollback() + replay_after_rollback = target.apply(captured) + + self.assertEqual( + replay_after_apply.phase, + TargetOperationPhase.SEALED, + ) + self.assertEqual( + replay_after_rollback.terminal_phase, + TargetTerminalPhase.ROLLED_BACK, + ) + self.assertEqual(target.writes, writes + 1) + self.assertNotIn("battery_rate_max", target.config) + + def test_higher_reserved_publication_cannot_be_cancelled_by_lower(self): + """A blocked v2 reservation makes a later v1 immediately stale.""" + store = InMemoryAtomicStateStore() + target = InMemoryWholeConfigTarget() + target.block_prepare = True + target.prepare_entered = threading.Event() + target.prepare_release = threading.Event() + higher, _store, _target = enabled_materializer( + store=store, + target=target, + ) + lower, _store, _target = enabled_materializer( + store=store, + target=target, + ) + results = {} + thread = threading.Thread( + target=lambda: results.setdefault( + "higher", + higher.apply(envelope(config_plan(2), 2, "two")), + ) + ) + thread.start() + self.assertTrue(target.prepare_entered.wait(5)) + + stale = lower.apply(envelope(config_plan(1), 1, "one")) + + self.assertEqual(stale.status, MaterializationStatus.SUPERSEDED) + self.assertEqual(store.state.publication_version, 2) + self.assertEqual(target.invalidate_calls, 0) + target.prepare_release.set() + thread.join(5) + self.assertEqual(results["higher"].status, MaterializationStatus.APPLIED) + self.assertEqual(target.config["battery_rate_max"], 2) + + def test_topology_is_recursively_bounded_before_any_seam(self): + """Infinite/faulting mapping proxies fail without len or full iteration.""" + plan = config_plan() + cases = ( + MappingProxyType(InfiniteMapping()), + MappingProxyType(FaultingMapping()), + { + "nested": MappingProxyType(FaultingMapping()), + }, + ) + for topology in cases: + with self.subTest(topology=type(topology).__name__): + materializer, store, target = enabled_materializer() + malformed = replace(plan, topology=topology) + with self.assertRaises(MaterializationValidationError) as caught: + materializer.apply(envelope(malformed)) + self.assertNotIn("secret-topology", str(caught.exception)) + self.assertIsNone(caught.exception.__context__) + self.assertEqual(store.writes, 0) + self.assertEqual(target.prepare_calls, 0) + + def test_all_numeric_sequences_are_bounded_before_external_calls(self): + """Huge config/generation/revision/version values fail before seams.""" + huge = MAX_SEQUENCE + 1 + plan = config_plan() + argument = replace(plan.config_arguments[0], value=huge) + huge_config = replace( + plan, + config_arguments=(argument,), + projected_config={"battery_rate_max": huge}, + ) + materializer, store, target = enabled_materializer() + with self.assertRaises(MaterializationValidationError): + materializer.apply(envelope(huge_config)) + self.assertEqual(store.writes, 0) + self.assertEqual(target.prepare_calls, 0) + + huge_generation = replace( + plan, + provider_generations=(("gateway", huge),), + ) + with self.assertRaises(MaterializationValidationError): + materializer.apply(envelope(huge_generation)) + self.assertEqual(store.writes, 0) + + provenance = replace(plan.provenance[0], generation=huge) + huge_provenance = replace(plan, provenance=(provenance,)) + with self.assertRaises(MaterializationValidationError): + materializer.apply(envelope(huge_provenance)) + self.assertEqual(store.writes, 0) + + with self.assertRaises(ValueError): + envelope(plan, huge, "huge-version") + + fence_store = InMemoryAtomicStateStore() + fence_store.fence_override = huge + fenced, _store, fence_target = enabled_materializer(store=fence_store) + with self.assertRaises(MaterializationStorageError): + fenced.apply(envelope()) + self.assertEqual(fence_target.prepare_calls, 0) + + reserved = reserve_only(materializer, envelope(plan, 1, "one")) + object.__setattr__(reserved, "revision", huge) + store.state = reserved + with self.assertRaises(MaterializationStorageError): + materializer.apply(envelope(config_plan(2), 2, "two")) + self.assertEqual(target.prepare_calls, 0) + + def test_crash_cuts_reservation_prepare_attach_and_applying(self): + """Every pre-commit crash cut is recoverable without raw config.""" + request = envelope() + + materializer, store, target = enabled_materializer() + reserved = reserve_only(materializer, request) + recovered = materializer.recover() + self.assertEqual(recovered.status, MaterializationStatus.NOT_APPLIED) + self.assertEqual(target.abort_calls, 1) + projection = _validate_envelope(request)[0] + with self.assertRaises(ValueError): + target.prepare( + reserved.reservation_key.fence, + reserved.projection_digest, + dict(projection), + ) + + materializer, store, target = enabled_materializer() + reserved = reserve_only(materializer, request) + target.prepare( + reserved.reservation_key.fence, + reserved.projection_digest, + dict(projection), + ) + recovered = materializer.recover() + self.assertEqual(recovered.status, MaterializationStatus.NOT_APPLIED) + self.assertEqual(target.invalidate_calls, 1) + + materializer, store, target = enabled_materializer() + reserved = reserve_only(materializer, request) + operation = target.prepare( + reserved.reservation_key.fence, + reserved.projection_digest, + dict(projection), + ) + prepared, terminal = materializer._attach_operation( + reserved, + operation, + ) + self.assertIsNone(terminal) + recovered = materializer.recover() + self.assertEqual(recovered.status, MaterializationStatus.NOT_APPLIED) + + materializer, store, target = enabled_materializer() + reserved = reserve_only(materializer, request) + operation = target.prepare( + reserved.reservation_key.fence, + reserved.projection_digest, + dict(projection), + ) + prepared, _terminal = materializer._attach_operation( + reserved, + operation, + ) + applying, _terminal = materializer._mark_applying(prepared) + self.assertEqual(applying.phase, MaterializationPhase.APPLYING) + recovered = materializer.recover() + self.assertEqual(recovered.status, MaterializationStatus.NOT_APPLIED) + + def test_sealed_target_recovers_across_final_checkpoint_cut(self): + """A durable target seal survives final CAS failure or ack loss.""" + store = InMemoryAtomicStateStore() + store.fail_on_calls[4] = "raise_before" + materializer, _store, target = enabled_materializer(store=store) + + with self.assertRaises(MaterializationStorageError): + materializer.apply(envelope()) + + self.assertEqual(store.state.phase, MaterializationPhase.APPLYING) + snapshot = target.lookup( + store.state.reservation_key.fence, + store.state.projection_digest, + ) + self.assertEqual(snapshot.phase, TargetOperationPhase.SEALED) + self.assertEqual( + snapshot.terminal_phase, + TargetTerminalPhase.APPLIED, + ) + self.assertEqual( + materializer.recover().status, + MaterializationStatus.APPLIED, + ) + + store = InMemoryAtomicStateStore() + store.fail_on_calls[4] = "raise_after" + materializer, _store, target = enabled_materializer(store=store) + applied = materializer.apply(envelope()) + self.assertEqual(applied.status, MaterializationStatus.APPLIED) + self.assertEqual(store.state.phase, MaterializationPhase.APPLIED) + self.assertEqual(target.writes, 1) + + def test_partial_target_apply_is_sealed_unknown(self): + """A partial target mutation is never guessed applied or rolled back.""" + target = InMemoryWholeConfigTarget() + target.partial_apply = {"partially": "written"} + materializer, store, _target = enabled_materializer(target=target) + + result = materializer.apply(envelope()) + + self.assertEqual(result.status, MaterializationStatus.UNKNOWN) + self.assertEqual(store.state.phase, MaterializationPhase.UNKNOWN) + self.assertEqual( + store.state.operation.phase, + TargetOperationPhase.SEALED, + ) + self.assertEqual( + store.state.operation.terminal_phase, + TargetTerminalPhase.UNKNOWN, + ) + self.assertEqual(target.config, {"partially": "written"}) + + def test_seal_abort_invalidate_failures_remain_recoverable(self): + """Failed lifecycle acknowledgements never become terminal guesses.""" + materializer, store, target = enabled_materializer() + target.raise_seal = True + unknown = materializer.apply(envelope()) + self.assertEqual(unknown.status, MaterializationStatus.UNKNOWN) + self.assertEqual(store.state.phase, MaterializationPhase.APPLYING) + target.raise_seal = False + recovered = materializer.recover() + self.assertEqual(recovered.status, MaterializationStatus.APPLIED) + + materializer, store, target = enabled_materializer() + reserve_only(materializer, envelope()) + target.raise_abort = True + unknown = materializer.recover() + self.assertEqual(unknown.status, MaterializationStatus.UNKNOWN) + self.assertEqual(store.state.phase, MaterializationPhase.RESERVED) + target.raise_abort = False + self.assertEqual( + materializer.recover().status, + MaterializationStatus.NOT_APPLIED, + ) + + materializer, store, target = enabled_materializer() + reserved = reserve_only(materializer, envelope()) + projection = _validate_envelope(envelope())[0] + operation = target.prepare( + reserved.reservation_key.fence, + reserved.projection_digest, + dict(projection), + ) + materializer._attach_operation(reserved, operation) + target.raise_invalidate = True + unknown = materializer.recover() + self.assertEqual(unknown.status, MaterializationStatus.UNKNOWN) + target.raise_invalidate = False + self.assertEqual( + materializer.recover().status, + MaterializationStatus.NOT_APPLIED, + ) + + def test_lookup_corruption_never_finalizes_checkpoint(self): + """Corrupt lookup metadata leaves the durable phase non-terminal.""" + materializer, store, target = enabled_materializer() + reserved = reserve_only(materializer, envelope()) + projection = _validate_envelope(envelope())[0] + operation = target.prepare( + reserved.reservation_key.fence, + reserved.projection_digest, + dict(projection), + ) + prepared, _terminal = materializer._attach_operation( + reserved, + operation, + ) + cases = ( + object(), + replace(operation, handle="different-handle"), + ) + for corrupt in cases: + with self.subTest(corrupt=type(corrupt).__name__): + target.lookup_override = corrupt + recovered = materializer.recover() + self.assertEqual( + recovered.status, + MaterializationStatus.UNKNOWN, + ) + self.assertEqual(store.state, prepared) + + def test_recovery_covers_every_target_operation_phase(self): + """No target phase is finalized before an acknowledged SEALED result.""" + phases = ( + TargetOperationPhase.PREPARED, + TargetOperationPhase.APPLYING, + TargetOperationPhase.APPLIED, + TargetOperationPhase.ROLLING_BACK, + TargetOperationPhase.ROLLED_BACK, + TargetOperationPhase.SEALED, + TargetOperationPhase.UNKNOWN, + ) + for phase in phases: + with self.subTest(phase=phase.value): + materializer, store, target = enabled_materializer() + reserved = reserve_only(materializer, envelope()) + projection = _validate_envelope(envelope())[0] + operation = target.prepare( + reserved.reservation_key.fence, + reserved.projection_digest, + dict(projection), + ) + record = target.records[target._tuple_key(reserved.reservation_key)] + record["phase"] = phase + if phase is TargetOperationPhase.APPLIED: + target.config = copy.deepcopy(dict(record["candidate"])) + elif phase is TargetOperationPhase.ROLLED_BACK: + target.config = copy.deepcopy(dict(record["preimage"])) + elif phase is TargetOperationPhase.SEALED: + target.config = copy.deepcopy(dict(record["candidate"])) + record["terminal"] = TargetTerminalPhase.APPLIED + unknown_state = reserved.with_operation( + MaterializationPhase.UNKNOWN, + operation, + ) + store.state = unknown_state + + recovered = materializer.recover() + + self.assertIn( + recovered.status, + ( + MaterializationStatus.APPLIED, + MaterializationStatus.NOT_APPLIED, + MaterializationStatus.UNKNOWN, + ), + ) + self.assertEqual( + target.lookup( + reserved.reservation_key.fence, + reserved.projection_digest, + ).phase, + TargetOperationPhase.SEALED, + ) + + def test_target_and_store_errors_are_value_free(self): + """No secret-bearing seam exception survives in message or context.""" + store = InMemoryAtomicStateStore() + store.raise_load = True + materializer, _store, target = enabled_materializer(store=store) + with self.assertRaises(MaterializationStorageError) as caught: + materializer.apply(envelope()) + self.assertNotIn("secret-store", str(caught.exception)) + self.assertIsNone(caught.exception.__context__) + self.assertEqual(target.prepare_calls, 0) + + store = InMemoryAtomicStateStore() + store.raise_fence = True + materializer, _store, target = enabled_materializer(store=store) + with self.assertRaises(MaterializationStorageError): + materializer.apply(envelope()) + self.assertEqual(target.prepare_calls, 0) + + target = InMemoryWholeConfigTarget() + target.raise_prepare = True + materializer, store, _target = enabled_materializer(target=target) + result = materializer.apply(envelope()) + self.assertEqual(result.status, MaterializationStatus.UNKNOWN) + self.assertEqual(store.state.phase, MaterializationPhase.RESERVED) + + for mode in ("reject", "raise_before", "raise_after"): + with self.subTest(mode=mode): + store = InMemoryAtomicStateStore() + store.fail_on_calls[1] = mode + materializer, _store, target = enabled_materializer(store=store) + if mode == "raise_after": + result = materializer.apply(envelope()) + self.assertEqual( + result.status, + MaterializationStatus.APPLIED, + ) + else: + with self.assertRaises(MaterializationStorageError): + materializer.apply(envelope()) + if mode != "raise_after": + self.assertEqual(target.prepare_calls, 0) + + def test_apply_and_rollback_exceptions_recover_by_lookup_and_seal(self): + """Before/after target exceptions resolve only through sealed lookup.""" + for attribute, expected in ( + ("raise_apply_before", MaterializationStatus.NOT_APPLIED), + ("raise_apply_after", MaterializationStatus.APPLIED), + ): + with self.subTest(attribute=attribute): + target = InMemoryWholeConfigTarget() + setattr(target, attribute, True) + materializer, _store, _target = enabled_materializer(target=target) + result = materializer.apply(envelope()) + self.assertEqual(result.status, expected) + + for attribute, expected in ( + ("raise_rollback_before", MaterializationStatus.APPLIED), + ("raise_rollback_after", MaterializationStatus.ROLLED_BACK), + ): + with self.subTest(attribute=attribute): + target = InMemoryWholeConfigTarget() + materializer, _store, _target = enabled_materializer(target=target) + materializer.apply(envelope()) + setattr(target, attribute, True) + result = materializer.rollback() + self.assertEqual(result.status, expected) + + def test_canonical_values_are_type_exact_and_non_finite_rejected(self): + """None/list/scalars survive while numeric ambiguities are rejected.""" + + class HostileInt(int): + def __lt__(self, _other): + raise RuntimeError("secret-hostile-integer") + + def __gt__(self, _other): + raise RuntimeError("secret-hostile-integer") + + class HostileFloat(float): + def __float__(self): + raise RuntimeError("secret-hostile-float") + + class HostileStr(str): + def strip(self, _chars=None): + raise RuntimeError("secret-hostile-string") + + def encode(self, _encoding="utf-8", _errors="strict"): + raise RuntimeError("secret-hostile-string") + + materializer, _store, target = enabled_materializer() + result = materializer.apply(envelope(values_plan())) + self.assertEqual(result.status, MaterializationStatus.APPLIED) + self.assertEqual( + target.config, + { + "battery_power": (5, None), + "givtcp_rest": None, + "inverter_type": "GEC", + "set_read_only": True, + }, + ) + self.assertFalse(_configs_equal({"value": True}, {"value": 1})) + self.assertFalse(_configs_equal({"value": 1}, {"value": 1.0})) + for value in (math.nan, math.inf, -math.inf): + with self.assertRaises(ValueError): + _canonical_config({"value": value}, "test") + for value in (HostileInt(1), HostileFloat(1.0), HostileStr("x")): + with self.subTest(value_type=type(value).__name__): + with self.assertRaises(ValueError) as caught: + _canonical_config({"value": value}, "test") + self.assertNotIn("secret-hostile", str(caught.exception)) + self.assertIsNone(caught.exception.__context__) + with self.assertRaises(ValueError): + _canonical_config({HostileStr("value"): 1}, "test") + + store = InMemoryAtomicStateStore() + store.fence_override = HostileInt(1) + materializer, _store, target = enabled_materializer(store=store) + with self.assertRaises(MaterializationStorageError) as caught: + materializer.apply(envelope()) + self.assertNotIn("secret-hostile", str(caught.exception)) + self.assertEqual(target.prepare_calls, 0) + + def test_corrupt_state_subtype_and_fingerprint_fail_closed(self): + """Exact base type and integrity reject corrupt durable checkpoints.""" + materializer, store, _target = enabled_materializer() + valid = materializer.apply(envelope()).state + subtype = type( + "BypassState", + (AtomicMaterializationState,), + {"__post_init__": lambda self: None}, + ) + corrupt = object.__new__(subtype) + for name, value in valid.__dict__.items(): + object.__setattr__(corrupt, name, value) + store.corrupt_load = corrupt + with self.assertRaises(MaterializationStorageError) as caught: + materializer.recover() + self.assertIsNone(caught.exception.__context__) + + store.corrupt_load = None + torn = object.__new__(AtomicMaterializationState) + for name, value in valid.__dict__.items(): + object.__setattr__(torn, name, value) + object.__setattr__(torn, "integrity_digest", "0" * 64) + store.corrupt_load = torn + with self.assertRaises(MaterializationStorageError) as caught: + materializer.recover() + self.assertIsNone(caught.exception.__context__) + + def test_infinite_argument_and_provenance_iterables_are_not_consumed(self): + """Non-tuple compiler fields fail immediately without unbounded reads.""" + plan = config_plan() + malformed = ( + replace( + plan, + config_arguments=itertools.repeat(plan.config_arguments[0]), + ), + replace( + plan, + provenance=itertools.repeat(plan.provenance[0]), + ), + ) + for candidate in malformed: + materializer, store, target = enabled_materializer() + with self.assertRaises(MaterializationValidationError): + materializer.apply(envelope(candidate)) + self.assertEqual(store.writes, 0) + self.assertEqual(target.prepare_calls, 0) + + def test_readiness_and_publication_identity_are_exact(self): + """Only the sole blocker and exact same-version audit may proceed.""" + plan = config_plan() + ready = replace( + plan, + materialization_readiness=MaterializationReadiness(True, ()), + ) + materializer, _store, target = enabled_materializer() + with self.assertRaises(MaterializationValidationError): + materializer.apply(envelope(ready)) + + materializer.apply(envelope(plan, 2, "two")) + with self.assertRaises(MaterializationValidationError): + materializer.apply(envelope(plan, 2, "different-token")) + with self.assertRaises(MaterializationValidationError): + materializer.apply(envelope(config_plan(2), 2, "conflict")) + self.assertEqual(target.config["battery_rate_max"], 7) + + +if __name__ == "__main__": + unittest.main()