Require live observations for convergence

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
+    # Cached observations remain visible, but only a successful
-        for observation in observations
+    # discovery from this exact run may establish desired state or
-        if observation.object.object_type
+    # make an endpoint eligible for mutation.
-        in {
+    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 = [
+    proxy_observations = _authoritative_observations(
-        observation
+        observations,
-        for observation in observations
+        live_pairs,
-        if observation.object.object_type == PROXY_HOSTS
+        frozenset({PROXY_HOSTS}),
-    ]
+    )
 
-    projected_dns = project_proxy_hosts_to_dns(
+    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
+    # produced portable desired state. Router DNS observations are
-    # observations must never participate in the proxy->DNS union.
+    # 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(
+        _planning_adapter_views(
             adapter_list,
+            live_pairs,
+        ),
         desired,
         observations,
     )

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,
+            len(result.plans),
-            PlanOperation.CREATE,
+            1,
+        )
+        self.assertEqual(
+            result.plans[0].actions,
+            (),
         )
         self.assertFalse(status.is_live)
         self.assertEqual(status.last_success_at, T1)

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()