david/dns-sync
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Resolve endpoint-local object dependencies

Browse Source
This commit is contained in:
Lyncoln
2026-09-19 21:34:23 -04:00
parent dc6505d165
commit 8caac79a1e
8 changed files with 1174 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dns_sync/dependencies.py
+98
View File
@@ -0,0 +1,98 @@
from collections.abc import Iterable
from .adapters import PlanAction
def _action_identity(
action: PlanAction,
) -> tuple[str, str, str]:
desired = action.desired
if desired is None:
raise RuntimeError(
"Dependency-ordered plan action requires desired state"
)
return (
action.endpoint_id,
desired.object_type,
desired.logical_key,
)
def order_plan_actions(
actions: Iterable[PlanAction],
) -> tuple[PlanAction, ...]:
action_list = list(actions)
by_identity: dict[
tuple[str, str, str],
PlanAction,
] = {}
positions: dict[
tuple[str, str, str],
int,
] = {}
for position, action in enumerate(action_list):
identity = _action_identity(action)
if identity in by_identity:
raise RuntimeError(
"Duplicate plan action identity: "
f"{identity[0]} {identity[1]} {identity[2]!r}"
)
by_identity[identity] = action
positions[identity] = position
dependencies: dict[
tuple[str, str, str],
set[tuple[str, str, str]],
] = {
identity: set()
for identity in by_identity
}
for identity, action in by_identity.items():
desired = action.desired
for reference in desired.references:
if reference.logical_key is None:
continue
dependency = (
action.endpoint_id,
reference.object_type,
reference.logical_key,
)
if dependency in by_identity:
dependencies[identity].add(dependency)
remaining = set(by_identity)
ordered: list[PlanAction] = []
while remaining:
ready = [
identity
for identity in remaining
if not dependencies[identity].intersection(remaining)
]
if not ready:
cycle = ", ".join(
f"{identity[1]} {identity[2]!r}"
for identity in sorted(
remaining,
key=lambda item: positions[item],
)
)
raise RuntimeError(
"Plan action dependency cycle: "
f"{cycle}"
)
ready.sort(
key=lambda identity: positions[identity]
)
for identity in ready:
ordered.append(by_identity[identity])
remaining.remove(identity)
return tuple(ordered)
dns_sync/executor.py
+267 -27
View File
@@ -1,35 +1,28 @@
from collections.abc import Iterable
from dataclasses import dataclass
from dataclasses import dataclass, replace
from datetime import datetime, timezone
from typing import Protocol
from .adapters import (
CERTIFICATES,
DNS_RECORDS,
PROXY_HOSTS,
ApplyResult,
Capability,
CapabilityOperation,
Endpoint,
EndpointPlan,
NormalizedObject,
ObjectReference,
PlanAction,
PlanOperation,
)
from .dependencies import order_plan_actions
from .persistence import SQLiteStateStore
_OBJECT_TYPE_ORDER = {
CERTIFICATES: 0,
PROXY_HOSTS: 1,
DNS_RECORDS: 2,
}
class ExecutionAdapter(Protocol):
endpoint: Endpoint
capabilities: frozenset[Capability]
def apply(self, action: PlanAction) -> str | None: ...
def apply(self, action: PlanAction) -> ApplyResult | None: ...
def verify(self, object: NormalizedObject) -> bool: ...
@@ -48,19 +41,238 @@ class ExecutionResult:
skipped: tuple[PlanAction, ...]
def _action_sort_key(
@dataclass
class _LocalReferenceIndex:
by_logical: dict[
tuple[str, str],
set[str],
]
by_local: set[tuple[str, str]]
def _remember_local_object(
index: _LocalReferenceIndex,
object: NormalizedObject,
*,
authoritative: bool = False,
) -> None:
local_key = object.local_key
if local_key is None:
return
identity = (
object.object_type,
object.logical_key,
)
if authoritative:
previous = index.by_logical.get(identity, set())
for previous_local_key in previous:
index.by_local.discard(
(
object.object_type,
previous_local_key,
)
)
index.by_logical[identity] = {local_key}
else:
index.by_logical.setdefault(
identity,
set(),
).add(local_key)
index.by_local.add(
(
object.object_type,
local_key,
)
)
def _local_reference_index(
store: SQLiteStateStore,
endpoint_id: str,
) -> _LocalReferenceIndex:
index = _LocalReferenceIndex(
by_logical={},
by_local=set(),
)
observed_identities: set[
tuple[str, str]
] = set()
for observation in store.load_observations():
if observation.endpoint_id != endpoint_id:
continue
object = observation.object
if object.local_key is None:
continue
identity = (
object.object_type,
object.logical_key,
)
observed_identities.add(identity)
_remember_local_object(index, object)
for applied in store.load_applied_state(
endpoint_id=endpoint_id,
):
if not applied.is_verified:
continue
object = applied.object
if object.local_key is None:
continue
identity = (
object.object_type,
object.logical_key,
)
# Endpoint observations are authoritative when present.
# Verified applied state is only a fallback when this
# logical identity is absent from the observation snapshot.
if identity in observed_identities:
continue
_remember_local_object(index, object)
return index
def _resolved_reference(
endpoint_id: str,
reference: ObjectReference,
local_references: _LocalReferenceIndex,
) -> ObjectReference:
if reference.local_key is not None:
local_identity = (
reference.object_type,
reference.local_key,
)
if local_identity not in local_references.by_local:
raise RuntimeError(
"Unresolved endpoint-local reference for "
f"{endpoint_id}: "
f"{reference.relation} -> "
f"{reference.object_type} "
f"local key {reference.local_key!r}"
)
return reference
identity = (
reference.object_type,
reference.logical_key,
)
candidates = local_references.by_logical.get(
identity,
set(),
)
if not candidates:
raise RuntimeError(
"Unresolved endpoint-local reference for "
f"{endpoint_id}: "
f"{reference.relation} -> "
f"{reference.object_type} "
f"{reference.logical_key!r}"
)
if len(candidates) > 1:
raise RuntimeError(
"Ambiguous endpoint-local reference for "
f"{endpoint_id}: "
f"{reference.relation} -> "
f"{reference.object_type} "
f"{reference.logical_key!r}: "
f"{', '.join(sorted(candidates))}"
)
local_key = next(iter(candidates))
return ObjectReference(
relation=reference.relation,
object_type=reference.object_type,
local_key=local_key,
)
def _resolve_action_references(
action: PlanAction,
) -> tuple[int, str, str]:
local_references: _LocalReferenceIndex,
) -> PlanAction:
desired = action.desired
if desired is None:
return (999, action.endpoint_id, "")
raise RuntimeError(
"Executable plan action requires desired state"
)
return (
_OBJECT_TYPE_ORDER.get(desired.object_type, 100),
action.endpoint_id,
desired.logical_key,
if not desired.references:
return action
references = tuple(
_resolved_reference(
action.endpoint_id,
reference,
local_references,
)
for reference in desired.references
)
resolved_desired = replace(
desired,
references=references,
)
return replace(
action,
desired=resolved_desired,
)
def _validated_apply_result(
action: PlanAction,
result: ApplyResult | None,
) -> ApplyResult:
desired = action.desired
if desired is None:
raise RuntimeError(
"Executable plan action requires desired state"
)
if not isinstance(result, ApplyResult):
raise RuntimeError(
f"{action.endpoint_id} "
f"{desired.object_type} "
f"{desired.logical_key}: "
"adapter apply did not return ApplyResult"
)
applied = result.object
if applied.object_type != desired.object_type:
raise RuntimeError(
"Applied object type does not match desired object type"
)
if applied.logical_key != desired.logical_key:
raise RuntimeError(
"Applied object logical key does not match desired "
"logical key"
)
if applied.references != desired.references:
raise RuntimeError(
"Applied object references do not match resolved "
"desired references"
)
return result
def _required_capability(
adapter: ExecutionAdapter,
@@ -164,25 +376,46 @@ def execute_endpoint_plans(
plans,
)
applied: list[ExecutedAction] = []
local_references: dict[
str,
_LocalReferenceIndex,
] = {}
for action in sorted(executable, key=_action_sort_key):
for endpoint_id in adapters_by_endpoint:
local_references[endpoint_id] = (
_local_reference_index(
store,
endpoint_id,
)
)
for action in order_plan_actions(executable):
adapter = adapters_by_endpoint[action.endpoint_id]
desired = action.desired
resolved_action = _resolve_action_references(
action,
local_references[action.endpoint_id],
)
desired = resolved_action.desired
result = _validated_apply_result(
resolved_action,
adapter.apply(resolved_action),
)
applied_object = result.object
adapter.apply(action)
store.record_applied(
action.endpoint_id,
desired,
applied_object,
timestamp,
)
try:
verified = adapter.verify(desired)
verified = adapter.verify(applied_object)
except Exception as exc:
error = str(exc)
store.record_verification_failure(
action.endpoint_id,
desired,
applied_object,
error,
timestamp,
)
@@ -196,7 +429,7 @@ def execute_endpoint_plans(
error = "adapter returned false"
store.record_verification_failure(
action.endpoint_id,
desired,
applied_object,
error,
timestamp,
)
@@ -208,9 +441,16 @@ def execute_endpoint_plans(
store.record_verification_success(
action.endpoint_id,
desired,
applied_object,
timestamp,
)
_remember_local_object(
local_references[action.endpoint_id],
applied_object,
authoritative=True,
)
applied.append(
ExecutedAction(
endpoint_id=action.endpoint_id,
dns_sync/interfaces.py
+2 -1
View File
@@ -1,6 +1,7 @@
from typing import Protocol
from .adapters import (
ApplyResult,
Capability,
DiscoveredObject,
Endpoint,
@@ -17,7 +18,7 @@ class AdapterPlugin(Protocol):
def connect(self) -> None: ...
def discover(self, object_type: str) -> list[DiscoveredObject]: ...
def apply(self, action: PlanAction) -> str | None: ...
def apply(self, action: PlanAction) -> ApplyResult | None: ...
def verify(self, object: NormalizedObject) -> bool: ...
dns_sync/plugins/adapters/verizon.py
+39 -7
View File
@@ -1,5 +1,6 @@
from ...adapters import (
DNS_RECORDS,
ApplyResult,
Capability,
CapabilityOperation,
DiscoveredObject,
@@ -119,7 +120,7 @@ class Plugin:
wildcard=False,
)
def apply(self, action: PlanAction) -> str | None:
def apply(self, action: PlanAction) -> ApplyResult | None:
if action.endpoint_id != self.endpoint.identity:
raise RuntimeError(
"Plan action belongs to a different endpoint"
@@ -139,13 +140,44 @@ class Plugin:
record = self._desired_record(action.desired)
if action.operation == PlanOperation.CREATE:
return self.provider.add_record(record)
if action.operation == PlanOperation.UPDATE:
return self.provider.update_record(record)
detail = self.provider.add_record(record)
elif action.operation == PlanOperation.UPDATE:
detail = self.provider.update_record(record)
else:
raise RuntimeError(
f"Verizon adapter does not support "
f"{action.operation.value} actions"
)
raise RuntimeError(
f"Verizon adapter does not support "
f"{action.operation.value} actions"
records = self.provider.list_records()
applied_record = records.get(record.hostname.lower())
if applied_record is None:
applied_record = records.get(record.hostname)
if applied_record is None:
raise RuntimeError(
"Verizon mutation succeeded but the applied DNS "
f"record {record.hostname!r} could not be rediscovered"
)
slot = applied_record.metadata.get("slot")
if isinstance(slot, bool) or not isinstance(slot, int):
raise RuntimeError(
f"Verizon DNS record {record.hostname!r} "
"has no valid endpoint-local slot after apply"
)
applied = NormalizedObject(
object_type=action.desired.object_type,
logical_key=action.desired.logical_key,
local_key=str(slot),
payload=dict(action.desired.payload),
references=action.desired.references,
)
return ApplyResult(
object=applied,
detail=detail,
)
def verify(self, object: NormalizedObject) -> bool:
dns_sync/reconciliation.py
+2 -1
View File
@@ -13,6 +13,7 @@ from .adapters import (
PlanAction,
PlanOperation,
)
from .dependencies import order_plan_actions
ObservationGroupKey = tuple[str, str]
@@ -255,7 +256,7 @@ def build_endpoint_plans(
plans.append(
EndpointPlan(
endpoint_id=endpoint_id,
actions=tuple(actions),
actions=order_plan_actions(actions),
)
)