diff --git a/dns_sync/reconciliation.py b/dns_sync/reconciliation.py new file mode 100644 index 0000000..42c2973 --- /dev/null +++ b/dns_sync/reconciliation.py @@ -0,0 +1,136 @@ +from collections import defaultdict +from collections.abc import Iterable, Mapping, Sequence +from typing import Protocol + +from .adapters import ( + DNS_RECORDS, + NormalizedObject, + Observation, +) + + +ObservationGroupKey = tuple[str, str] + + +class ReconciliationConflict(RuntimeError): + """Raised when observations cannot be reconciled safely.""" + + +class ReconciliationPolicy(Protocol): + object_type: str + + def reconcile( + self, + logical_key: str, + observations: Sequence[Observation], + ) -> NormalizedObject: ... + + +def _observation_sort_key( + observation: Observation, +) -> tuple[str, str, str]: + return ( + observation.endpoint_id, + observation.object.local_key or "", + observation.observed_at.isoformat(), + ) + + +def group_observations( + observations: Iterable[Observation], +) -> dict[ObservationGroupKey, tuple[Observation, ...]]: + grouped: dict[ + ObservationGroupKey, + list[Observation], + ] = defaultdict(list) + + for observation in observations: + normalized = observation.object + if not observation.endpoint_id: + raise RuntimeError("Observation endpoint_id cannot be empty") + if not normalized.object_type: + raise RuntimeError("Observation object_type cannot be empty") + if not normalized.logical_key: + raise RuntimeError("Observation logical_key cannot be empty") + + key = ( + normalized.object_type, + normalized.logical_key, + ) + grouped[key].append(observation) + + return { + key: tuple(sorted(rows, key=_observation_sort_key)) + for key, rows in sorted(grouped.items()) + } + + +class DnsUnionPolicy: + object_type = DNS_RECORDS + + def reconcile( + self, + logical_key: str, + observations: Sequence[Observation], + ) -> NormalizedObject: + if not observations: + raise RuntimeError( + "DNS union policy requires at least one observation" + ) + + first = observations[0].object + conflicting_endpoints = { + observation.endpoint_id + for observation in observations + if observation.object.payload != first.payload + } + if conflicting_endpoints: + conflicting_endpoints.add(observations[0].endpoint_id) + endpoint_list = ", ".join( + sorted(conflicting_endpoints) + ) + raise ReconciliationConflict( + f"Conflicting {DNS_RECORDS} observations for " + f"{logical_key!r} from {endpoint_list}" + ) + + return NormalizedObject( + object_type=DNS_RECORDS, + logical_key=logical_key, + payload=dict(first.payload), + ) + + +DEFAULT_POLICIES: Mapping[str, ReconciliationPolicy] = { + DNS_RECORDS: DnsUnionPolicy(), +} + + +def reconcile_observations( + observations: Iterable[Observation], + policies: Mapping[str, ReconciliationPolicy] | None = None, +) -> list[NormalizedObject]: + selected_policies = ( + DEFAULT_POLICIES + if policies is None + else policies + ) + desired: list[NormalizedObject] = [] + + for (object_type, logical_key), rows in ( + group_observations(observations).items() + ): + policy = selected_policies.get(object_type) + if policy is None: + raise RuntimeError( + f"No reconciliation policy for {object_type!r}" + ) + if policy.object_type != object_type: + raise RuntimeError( + f"Policy for {object_type!r} advertises " + f"{policy.object_type!r}" + ) + + desired.append(policy.reconcile(logical_key, rows)) + + return desired diff --git a/tests/test_reconciliation.py b/tests/test_reconciliation.py new file mode 100644 index 0000000..7363775 --- /dev/null +++ b/tests/test_reconciliation.py @@ -0,0 +1,198 @@ +import unittest +from datetime import datetime, timezone + +from dns_sync.adapters import ( + CERTIFICATES, + DNS_RECORDS, + NormalizedObject, + Observation, +) +from dns_sync.reconciliation import ( + ReconciliationConflict, + group_observations, + reconcile_observations, +) + + +NOW = datetime(2026, 9, 19, 22, 0, tzinfo=timezone.utc) + + +def dns_observation( + endpoint_id: str, + hostname: str, + ip: str, + local_key: str, +) -> Observation: + return Observation( + endpoint_id=endpoint_id, + object=NormalizedObject( + object_type=DNS_RECORDS, + logical_key=hostname, + local_key=local_key, + payload={ + "hostname": hostname, + "ip": ip, + "record_type": "A", + "wildcard": False, + }, + ), + observed_at=NOW, + ) + + +class ObservationGroupingTests(unittest.TestCase): + def test_groups_by_type_and_logical_identity(self): + observations = [ + dns_observation( + "npm://192.168.1.4:81", + "app.example.com", + "192.168.1.101", + "22", + ), + dns_observation( + "npm://192.168.1.3:81", + "app.example.com", + "192.168.1.101", + "12", + ), + dns_observation( + "npm://192.168.1.3:81", + "other.example.com", + "192.168.1.101", + "13", + ), + ] + + grouped = group_observations(observations) + + self.assertEqual( + tuple(grouped), + ( + (DNS_RECORDS, "app.example.com"), + (DNS_RECORDS, "other.example.com"), + ), + ) + self.assertEqual( + tuple( + item.endpoint_id + for item in grouped[ + (DNS_RECORDS, "app.example.com") + ] + ), + ( + "npm://192.168.1.3:81", + "npm://192.168.1.4:81", + ), + ) + + +class DnsUnionPolicyTests(unittest.TestCase): + def test_union_keeps_every_unique_hostname(self): + observations = [ + dns_observation( + "npm://192.168.1.3:81", + "one.example.com", + "192.168.1.101", + "12", + ), + dns_observation( + "npm://192.168.1.4:81", + "two.example.com", + "192.168.1.101", + "22", + ), + ] + + desired = reconcile_observations(observations) + + self.assertEqual( + tuple(item.logical_key for item in desired), + ("one.example.com", "two.example.com"), + ) + self.assertTrue( + all(item.local_key is None for item in desired) + ) + + def test_identical_observations_converge_to_one_object(self): + observations = [ + dns_observation( + "npm://192.168.1.3:81", + "app.example.com", + "192.168.1.101", + "12", + ), + dns_observation( + "npm://192.168.1.4:81", + "app.example.com", + "192.168.1.101", + "22", + ), + ] + + desired = reconcile_observations(observations) + + self.assertEqual(len(desired), 1) + self.assertEqual( + desired[0].payload["ip"], + "192.168.1.101", + ) + self.assertIsNone(desired[0].local_key) + + def test_conflicting_values_fail_without_exposing_payload(self): + observations = [ + dns_observation( + "npm://node-one:81", + "app.example.com", + "192.168.1.101", + "12", + ), + dns_observation( + "npm://node-two:81", + "app.example.com", + "203.0.113.77", + "22", + ), + ] + + with self.assertRaises(ReconciliationConflict) as raised: + reconcile_observations(observations) + + message = str(raised.exception) + self.assertIn("app.example.com", message) + self.assertIn("npm://node-one:81", message) + self.assertIn("npm://node-two:81", message) + self.assertNotIn("192.168.1.101", message) + self.assertNotIn("203.0.113.77", message) + + def test_missing_policy_fails_explicitly(self): + observation = Observation( + endpoint_id="npm://node-one:81", + object=NormalizedObject( + object_type=CERTIFICATES, + logical_key="app.example.com", + local_key="67", + payload={"expires_on": "2026-12-15"}, + ), + observed_at=NOW, + ) + + with self.assertRaisesRegex( + RuntimeError, + "No reconciliation policy", + ): + reconcile_observations([observation]) + + def test_explicit_empty_policy_registry_uses_no_defaults(self): + observation = dns_observation( + "npm://node-one:81", + "app.example.com", + "192.168.1.101", + "12", + ) + + with self.assertRaisesRegex(RuntimeError, "No reconciliation policy"): + reconcile_observations([observation], policies={}) + + +if __name__ == "__main__": + unittest.main()