Persist generic object references

dns_sync/persistence.py

@@ -6,7 +6,11 @@ from datetime import datetime, timezone
 from os import PathLike
 from typing import Any
 
-from .adapters import NormalizedObject, Observation
+from .adapters import (
+    NormalizedObject,
+    ObjectReference,
+    Observation,
+)
 
 
 _FORBIDDEN_PERSISTED_KEYS = frozenset(
@@ -116,6 +120,71 @@ def _deserialize_payload(value: str) -> dict[str, Any]:
     return payload
 
 
+def _serialize_references(
+    references: Iterable[ObjectReference],
+) -> str:
+    return json.dumps(
+        [
+            {
+                "relation": reference.relation,
+                "object_type": reference.object_type,
+                "local_key": reference.local_key,
+                "logical_key": reference.logical_key,
+            }
+            for reference in references
+        ],
+        sort_keys=True,
+        separators=(",", ":"),
+    )
+
+
+def _deserialize_references(
+    value: str,
+) -> tuple[ObjectReference, ...]:
+    try:
+        serialized = json.loads(value)
+    except json.JSONDecodeError as exc:
+        raise RuntimeError(
+            "Persisted object references are invalid JSON"
+        ) from exc
+
+    if not isinstance(serialized, list):
+        raise RuntimeError(
+            "Persisted object references are not a list"
+        )
+
+    expected_fields = {
+        "relation",
+        "object_type",
+        "local_key",
+        "logical_key",
+    }
+    references: list[ObjectReference] = []
+    for item in serialized:
+        if (
+            not isinstance(item, dict)
+            or set(item) != expected_fields
+        ):
+            raise RuntimeError(
+                "Persisted object reference has invalid fields"
+            )
+        try:
+            references.append(
+                ObjectReference(
+                    relation=item["relation"],
+                    object_type=item["object_type"],
+                    local_key=item["local_key"],
+                    logical_key=item["logical_key"],
+                )
+            )
+        except (TypeError, ValueError) as exc:
+            raise RuntimeError(
+                "Persisted object reference is invalid"
+            ) from exc
+
+    return tuple(references)
+
+
 class SQLiteStateStore:
     def __init__(self, path: str | PathLike[str]):
         self.path = str(path)
@@ -185,7 +254,7 @@ class SQLiteStateStore:
                 )
 
             version = versions[0]["version"]
-            if version not in (1, 2):
+            if version not in (1, 2, 3):
                 raise RuntimeError(
                     "Unsupported SQLite state schema version: "
                     f"{version!r}"
@@ -232,6 +301,31 @@ class SQLiteStateStore:
                     SET version = 2;
                     """
                 )
+                version = 2
+
+            if version == 2:
+                migrations = (
+                    ("observations", "references_json"),
+                    ("desired_state", "references_json"),
+                    ("applied_state", "references_json"),
+                )
+                for table, column in migrations:
+                    existing_columns = {
+                        row["name"]
+                        for row in self.connection.execute(
+                            f"PRAGMA table_info({table})"
+                        )
+                    }
+                    if column not in existing_columns:
+                        self.connection.execute(
+                            f"ALTER TABLE {table} "
+                            f"ADD COLUMN {column} TEXT NOT NULL "
+                            "DEFAULT '[]'"
+                        )
+
+                self.connection.execute(
+                    "UPDATE schema_meta SET version = 3"
+                )
 
     def close(self) -> None:
         self.connection.close()
@@ -254,7 +348,9 @@ class SQLiteStateStore:
             snapshot_at,
             "Observation snapshot timestamp",
         )
-        prepared: list[tuple[str, str, str, str, str, str]] = []
+        prepared: list[
+            tuple[str, str, str, str, str, str, str]
+        ] = []
         local_keys: set[str] = set()
 
         for observation in rows:
@@ -284,6 +380,7 @@ class SQLiteStateStore:
                     normalized.logical_key,
                     normalized.local_key,
                     _serialize_payload(normalized.payload),
+                    _serialize_references(normalized.references),
                     _timestamp(
                         observation.observed_at,
                         "Observation timestamp",
@@ -307,8 +404,9 @@ class SQLiteStateStore:
                     logical_key,
                     local_key,
                     payload_json,
+                    references_json,
                     observed_at
-                ) VALUES (?, ?, ?, ?, ?, ?)
+                ) VALUES (?, ?, ?, ?, ?, ?, ?)
                 """,
                 prepared,
             )
@@ -378,6 +476,7 @@ class SQLiteStateStore:
                 logical_key,
                 local_key,
                 payload_json,
+                references_json,
                 observed_at
             FROM observations
         """
@@ -402,6 +501,9 @@ class SQLiteStateStore:
                     logical_key=row["logical_key"],
                     local_key=row["local_key"],
                     payload=_deserialize_payload(row["payload_json"]),
+                    references=_deserialize_references(
+                        row["references_json"]
+                    ),
                 ),
                 observed_at=_parse_timestamp(row["observed_at"]),
             )
@@ -419,7 +521,9 @@ class SQLiteStateStore:
             selected_at,
             "Desired-state selection timestamp",
         )
-        prepared: list[tuple[str, str, str, str]] = []
+        prepared: list[
+            tuple[str, str, str, str, str]
+        ] = []
         logical_keys: set[str] = set()
 
         for normalized in rows:
@@ -444,6 +548,7 @@ class SQLiteStateStore:
                     object_type,
                     normalized.logical_key,
                     _serialize_payload(normalized.payload),
+                    _serialize_references(normalized.references),
                     selected_timestamp,
                 )
             )
@@ -462,8 +567,9 @@ class SQLiteStateStore:
                     object_type,
                     logical_key,
                     payload_json,
+                    references_json,
                     selected_at
-                ) VALUES (?, ?, ?, ?)
+                ) VALUES (?, ?, ?, ?, ?)
                 """,
                 prepared,
             )
@@ -477,6 +583,7 @@ class SQLiteStateStore:
                 object_type,
                 logical_key,
                 payload_json,
+                references_json,
                 selected_at
             FROM desired_state
         """
@@ -493,6 +600,9 @@ class SQLiteStateStore:
                     object_type=row["object_type"],
                     logical_key=row["logical_key"],
                     payload=_deserialize_payload(row["payload_json"]),
+                    references=_deserialize_references(
+                        row["references_json"]
+                    ),
                 ),
                 selected_at=_parse_timestamp(row["selected_at"]),
             )
@@ -509,6 +619,9 @@ class SQLiteStateStore:
             raise RuntimeError("Applied-state endpoint_id cannot be empty")
 
         payload_json = _serialize_payload(object.payload)
+        references_json = _serialize_references(
+            object.references
+        )
         applied_timestamp = _timestamp(
             applied_at,
             "Application timestamp",
@@ -523,11 +636,12 @@ class SQLiteStateStore:
                     logical_key,
                     local_key,
                     payload_json,
+                    references_json,
                     applied_at,
                     verification_attempted_at,
                     verified_at,
                     verification_error
-                ) VALUES (?, ?, ?, ?, ?, ?, NULL, NULL, NULL)
+                ) VALUES (?, ?, ?, ?, ?, ?, ?, NULL, NULL, NULL)
                 ON CONFLICT(
                     endpoint_id,
                     object_type,
@@ -535,6 +649,7 @@ class SQLiteStateStore:
                 ) DO UPDATE SET
                     local_key = excluded.local_key,
                     payload_json = excluded.payload_json,
+                    references_json = excluded.references_json,
                     applied_at = excluded.applied_at,
                     verification_attempted_at = NULL,
                     verified_at = NULL,
@@ -546,6 +661,7 @@ class SQLiteStateStore:
                     object.logical_key,
                     object.local_key,
                     payload_json,
+                    references_json,
                     applied_timestamp,
                 ),
             )
@@ -564,6 +680,9 @@ class SQLiteStateStore:
             "Verification timestamp",
         )
         payload_json = _serialize_payload(object.payload)
+        references_json = _serialize_references(
+            object.references
+        )
 
         if succeeded:
             verified_at = attempted_timestamp
@@ -587,6 +706,7 @@ class SQLiteStateStore:
                     AND object_type = ?
                     AND logical_key = ?
                     AND payload_json = ?
+                    AND references_json = ?
                 """,
                 (
                     attempted_timestamp,
@@ -596,6 +716,7 @@ class SQLiteStateStore:
                     object.object_type,
                     object.logical_key,
                     payload_json,
+                    references_json,
                 ),
             )
             if cursor.rowcount != 1:
@@ -655,6 +776,7 @@ class SQLiteStateStore:
                 logical_key,
                 local_key,
                 payload_json,
+                references_json,
                 applied_at,
                 verification_attempted_at,
                 verified_at,
@@ -682,6 +804,9 @@ class SQLiteStateStore:
                     logical_key=row["logical_key"],
                     local_key=row["local_key"],
                     payload=_deserialize_payload(row["payload_json"]),
+                    references=_deserialize_references(
+                        row["references_json"]
+                    ),
                 ),
                 applied_at=_parse_timestamp(row["applied_at"]),
                 verification_attempted_at=_parse_timestamp(

tests/test_reconciliation_persistence.py

@@ -203,7 +203,7 @@ class AppliedStatePersistenceTests(unittest.TestCase):
 
 
 class SchemaMigrationTests(unittest.TestCase):
-    def test_version_one_database_is_migrated_to_version_two(self):
+    def test_version_one_database_is_migrated_to_version_three(self):
         with tempfile.TemporaryDirectory() as directory:
             path = Path(directory) / "state.sqlite3"
             connection = sqlite3.connect(path)
@@ -230,10 +230,29 @@ class SchemaMigrationTests(unittest.TestCase):
                         """
                     )
                 }
+                reference_columns = {
+                    table: {
+                        row[1]
+                        for row in store.connection.execute(
+                            f"PRAGMA table_info({table})"
+                        )
+                    }
+                    for table in (
+                        "observations",
+                        "desired_state",
+                        "applied_state",
+                    )
+                }
 
-        self.assertEqual(version, 2)
+        self.assertEqual(version, 3)
         self.assertIn("desired_state", tables)
         self.assertIn("applied_state", tables)
+        self.assertTrue(
+            all(
+                "references_json" in columns
+                for columns in reference_columns.values()
+            )
+        )
 
 
 if __name__ == "__main__":

tests/test_reference_persistence.py

@@ -0,0 +1,142 @@
+import unittest
+from datetime import datetime, timezone
+
+from dns_sync.adapters import (
+    CERTIFICATES,
+    PROXY_HOSTS,
+    NormalizedObject,
+    ObjectReference,
+    Observation,
+)
+from dns_sync.persistence import SQLiteStateStore
+
+
+ENDPOINT = "npm://192.168.1.4:81"
+NOW = datetime(2026, 9, 20, 0, 0, tzinfo=timezone.utc)
+
+
+def local_certificate_reference() -> ObjectReference:
+    return ObjectReference(
+        relation="certificate",
+        object_type=CERTIFICATES,
+        local_key="67",
+    )
+
+
+def logical_certificate_reference() -> ObjectReference:
+    return ObjectReference(
+        relation="certificate",
+        object_type=CERTIFICATES,
+        logical_key="app.example.com",
+    )
+
+
+def proxy_host(
+    *,
+    local_key: str | None,
+    reference: ObjectReference,
+) -> 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": "192.168.1.20",
+            "forward_port": 8080,
+            "enabled": True,
+        },
+        references=(reference,),
+    )
+
+
+class ReferencePersistenceTests(unittest.TestCase):
+    def test_observed_local_reference_round_trips(self):
+        observed = Observation(
+            endpoint_id=ENDPOINT,
+            object=proxy_host(
+                local_key="12",
+                reference=local_certificate_reference(),
+            ),
+            observed_at=NOW,
+        )
+
+        with SQLiteStateStore(":memory:") as store:
+            store.replace_observations(
+                ENDPOINT,
+                PROXY_HOSTS,
+                [observed],
+                NOW,
+            )
+            loaded = store.load_observations(PROXY_HOSTS)
+
+        self.assertEqual(loaded, [observed])
+        self.assertEqual(
+            loaded[0].object.references[0].local_key,
+            "67",
+        )
+
+    def test_desired_logical_reference_round_trips(self):
+        desired = proxy_host(
+            local_key=None,
+            reference=logical_certificate_reference(),
+        )
+
+        with SQLiteStateStore(":memory:") as store:
+            store.replace_desired_state(
+                PROXY_HOSTS,
+                [desired],
+                NOW,
+            )
+            loaded = store.load_desired_state(PROXY_HOSTS)
+
+        self.assertEqual(loaded[0].object, desired)
+        self.assertEqual(
+            loaded[0].object.references[0].logical_key,
+            "app.example.com",
+        )
+
+    def test_applied_reference_participates_in_verification(self):
+        applied = proxy_host(
+            local_key="22",
+            reference=logical_certificate_reference(),
+        )
+        different_reference = proxy_host(
+            local_key="22",
+            reference=ObjectReference(
+                relation="certificate",
+                object_type=CERTIFICATES,
+                logical_key="other.example.com",
+            ),
+        )
+
+        with SQLiteStateStore(":memory:") as store:
+            store.record_applied(ENDPOINT, applied, NOW)
+
+            with self.assertRaisesRegex(
+                RuntimeError,
+                "does not match",
+            ):
+                store.record_verification_success(
+                    ENDPOINT,
+                    different_reference,
+                    NOW,
+                )
+
+            store.record_verification_success(
+                ENDPOINT,
+                applied,
+                NOW,
+            )
+            loaded = store.load_applied_state(
+                ENDPOINT,
+                PROXY_HOSTS,
+            )
+
+        self.assertEqual(loaded[0].object, applied)
+        self.assertTrue(loaded[0].is_verified)
+
+
+if __name__ == "__main__":
+    unittest.main()