diff --git a/dns_sync/orchestrator.py b/dns_sync/orchestrator.py index e24418c..ff65c2d 100644 --- a/dns_sync/orchestrator.py +++ b/dns_sync/orchestrator.py @@ -52,6 +52,12 @@ class ReconciliationRun: failures: tuple[DiscoveryFailure, ...] +@dataclass(frozen=True) +class _PlanningAdapterView: + endpoint: Endpoint + capabilities: frozenset[Capability] + + def _discovery_types( adapter: ReconciliationAdapter, ) -> tuple[str, ...]: @@ -75,6 +81,90 @@ def _configured_observations( ] +def _live_discovery_pairs( + adapters: Iterable[ReconciliationAdapter], + store: SQLiteStateStore, + attempted_at: datetime, +) -> frozenset[tuple[str, str]]: + live: set[tuple[str, str]] = set() + + for adapter in adapters: + endpoint_id = adapter.endpoint.identity + + for object_type in _discovery_types(adapter): + status = store.observation_status( + endpoint_id, + object_type, + ) + + if ( + status is not None + and status.is_live + and status.last_attempt_at == attempted_at + ): + live.add( + ( + endpoint_id, + object_type, + ) + ) + + return frozenset(live) + + +def _authoritative_observations( + observations: Iterable[Observation], + live_pairs: frozenset[tuple[str, str]], + object_types: frozenset[str], +) -> list[Observation]: + return [ + observation + for observation in observations + if ( + observation.object.object_type + in object_types + and ( + observation.endpoint_id, + observation.object.object_type, + ) + in live_pairs + ) + ] + + +def _planning_adapter_views( + adapters: Iterable[ReconciliationAdapter], + live_pairs: frozenset[tuple[str, str]], +) -> list[_PlanningAdapterView]: + views: list[_PlanningAdapterView] = [] + + for adapter in adapters: + endpoint_id = adapter.endpoint.identity + + capabilities = frozenset( + capability + for capability in adapter.capabilities + if ( + capability.operation + != CapabilityOperation.APPLY + or ( + endpoint_id, + capability.object_type, + ) + in live_pairs + ) + ) + + views.append( + _PlanningAdapterView( + endpoint=adapter.endpoint, + capabilities=capabilities, + ) + ) + + return views + + def _filter_projected_dns( observations: Iterable[Observation], *, @@ -189,15 +279,35 @@ def discover_reconcile_plan( store, set(adapters_by_endpoint), ) - npm_observations = [ - observation - for observation in observations - if observation.object.object_type - in { - CERTIFICATES, - PROXY_HOSTS, - } - ] + + # Cached observations remain visible, but only a successful + # discovery from this exact run may establish desired state or + # make an endpoint eligible for mutation. + live_pairs = _live_discovery_pairs( + adapter_list, + store, + timestamp, + ) + + certificate_authoritative = any( + object_type == CERTIFICATES + for _, object_type in live_pairs + ) + proxy_authoritative = any( + object_type == PROXY_HOSTS + for _, object_type in live_pairs + ) + + npm_observations = _authoritative_observations( + observations, + live_pairs, + frozenset( + { + CERTIFICATES, + PROXY_HOSTS, + } + ), + ) npm_desired = ( reconcile_observations(npm_observations) @@ -205,17 +315,22 @@ def discover_reconcile_plan( else [] ) - proxy_observations = [ - observation - for observation in observations - if observation.object.object_type == PROXY_HOSTS - ] - - projected_dns = project_proxy_hosts_to_dns( - proxy_observations, - target_ip, - include_disabled=include_disabled, + proxy_observations = _authoritative_observations( + observations, + live_pairs, + frozenset({PROXY_HOSTS}), ) + + projected_dns = ( + project_proxy_hosts_to_dns( + proxy_observations, + target_ip, + include_disabled=include_disabled, + ) + if proxy_authoritative + else [] + ) + projected_dns = _filter_projected_dns( projected_dns, domain_matching=domain_matching, @@ -229,18 +344,31 @@ def discover_reconcile_plan( ) # Keep the reconciliation domains separate until each has - # produced portable desired state. In particular, router DNS - # observations must never participate in the proxy->DNS union. + # produced portable desired state. Router DNS observations are + # current target state, never a desired-state authority. desired = [ *npm_desired, *dns_desired, ] + authority_by_type = { + CERTIFICATES: certificate_authoritative, + PROXY_HOSTS: proxy_authoritative, + # DNS desired state is projected from proxy-host authority. + DNS_RECORDS: proxy_authoritative, + } + for object_type in ( CERTIFICATES, PROXY_HOSTS, DNS_RECORDS, ): + if not authority_by_type[object_type]: + # No successful source discovery this run: retain the + # previously selected desired snapshot for diagnostics, + # but do not use it for this run's plans. + continue + store.replace_desired_state( object_type, [ @@ -252,7 +380,10 @@ def discover_reconcile_plan( ) plans = build_endpoint_plans( - adapter_list, + _planning_adapter_views( + adapter_list, + live_pairs, + ), desired, observations, ) diff --git a/tests/test_orchestrator.py b/tests/test_orchestrator.py index 6479356..952cf3f 100644 --- a/tests/test_orchestrator.py +++ b/tests/test_orchestrator.py @@ -235,11 +235,15 @@ class ReadOnlyOrchestrationTests(unittest.TestCase): for item in result.desired if item.object_type == DNS_RECORDS ), - ("cached.example.com",), + (), ) self.assertEqual( - result.plans[0].actions[0].operation, - PlanOperation.CREATE, + len(result.plans), + 1, + ) + self.assertEqual( + result.plans[0].actions, + (), ) self.assertFalse(status.is_live) self.assertEqual(status.last_success_at, T1) diff --git a/tests/test_orchestrator_authority.py b/tests/test_orchestrator_authority.py new file mode 100644 index 0000000..3eecd16 --- /dev/null +++ b/tests/test_orchestrator_authority.py @@ -0,0 +1,322 @@ +import unittest +from dataclasses import dataclass, field +from datetime import datetime, timezone + +from dns_sync.adapters import ( + CERTIFICATES, + DNS_RECORDS, + PROXY_HOSTS, + Capability, + CapabilityOperation, + DiscoveredObject, + Endpoint, + NormalizedObject, + Observation, +) +from dns_sync.orchestrator import discover_reconcile_plan +from dns_sync.persistence import SQLiteStateStore + + +T1 = datetime( + 2026, 9, 20, 8, 0, tzinfo=timezone.utc +) +T2 = datetime( + 2026, 9, 20, 9, 0, tzinfo=timezone.utc +) + + +@dataclass +class FakeAdapter: + endpoint: Endpoint + capabilities: frozenset[Capability] + discovered: dict[str, list[DiscoveredObject]] = field( + default_factory=dict + ) + failures: dict[str, str] = field( + default_factory=dict + ) + + def connect(self) -> None: + return None + + def discover( + self, + object_type: str, + ) -> list[DiscoveredObject]: + if object_type in self.failures: + raise RuntimeError( + self.failures[object_type] + ) + return list( + self.discovered.get( + object_type, + [], + ) + ) + + +def capability_set( + *object_types: str, + apply: bool = False, +) -> frozenset[Capability]: + rows = { + Capability( + object_type, + CapabilityOperation.DISCOVER, + ) + for object_type in object_types + } + + if apply: + rows.update( + Capability( + object_type, + CapabilityOperation.APPLY, + ) + for object_type in object_types + ) + + return frozenset(rows) + + +def normalized_certificate( + local_key: str, + *, + valid_from: str, + expires_on: str, +) -> NormalizedObject: + return NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + local_key=local_key, + payload={ + "domain_names": ["app.example.com"], + "provider": "other", + "valid_from": valid_from, + "expires_on": expires_on, + }, + ) + + +def discovered_certificate( + local_key: str, + *, + valid_from: str, + expires_on: str, +) -> DiscoveredObject: + return DiscoveredObject( + object_type=CERTIFICATES, + local_key=local_key, + payload={ + "domain_names": ["app.example.com"], + "provider": "other", + "valid_from": valid_from, + "expires_on": expires_on, + }, + ) + + +def normalized_proxy( + local_key: str, +) -> NormalizedObject: + return NormalizedObject( + object_type=PROXY_HOSTS, + logical_key="app.example.com", + local_key=local_key, + payload={ + "domain_names": ["app.example.com"], + "forward_scheme": "http", + "forward_host": "backend.internal", + "forward_port": 8080, + "enabled": True, + }, + ) + + +class ObservationAuthorityTests(unittest.TestCase): + def test_failed_newer_certificate_cache_cannot_override_live_certificate(self): + live_endpoint = Endpoint( + "npm", + "192.168.1.3:81", + ) + failed_endpoint = Endpoint( + "npm", + "192.168.1.4:81", + ) + + live = FakeAdapter( + endpoint=live_endpoint, + capabilities=capability_set( + CERTIFICATES, + ), + discovered={ + CERTIFICATES: [ + discovered_certificate( + "20", + valid_from=( + "2026-09-15T00:00:00+00:00" + ), + expires_on=( + "2026-12-15T00:00:00+00:00" + ), + ) + ] + }, + ) + + failed = FakeAdapter( + endpoint=failed_endpoint, + capabilities=capability_set( + CERTIFICATES, + apply=True, + ), + failures={ + CERTIFICATES: "offline" + }, + ) + + cached_newer = normalized_certificate( + "99", + valid_from=( + "2026-09-19T00:00:00+00:00" + ), + expires_on=( + "2026-12-19T00:00:00+00:00" + ), + ) + + with SQLiteStateStore(":memory:") as store: + store.replace_observations( + failed_endpoint.identity, + CERTIFICATES, + [ + Observation( + failed_endpoint.identity, + cached_newer, + T1, + ) + ], + T1, + ) + + result = discover_reconcile_plan( + [ + live, + failed, + ], + store, + target_ip="192.168.1.101", + run_at=T2, + ) + + certificates = [ + object + for object in result.desired + if object.object_type == CERTIFICATES + ] + + self.assertEqual( + len(certificates), + 1, + ) + self.assertEqual( + certificates[0].payload["valid_from"], + "2026-09-15T00:00:00+00:00", + ) + + # The failed target advertised APPLY, but it must not + # receive a certificate mutation plan this run. + self.assertEqual( + result.plans, + (), + ) + + self.assertTrue( + any( + observation.endpoint_id + == failed_endpoint.identity + and observation.object.local_key == "99" + for observation in result.observations + ) + ) + + def test_all_failed_sources_preserve_persisted_desired_without_planning(self): + npm_endpoint = Endpoint( + "npm", + "192.168.1.3:81", + ) + + npm = FakeAdapter( + endpoint=npm_endpoint, + capabilities=capability_set( + PROXY_HOSTS, + apply=True, + ), + failures={ + PROXY_HOSTS: "offline" + }, + ) + + cached_proxy = normalized_proxy("12") + + previous_dns = NormalizedObject( + object_type=DNS_RECORDS, + logical_key="app.example.com", + payload={ + "hostname": "app.example.com", + "ip_address": "192.168.1.101", + }, + ) + + with SQLiteStateStore(":memory:") as store: + store.replace_observations( + npm_endpoint.identity, + PROXY_HOSTS, + [ + Observation( + npm_endpoint.identity, + cached_proxy, + T1, + ) + ], + T1, + ) + + store.replace_desired_state( + DNS_RECORDS, + [previous_dns], + T1, + ) + + result = discover_reconcile_plan( + [npm], + store, + target_ip="192.168.1.101", + run_at=T2, + ) + + persisted_dns = store.load_desired_state( + DNS_RECORDS + ) + + self.assertEqual( + result.desired, + (), + ) + self.assertEqual( + result.plans, + (), + ) + + self.assertEqual( + len(persisted_dns), + 1, + ) + self.assertEqual( + persisted_dns[0].object, + previous_dns, + ) + + +if __name__ == "__main__": + unittest.main()