Persist desired and verified reconciliation state

dns_sync/persistence.py

@@ -33,6 +33,29 @@ class ObservationStatus:
         return self.last_error is None
 
 
+@dataclass(frozen=True)
+class DesiredState:
+    object: NormalizedObject
+    selected_at: datetime
+
+
+@dataclass(frozen=True)
+class AppliedState:
+    endpoint_id: str
+    object: NormalizedObject
+    applied_at: datetime
+    verification_attempted_at: datetime | None
+    verified_at: datetime | None
+    verification_error: str | None
+
+    @property
+    def is_verified(self) -> bool:
+        return (
+            self.verified_at is not None
+            and self.verification_error is None
+        )
+
+
 def _utc(value: datetime, label: str) -> datetime:
     if value.tzinfo is None:
         raise RuntimeError(f"{label} must be timezone-aware")
@@ -153,6 +176,63 @@ class SQLiteStateStore:
                 """
             )
 
+            versions = self.connection.execute(
+                "SELECT version FROM schema_meta"
+            ).fetchall()
+            if len(versions) != 1:
+                raise RuntimeError(
+                    "SQLite state schema has invalid version metadata"
+                )
+
+            version = versions[0]["version"]
+            if version not in (1, 2):
+                raise RuntimeError(
+                    "Unsupported SQLite state schema version: "
+                    f"{version!r}"
+                )
+
+            if version == 1:
+                self.connection.executescript(
+                    """
+                    CREATE TABLE desired_state (
+                        object_type TEXT NOT NULL,
+                        logical_key TEXT NOT NULL,
+                        payload_json TEXT NOT NULL,
+                        selected_at TEXT NOT NULL,
+                        PRIMARY KEY (
+                            object_type,
+                            logical_key
+                        )
+                    );
+
+                    CREATE TABLE applied_state (
+                        endpoint_id TEXT NOT NULL,
+                        object_type TEXT NOT NULL,
+                        logical_key TEXT NOT NULL,
+                        local_key TEXT,
+                        payload_json TEXT NOT NULL,
+                        applied_at TEXT NOT NULL,
+                        verification_attempted_at TEXT,
+                        verified_at TEXT,
+                        verification_error TEXT,
+                        PRIMARY KEY (
+                            endpoint_id,
+                            object_type,
+                            logical_key
+                        )
+                    );
+
+                    CREATE INDEX applied_state_by_object
+                    ON applied_state (
+                        object_type,
+                        logical_key
+                    );
+
+                    UPDATE schema_meta
+                    SET version = 2;
+                    """
+                )
+
     def close(self) -> None:
         self.connection.close()
 
@@ -328,6 +408,291 @@ class SQLiteStateStore:
             for row in rows
         ]
 
+    def replace_desired_state(
+        self,
+        object_type: str,
+        objects: Iterable[NormalizedObject],
+        selected_at: datetime,
+    ) -> None:
+        rows = list(objects)
+        selected_timestamp = _timestamp(
+            selected_at,
+            "Desired-state selection timestamp",
+        )
+        prepared: list[tuple[str, str, str, str]] = []
+        logical_keys: set[str] = set()
+
+        for normalized in rows:
+            if normalized.object_type != object_type:
+                raise RuntimeError(
+                    "Desired object type does not match snapshot type"
+                )
+            if normalized.local_key is not None:
+                raise RuntimeError(
+                    "Canonical desired objects cannot contain "
+                    "an endpoint-local key"
+                )
+            if normalized.logical_key in logical_keys:
+                raise RuntimeError(
+                    "Duplicate desired logical key: "
+                    f"{normalized.logical_key}"
+                )
+
+            logical_keys.add(normalized.logical_key)
+            prepared.append(
+                (
+                    object_type,
+                    normalized.logical_key,
+                    _serialize_payload(normalized.payload),
+                    selected_timestamp,
+                )
+            )
+
+        with self.connection:
+            self.connection.execute(
+                """
+                DELETE FROM desired_state
+                WHERE object_type = ?
+                """,
+                (object_type,),
+            )
+            self.connection.executemany(
+                """
+                INSERT INTO desired_state (
+                    object_type,
+                    logical_key,
+                    payload_json,
+                    selected_at
+                ) VALUES (?, ?, ?, ?)
+                """,
+                prepared,
+            )
+
+    def load_desired_state(
+        self,
+        object_type: str | None = None,
+    ) -> list[DesiredState]:
+        sql = """
+            SELECT
+                object_type,
+                logical_key,
+                payload_json,
+                selected_at
+            FROM desired_state
+        """
+        parameters: tuple[str, ...] = ()
+        if object_type is not None:
+            sql += " WHERE object_type = ?"
+            parameters = (object_type,)
+        sql += " ORDER BY object_type, logical_key"
+
+        rows = self.connection.execute(sql, parameters).fetchall()
+        return [
+            DesiredState(
+                object=NormalizedObject(
+                    object_type=row["object_type"],
+                    logical_key=row["logical_key"],
+                    payload=_deserialize_payload(row["payload_json"]),
+                ),
+                selected_at=_parse_timestamp(row["selected_at"]),
+            )
+            for row in rows
+        ]
+
+    def record_applied(
+        self,
+        endpoint_id: str,
+        object: NormalizedObject,
+        applied_at: datetime,
+    ) -> None:
+        if not endpoint_id:
+            raise RuntimeError("Applied-state endpoint_id cannot be empty")
+
+        payload_json = _serialize_payload(object.payload)
+        applied_timestamp = _timestamp(
+            applied_at,
+            "Application timestamp",
+        )
+
+        with self.connection:
+            self.connection.execute(
+                """
+                INSERT INTO applied_state (
+                    endpoint_id,
+                    object_type,
+                    logical_key,
+                    local_key,
+                    payload_json,
+                    applied_at,
+                    verification_attempted_at,
+                    verified_at,
+                    verification_error
+                ) VALUES (?, ?, ?, ?, ?, ?, NULL, NULL, NULL)
+                ON CONFLICT(
+                    endpoint_id,
+                    object_type,
+                    logical_key
+                ) DO UPDATE SET
+                    local_key = excluded.local_key,
+                    payload_json = excluded.payload_json,
+                    applied_at = excluded.applied_at,
+                    verification_attempted_at = NULL,
+                    verified_at = NULL,
+                    verification_error = NULL
+                """,
+                (
+                    endpoint_id,
+                    object.object_type,
+                    object.logical_key,
+                    object.local_key,
+                    payload_json,
+                    applied_timestamp,
+                ),
+            )
+
+    def _record_verification(
+        self,
+        endpoint_id: str,
+        object: NormalizedObject,
+        attempted_at: datetime,
+        *,
+        succeeded: bool,
+        error: str | None,
+    ) -> None:
+        attempted_timestamp = _timestamp(
+            attempted_at,
+            "Verification timestamp",
+        )
+        payload_json = _serialize_payload(object.payload)
+
+        if succeeded:
+            verified_at = attempted_timestamp
+            verification_error = None
+        else:
+            verified_at = None
+            verification_error = str(
+                error or "verification failed"
+            )
+
+        with self.connection:
+            cursor = self.connection.execute(
+                """
+                UPDATE applied_state
+                SET
+                    verification_attempted_at = ?,
+                    verified_at = COALESCE(?, verified_at),
+                    verification_error = ?
+                WHERE
+                    endpoint_id = ?
+                    AND object_type = ?
+                    AND logical_key = ?
+                    AND payload_json = ?
+                """,
+                (
+                    attempted_timestamp,
+                    verified_at,
+                    verification_error,
+                    endpoint_id,
+                    object.object_type,
+                    object.logical_key,
+                    payload_json,
+                ),
+            )
+            if cursor.rowcount != 1:
+                raise RuntimeError(
+                    "Verification does not match the currently "
+                    "applied object"
+                )
+
+    def record_verification_success(
+        self,
+        endpoint_id: str,
+        object: NormalizedObject,
+        verified_at: datetime,
+    ) -> None:
+        self._record_verification(
+            endpoint_id,
+            object,
+            verified_at,
+            succeeded=True,
+            error=None,
+        )
+
+    def record_verification_failure(
+        self,
+        endpoint_id: str,
+        object: NormalizedObject,
+        error: str,
+        attempted_at: datetime,
+    ) -> None:
+        self._record_verification(
+            endpoint_id,
+            object,
+            attempted_at,
+            succeeded=False,
+            error=error,
+        )
+
+    def load_applied_state(
+        self,
+        endpoint_id: str | None = None,
+        object_type: str | None = None,
+    ) -> list[AppliedState]:
+        clauses: list[str] = []
+        parameters: list[str] = []
+
+        if endpoint_id is not None:
+            clauses.append("endpoint_id = ?")
+            parameters.append(endpoint_id)
+        if object_type is not None:
+            clauses.append("object_type = ?")
+            parameters.append(object_type)
+
+        sql = """
+            SELECT
+                endpoint_id,
+                object_type,
+                logical_key,
+                local_key,
+                payload_json,
+                applied_at,
+                verification_attempted_at,
+                verified_at,
+                verification_error
+            FROM applied_state
+        """
+        if clauses:
+            sql += " WHERE " + " AND ".join(clauses)
+        sql += """
+            ORDER BY
+                endpoint_id,
+                object_type,
+                logical_key
+        """
+
+        rows = self.connection.execute(
+            sql,
+            tuple(parameters),
+        ).fetchall()
+        return [
+            AppliedState(
+                endpoint_id=row["endpoint_id"],
+                object=NormalizedObject(
+                    object_type=row["object_type"],
+                    logical_key=row["logical_key"],
+                    local_key=row["local_key"],
+                    payload=_deserialize_payload(row["payload_json"]),
+                ),
+                applied_at=_parse_timestamp(row["applied_at"]),
+                verification_attempted_at=_parse_timestamp(
+                    row["verification_attempted_at"]
+                ),
+                verified_at=_parse_timestamp(row["verified_at"]),
+                verification_error=row["verification_error"],
+            )
+            for row in rows
+        ]
+
     def observation_status(
         self,
         endpoint_id: str,

tests/test_reconciliation_persistence.py

@@ -0,0 +1,240 @@
+import sqlite3
+import tempfile
+import unittest
+from datetime import datetime, timezone
+from pathlib import Path
+
+from dns_sync.adapters import CERTIFICATES, NormalizedObject
+from dns_sync.persistence import SQLiteStateStore
+
+
+ENDPOINT = "npm://192.168.1.3:81"
+T1 = datetime(2026, 9, 19, 21, 0, tzinfo=timezone.utc)
+T2 = datetime(2026, 9, 19, 21, 5, tzinfo=timezone.utc)
+T3 = datetime(2026, 9, 19, 21, 10, tzinfo=timezone.utc)
+
+
+def certificate(
+    expires_on: str,
+    local_key: str | None = None,
+) -> NormalizedObject:
+    return NormalizedObject(
+        object_type=CERTIFICATES,
+        logical_key="app.example.com",
+        local_key=local_key,
+        payload={
+            "domain_names": ["app.example.com"],
+            "expires_on": expires_on,
+        },
+    )
+
+
+class DesiredStatePersistenceTests(unittest.TestCase):
+    def test_snapshot_round_trips_and_can_be_cleared(self):
+        desired = certificate("2026-12-15T00:00:00+00:00")
+
+        with SQLiteStateStore(":memory:") as store:
+            store.replace_desired_state(
+                CERTIFICATES,
+                [desired],
+                T1,
+            )
+            loaded = store.load_desired_state(CERTIFICATES)
+
+            store.replace_desired_state(
+                CERTIFICATES,
+                [],
+                T2,
+            )
+            cleared = store.load_desired_state(CERTIFICATES)
+
+        self.assertEqual(len(loaded), 1)
+        self.assertEqual(loaded[0].object, desired)
+        self.assertEqual(loaded[0].selected_at, T1)
+        self.assertEqual(cleared, [])
+
+    def test_endpoint_local_identity_is_rejected(self):
+        with SQLiteStateStore(":memory:") as store:
+            with self.assertRaisesRegex(
+                RuntimeError,
+                "endpoint-local key",
+            ):
+                store.replace_desired_state(
+                    CERTIFICATES,
+                    [certificate("new", local_key="67")],
+                    T1,
+                )
+
+    def test_sensitive_payload_is_rejected_before_snapshot_changes(self):
+        safe = certificate("safe")
+        unsafe = NormalizedObject(
+            object_type=CERTIFICATES,
+            logical_key="app.example.com",
+            payload={
+                "domain_names": ["app.example.com"],
+                "nested": {
+                    "private_key": "must-never-persist",
+                },
+            },
+        )
+
+        with SQLiteStateStore(":memory:") as store:
+            store.replace_desired_state(
+                CERTIFICATES,
+                [safe],
+                T1,
+            )
+
+            with self.assertRaisesRegex(
+                RuntimeError,
+                "sensitive field",
+            ):
+                store.replace_desired_state(
+                    CERTIFICATES,
+                    [unsafe],
+                    T2,
+                )
+
+            loaded = store.load_desired_state(CERTIFICATES)
+
+        self.assertEqual(len(loaded), 1)
+        self.assertEqual(loaded[0].object, safe)
+        self.assertEqual(loaded[0].selected_at, T1)
+
+
+class AppliedStatePersistenceTests(unittest.TestCase):
+    def test_apply_requires_successful_verification(self):
+        first = certificate("first", local_key="91")
+        second = certificate("second", local_key="91")
+
+        with SQLiteStateStore(":memory:") as store:
+            store.record_applied(ENDPOINT, first, T1)
+            before = store.load_applied_state(
+                ENDPOINT,
+                CERTIFICATES,
+            )[0]
+
+            store.record_verification_success(
+                ENDPOINT,
+                first,
+                T2,
+            )
+            verified = store.load_applied_state(
+                ENDPOINT,
+                CERTIFICATES,
+            )[0]
+
+            store.record_applied(ENDPOINT, second, T3)
+            reapplied = store.load_applied_state(
+                ENDPOINT,
+                CERTIFICATES,
+            )[0]
+
+        self.assertFalse(before.is_verified)
+        self.assertIsNone(before.verified_at)
+
+        self.assertTrue(verified.is_verified)
+        self.assertEqual(verified.verified_at, T2)
+        self.assertEqual(verified.verification_attempted_at, T2)
+
+        self.assertEqual(
+            reapplied.object.payload["expires_on"],
+            "second",
+        )
+        self.assertEqual(reapplied.applied_at, T3)
+        self.assertFalse(reapplied.is_verified)
+        self.assertIsNone(reapplied.verification_attempted_at)
+        self.assertIsNone(reapplied.verified_at)
+
+    def test_stale_payload_cannot_verify_current_applied_state(self):
+        old = certificate("old", local_key="91")
+        current = certificate("current", local_key="91")
+
+        with SQLiteStateStore(":memory:") as store:
+            store.record_applied(ENDPOINT, current, T1)
+
+            with self.assertRaisesRegex(
+                RuntimeError,
+                "does not match",
+            ):
+                store.record_verification_success(
+                    ENDPOINT,
+                    old,
+                    T2,
+                )
+
+            state = store.load_applied_state(
+                ENDPOINT,
+                CERTIFICATES,
+            )[0]
+
+        self.assertFalse(state.is_verified)
+        self.assertIsNone(state.verification_attempted_at)
+        self.assertIsNone(state.verified_at)
+
+    def test_failed_verification_is_not_reported_as_verified(self):
+        applied = certificate("current", local_key="91")
+
+        with SQLiteStateStore(":memory:") as store:
+            store.record_applied(ENDPOINT, applied, T1)
+            store.record_verification_success(
+                ENDPOINT,
+                applied,
+                T2,
+            )
+            store.record_verification_failure(
+                ENDPOINT,
+                applied,
+                "target no longer matches",
+                T3,
+            )
+            state = store.load_applied_state(
+                ENDPOINT,
+                CERTIFICATES,
+            )[0]
+
+        self.assertFalse(state.is_verified)
+        self.assertEqual(state.verified_at, T2)
+        self.assertEqual(state.verification_attempted_at, T3)
+        self.assertEqual(
+            state.verification_error,
+            "target no longer matches",
+        )
+
+
+class SchemaMigrationTests(unittest.TestCase):
+    def test_version_one_database_is_migrated_to_version_two(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "state.sqlite3"
+            connection = sqlite3.connect(path)
+            connection.execute(
+                "CREATE TABLE schema_meta (version INTEGER NOT NULL)"
+            )
+            connection.execute(
+                "INSERT INTO schema_meta(version) VALUES (1)"
+            )
+            connection.commit()
+            connection.close()
+
+            with SQLiteStateStore(path) as store:
+                version = store.connection.execute(
+                    "SELECT version FROM schema_meta"
+                ).fetchone()[0]
+                tables = {
+                    row[0]
+                    for row in store.connection.execute(
+                        """
+                        SELECT name
+                        FROM sqlite_master
+                        WHERE type = 'table'
+                        """
+                    )
+                }
+
+        self.assertEqual(version, 2)
+        self.assertIn("desired_state", tables)
+        self.assertIn("applied_state", tables)
+
+
+if __name__ == "__main__":
+    unittest.main()