diff --git a/apps/predbat/lattice_fragment_adapters.py b/apps/predbat/lattice_fragment_adapters.py index e8c4623d1..22f21e1d5 100644 --- a/apps/predbat/lattice_fragment_adapters.py +++ b/apps/predbat/lattice_fragment_adapters.py @@ -20,6 +20,7 @@ import hashlib import threading from dataclasses import dataclass +from functools import partial from types import MappingProxyType from typing import Optional, Protocol @@ -110,6 +111,43 @@ def __post_init__(self): object.__setattr__(self, "provider_id", provider_id) +def _compiler_fragment_snapshot(adapter): + """Fresh-read one compiler input, translating only durable removals.""" + state = adapter.read_state() + if not isinstance(state, FragmentAdapterState): + raise FragmentAdapterReadError("adapter read_state must return FragmentAdapterState") + try: + state.__post_init__() + except ValueError as exc: + raise FragmentAdapterReadError(str(exc)) from exc + provider_id = _validate_provider_id(getattr(adapter, "provider_id", None)) + if state.provider_id != provider_id: + raise FragmentAdapterReadError("adapter state belongs to provider {}".format(state.provider_id)) + if not state.removed: + return state.snapshot + return ProviderSnapshot( + provider_id=state.provider_id, + generation=state.generation, + health=ProviderHealth.HEALTHY, + topology_fragment={ + "topologyVersion": "0.3.0", + "scope": "fragment", + "docVersion": state.generation, + "producer": { + "name": "PredBat fragment tombstone", + "provider": state.provider_id, + "authority": 0, + }, + "nodes": [], + "relationships": [], + }, + aliases=(), + identity_aliases=(), + role_assignments=(), + config_projections=(), + ) + + class FragmentAdapterStateStore: """Required atomic durable-store protocol owned by one integration.""" @@ -484,7 +522,13 @@ def create_compiler(self, state_store, override_reader=None): raise RuntimeError("cannot create a compiler without fragment adapters") for adapter in self._adapters.values(): self._validate_adapter(adapter) - readers = {provider_id: adapter.read_snapshot for provider_id, adapter in self._adapters.items()} + readers = { + provider_id: partial( + _compiler_fragment_snapshot, + adapter, + ) + for provider_id, adapter in self._adapters.items() + } compiler = CompiledLatticeCompiler( readers, state_store=state_store, diff --git a/apps/predbat/tests/test_lattice_fragment_adapters.py b/apps/predbat/tests/test_lattice_fragment_adapters.py index b0de1b4f0..adf075213 100644 --- a/apps/predbat/tests/test_lattice_fragment_adapters.py +++ b/apps/predbat/tests/test_lattice_fragment_adapters.py @@ -401,8 +401,8 @@ def test_other_provider_invalidation_exposes_reader_failure_fail_closed(self): {issue.code for issue in run.issues}, ) - def test_removal_invalidates_and_preserves_last_known_good(self): - """A durable removal triggers a bounded fail-closed recompile.""" + def test_removal_invalidates_and_clears_prior_provider_contribution(self): + """A durable removal compiles as an acknowledged empty fragment.""" adapter, _store, _initial = publisher("gateway") _registry, compiler, _compiled_store, _ids = compiled_registry(adapter) first = compiler.drain() @@ -411,13 +411,16 @@ def test_removal_invalidates_and_preserves_last_known_good(self): run = compiler.drain() self.assertEqual(run.attempts, 1) - self.assertEqual(run.status, CompileStatus.STALE) - self.assertTrue(run.pending) - self.assertIs(run.publication, first.publication) - self.assertIn( - "provider_read_failed", - {issue.code for issue in run.issues}, + self.assertEqual(run.status, CompileStatus.FRESH) + self.assertFalse(run.pending) + self.assertTrue(run.published) + self.assertEqual(run.publication.lattice_version, 2) + self.assertEqual( + dict(run.publication.provider_generations), + {"gateway": 2}, ) + self.assertEqual(run.plan.topology["nodes"], ()) + self.assertNotEqual(run.publication, first.publication) def test_restart_restores_adapter_and_compiler_cursor_protection(self): """Both durable layers reject reuse after a complete process restart.""" @@ -482,7 +485,8 @@ def test_invalidation_during_read_gets_one_bounded_follow_up(self): adapter, _store, initial = publisher("gateway") registry = FragmentAdapterRegistry(enabled=True) registry.register(adapter) - original_reader = adapter.read_snapshot + compiler = registry.create_compiler(InMemoryCompiledLatticeStateStore()) + original_reader = adapter.read_state fired = [False] def invalidating_reader(): @@ -496,8 +500,7 @@ def invalidating_reader(): ) return value - adapter.read_snapshot = invalidating_reader - compiler = registry.create_compiler(InMemoryCompiledLatticeStateStore()) + adapter.read_state = invalidating_reader run = compiler.drain() self.assertEqual(run.attempts, 2) diff --git a/apps/predbat/tests/test_lattice_fragment_tombstone_compilation.py b/apps/predbat/tests/test_lattice_fragment_tombstone_compilation.py new file mode 100644 index 000000000..cb251b76c --- /dev/null +++ b/apps/predbat/tests/test_lattice_fragment_tombstone_compilation.py @@ -0,0 +1,258 @@ +"""Focused coverage for compiler-only fragment tombstone translation.""" + +# cspell:ignore autoconfig + +import os +import sys +import unittest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from lattice_autoconfig import CompileStatus, ProviderHealth # noqa: E402 +from lattice_compiled_publication import ( # noqa: E402 + InMemoryCompiledLatticeStateStore, +) +from lattice_fragment_adapters import ( # noqa: E402 + DurableFragmentAdapter, + FragmentAdapterState, + FragmentAdapterRegistry, + FragmentAdapterRemoved, + InMemoryFragmentAdapterStateStore, + _compiler_fragment_snapshot, +) +from tests.test_lattice_autoconfig import snapshot # noqa: E402 + + +def publisher(provider_id, node_id): + """Create one seeded integration-owned fragment publisher.""" + store = InMemoryFragmentAdapterStateStore() + adapter = DurableFragmentAdapter(provider_id, store) + initial = snapshot( + provider_id, + generation=1, + node_id=node_id, + ) + adapter.publish(initial, "initial fragment") + return adapter, store, initial + + +def compiler_for(store, *adapters): + """Create one explicitly enabled compiler over fixed membership.""" + registry = FragmentAdapterRegistry(enabled=True) + for adapter in adapters: + registry.register(adapter) + return registry, registry.create_compiler(store) + + +def structurally_corrupt_state( + provider_id, + generation, + semantic_fingerprint, + provider_snapshot, + removed=False, +): + """Bypass construction validation to model a corrupt structural reader.""" + state = object.__new__(FragmentAdapterState) + object.__setattr__(state, "provider_id", provider_id) + object.__setattr__(state, "generation", generation) + object.__setattr__( + state, + "semantic_fingerprint", + semantic_fingerprint, + ) + object.__setattr__(state, "snapshot", provider_snapshot) + object.__setattr__(state, "removed", removed) + return state + + +class TestCompilerFragmentTombstones(unittest.TestCase): + """Removal is empty compiler input without weakening adapter reads.""" + + def test_live_snapshot_is_exact_and_removed_snapshot_is_empty(self): + """Translation changes only a durable removal state.""" + adapter, _store, initial = publisher("gateway", "GW-INV") + + self.assertIs(_compiler_fragment_snapshot(adapter), initial) + self.assertTrue(adapter.remove(2, "integration removed")) + + with self.assertRaises(FragmentAdapterRemoved): + adapter.read_snapshot() + tombstone = _compiler_fragment_snapshot(adapter) + + self.assertEqual(tombstone.provider_id, "gateway") + self.assertEqual(tombstone.generation, 2) + self.assertEqual(tombstone.health, ProviderHealth.HEALTHY) + self.assertEqual(tombstone.topology_fragment["nodes"], ()) + self.assertEqual(tombstone.topology_fragment["relationships"], ()) + self.assertEqual(tombstone.aliases, ()) + self.assertEqual(tombstone.identity_aliases, ()) + self.assertEqual(tombstone.role_assignments, ()) + self.assertEqual(tombstone.config_projections, ()) + + def test_all_removed_publishes_deterministic_empty_plan_and_restarts(self): + """All tombstones settle, persist, and restore as one empty plan.""" + alpha, alpha_store, _alpha = publisher("alpha", "ALPHA-INV") + beta, beta_store, _beta = publisher("beta", "BETA-INV") + compiled_store = InMemoryCompiledLatticeStateStore() + _registry, compiler = compiler_for(compiled_store, alpha, beta) + baseline = compiler.drain() + + self.assertTrue(beta.remove(2, "beta removed")) + self.assertTrue(alpha.remove(2, "alpha removed")) + removed = compiler.drain() + + self.assertEqual(removed.status, CompileStatus.FRESH) + self.assertTrue(removed.published) + self.assertFalse(removed.pending) + self.assertEqual(removed.plan.topology["nodes"], ()) + self.assertEqual(removed.plan.aliases, ()) + self.assertEqual( + dict(removed.publication.provider_generations), + {"alpha": 2, "beta": 2}, + ) + self.assertEqual( + dict(removed.publication.provider_requested_generations), + {"alpha": 2, "beta": 2}, + ) + self.assertNotEqual( + removed.publication.digest, + baseline.publication.digest, + ) + + restarted_alpha = DurableFragmentAdapter("alpha", alpha_store) + restarted_beta = DurableFragmentAdapter("beta", beta_store) + _registry, restarted = compiler_for( + compiled_store, + restarted_beta, + restarted_alpha, + ) + settled = restarted.drain() + + self.assertFalse(settled.published) + self.assertFalse(settled.pending) + self.assertEqual(settled.publication, removed.publication) + self.assertEqual(settled.plan.digest, removed.plan.digest) + + def test_removed_payload_cannot_reappear_at_the_same_generation(self): + """A tombstone never leaks its payload and cursor reuse remains unsafe.""" + adapter, _store, initial = publisher("cloud", "CLOUD-INV") + self.assertTrue(adapter.remove(2, "cloud removed")) + + empty = _compiler_fragment_snapshot(adapter) + self.assertEqual(empty.topology_fragment["nodes"], ()) + with self.assertRaisesRegex(ValueError, "reused"): + adapter.publish( + snapshot( + "cloud", + generation=2, + node_id=initial.topology_fragment["nodes"][0]["id"], + ), + "attempted resurrection", + ) + self.assertEqual( + _compiler_fragment_snapshot(adapter).topology_fragment["nodes"], + (), + ) + + def test_fresh_read_revalidates_semantic_fingerprint_and_keeps_lkg(self): + """Post-registration state corruption cannot reach publication.""" + adapter, _store, _initial = publisher("gateway", "GW-INV") + compiled_store = InMemoryCompiledLatticeStateStore() + _registry, compiler = compiler_for(compiled_store, adapter) + baseline = compiler.drain() + corrupt = structurally_corrupt_state( + "gateway", + 2, + "0" * 64, + snapshot( + "gateway", + generation=2, + node_id="GW-INV", + ), + ) + adapter.read_state = lambda: corrupt + + self.assertTrue( + compiler.invalidate( + "gateway", + 2, + "corrupt structural read", + ) + ) + failed = compiler.drain() + + self.assertEqual(failed.status, CompileStatus.STALE) + self.assertTrue(failed.pending) + self.assertFalse(failed.published) + self.assertIs(failed.publication, baseline.publication) + self.assertEqual(compiled_store.writes, 1) + self.assertIn( + "provider_read_failed", + {issue.code for issue in failed.issues}, + ) + + def test_fresh_read_revalidates_snapshot_cursor_bindings(self): + """Provider and generation binding corruption both fail closed.""" + cases = ( + ( + "provider", + snapshot( + "other-provider", + generation=2, + node_id="GW-INV", + ), + ), + ( + "generation", + snapshot( + "gateway", + generation=3, + node_id="GW-INV", + ), + ), + ) + for label, corrupt_snapshot in cases: + with self.subTest(binding=label): + adapter, _store, _initial = publisher( + "gateway", + "GW-INV", + ) + compiled_store = InMemoryCompiledLatticeStateStore() + _registry, compiler = compiler_for( + compiled_store, + adapter, + ) + baseline = compiler.drain() + corrupt = structurally_corrupt_state( + "gateway", + 2, + adapter.read_state().semantic_fingerprint, + corrupt_snapshot, + ) + adapter.read_state = lambda value=corrupt: value + + self.assertTrue( + compiler.invalidate( + "gateway", + 2, + "{} binding corruption".format(label), + ) + ) + failed = compiler.drain() + + self.assertEqual(failed.status, CompileStatus.STALE) + self.assertTrue(failed.pending) + self.assertFalse(failed.published) + self.assertIs( + failed.publication, + baseline.publication, + ) + self.assertEqual(compiled_store.writes, 1) + self.assertIn( + "provider_read_failed", + {issue.code for issue in failed.issues}, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/apps/predbat/tests/test_lattice_solis_fragment.py b/apps/predbat/tests/test_lattice_solis_fragment.py index 75ed72c2a..802dfcf77 100644 --- a/apps/predbat/tests/test_lattice_solis_fragment.py +++ b/apps/predbat/tests/test_lattice_solis_fragment.py @@ -398,7 +398,13 @@ def test_all_changes_invalidate_common_compiler_with_fresh_reads(self): self.assertTrue(offline.pending) self.assertTrue(adapter.remove()) removed = compiler.drain() - self.assertEqual(removed.status, CompileStatus.STALE) + self.assertEqual(removed.status, CompileStatus.FRESH) + self.assertFalse(removed.pending) + self.assertEqual(removed.plan.topology["nodes"], ()) + self.assertEqual( + dict(removed.publication.provider_generations), + {"solis": 5}, + ) self.assertEqual( tuple(item[:3] for item in invalidations),