Add @kanta.validate integrity-validation callbacks

Validators receive the live data object (and optionally the Kanta
instance) and raise on inconsistency. They run after replay/migrations
during open() and after each transaction before the change is queued;
a failure rolls back the transaction or aborts the open. Multiple
validators run in registration order until the first failure. Sync-only,
since transactions are synchronous.

kanta/callbacks.py

@@ -136,6 +136,7 @@ class CallbackRegistry:
             "bootstrap": [],
             "fatal_error": [],
             "logmigr": [],
+            "validate": [],
         }
         self._logfmt_callbacks: list[_LogFmtFunctionSpec | _LogFmtClassSpec] = []
         self._logemit_callbacks: list[Callable[..., Any]] = []
@@ -174,6 +175,10 @@ class CallbackRegistry:
             raise TypeError(f"{kind} callbacks must be functions, not classes")
         if not callable(callback):
             raise TypeError(f"{kind} callback must be callable")
+        if kind == "validate" and inspect.iscoroutinefunction(callback):
+            raise TypeError(
+                "validate callbacks must not be async (transactions are synchronous)"
+            )
 
         params = self._validate_function(callback, kind)
         is_async = inspect.iscoroutinefunction(callback)
@@ -215,6 +220,16 @@ class CallbackRegistry:
                     break
         return results
 
+    def invoke_sync(self, kind: str, ctx: InjectionContext) -> None:
+        """Invoke all sync callbacks of *kind* in order; first exception raises.
+
+        Used for ``validate`` callbacks, which run inside synchronous
+        transactions and therefore must not be async.
+        """
+        for reg in self._callbacks[kind]:
+            kwargs = self._build_kwargs(reg.params, ctx)
+            reg.callback(**kwargs)
+
     def has(self, kind: str) -> bool:
         """Return True if any callback of *kind* is registered."""
         if kind == "logfmt":
@@ -531,22 +546,23 @@ class CallbackRegistry:
         if bare is MigrationReport:
             return kind == "logmigr"
         if self._data_type is not None and bare is self._data_type:
-            return kind == "bootstrap"
+            return kind in {"bootstrap", "validate"}
         if self._kanta_class is not None and bare is self._kanta_class:
             return kind in {
                 "bootstrap",
                 "fatal_error",
                 "logfmt",
                 "logmigr",
+                "validate",
             }
         return False
 
     def _allowed_message(self, kind: str) -> str:
         parts: list[str] = []
-        if kind == "bootstrap":
+        if kind in {"bootstrap", "validate"}:
             if self._data_type is not None:
                 parts.append(self._data_type.__name__)
-        if kind in {"bootstrap", "fatal_error", "logfmt", "logmigr"}:
+        if kind in {"bootstrap", "fatal_error", "logfmt", "logmigr", "validate"}:
             if self._kanta_class is not None:
                 parts.append(self._kanta_class.__name__)
         if kind == "fatal_error":

kanta/kanta.py

@@ -249,6 +249,27 @@ class Kanta(Generic[T]):
             return _register
         return _register(fn)
 
+    def validate(self, fn):
+        """Register a data validation callback.
+
+        Used as ``@kanta.validate``. The callback receives the live data
+        object (and optionally the ``Kanta`` instance) and must raise an
+        exception when the data is inconsistent. Validators run after replay
+        during :meth:`open` (after msgspec decoding and migrations) and after
+        each transaction, before the change is committed to history. Multiple
+        validators run in registration order until the first failure.
+
+        Validators must be synchronous and must not modify the data — they
+        only fail. A failure inside a transaction rolls the transaction back;
+        a failure during open aborts the open.
+        """
+
+        def _register(callback):
+            self._impl.add_validate(callback)
+            return callback
+
+        return _register(fn)
+
     def fatal_error(self, fn=None):
         """Register fatal error handler callback.
 

kanta/kantaimpl.py

@@ -96,6 +96,10 @@ class KantaImpl(PersistenceMixin, Generic[T]):
         """Register one migration logging callback."""
         self.callback_registry.register("logmigr", callback)
 
+    def add_validate(self, callback) -> None:
+        """Register one data validation callback."""
+        self.callback_registry.register("validate", callback)
+
     def add_logemit(self, callback) -> None:
         """Register one log emitter callback."""
         self.callback_registry.register("logemit", callback)
@@ -247,6 +251,16 @@ class KantaImpl(PersistenceMixin, Generic[T]):
                 self.data_type,
                 serializer=self.serializer,
             )
+            if self.callback_registry.has("validate"):
+                try:
+                    self.callback_registry.invoke_sync(
+                        "validate",
+                        InjectionContext(data=self.data, kanta=self._kanta),
+                    )
+                except Exception:
+                    self.opened = False
+                    self.file.close()
+                    raise
             self.version = rr.version
             self.mtime = rr.m
             if log is not False and not migrations_ran:
@@ -322,6 +336,12 @@ class KantaImpl(PersistenceMixin, Generic[T]):
                         InjectionContext(data=self.data, kanta=self._kanta),
                     )
 
+                if self.callback_registry.has("validate"):
+                    self.callback_registry.invoke_sync(
+                        "validate",
+                        InjectionContext(data=self.data, kanta=self._kanta),
+                    )
+
                 self.statedict = {}
                 current = struct_to_dict(self.data, serializer=self.serializer)
                 record = self.queue_change(

kanta/transaction.py

@@ -88,6 +88,11 @@ def transaction(
         new_dict = struct_to_dict(impl.data, serializer=impl.serializer)
         diff = compute_diff(impl.statedict, new_dict)
         if diff:
+            if impl.callback_registry.has("validate"):
+                impl.callback_registry.invoke_sync(
+                    "validate",
+                    InjectionContext(data=impl.data, kanta=impl._kanta),
+                )
             previous = impl.statedict
             record = impl.queue_change(action, new_dict, user=user, mtime=mtime)
             if record is not None:

tests/test_validate.py

@@ -0,0 +1,116 @@
+"""Tests for the @kanta.validate integrity-validation callbacks."""
+
+import pytest
+
+from tests.support import Data, make_kanta
+
+pytestmark = pytest.mark.asyncio
+
+
+async def test_validate_passes_on_valid_data(tmp_path, format_config):
+    kanta = make_kanta(tmp_path / "d.kantadb", Data, format_config)
+    calls = []
+
+    @kanta.validate
+    def check(data: Data):
+        calls.append(data.counter)
+        assert data.counter >= 0
+
+    async with kanta:
+        with kanta.transaction("inc", log=False) as data:
+            data.counter = 1
+
+    assert calls  # ran during bootstrap/open and the transaction
+
+
+async def test_validate_failure_rolls_back_transaction(tmp_path, format_config):
+    kanta = make_kanta(tmp_path / "d.kantadb", Data, format_config)
+
+    @kanta.validate
+    def check(data: Data):
+        if data.counter < 0:
+            raise ValueError("counter must not go negative")
+
+    await kanta.open(log=False)
+    with pytest.raises(ValueError, match="negative"):
+        with kanta.transaction("dec", log=False) as data:
+            data.counter = -1
+    assert kanta.data.counter == 0  # rolled back
+    await kanta.close()
+
+    # The invalid change never reached the history.
+    kanta2 = make_kanta(tmp_path / "d.kantadb", Data, format_config)
+    async with kanta2:
+        assert kanta2.data.counter == 0
+
+
+async def test_validate_runs_on_open_after_replay(tmp_path, format_config):
+    kanta = make_kanta(tmp_path / "d.kantadb", Data, format_config)
+    async with kanta:
+        with kanta.transaction("set", log=False) as data:
+            data.counter = 5
+
+    kanta2 = make_kanta(tmp_path / "d.kantadb", Data, format_config)
+    seen = []
+
+    @kanta2.validate
+    def check(data: Data):
+        seen.append(data.counter)
+
+    async with kanta2:
+        pass
+    assert 5 in seen
+
+
+async def test_validate_failure_aborts_open(tmp_path, format_config):
+    kanta = make_kanta(tmp_path / "d.kantadb", Data, format_config)
+    async with kanta:
+        with kanta.transaction("set", log=False) as data:
+            data.counter = 5
+
+    kanta2 = make_kanta(tmp_path / "d.kantadb", Data, format_config)
+
+    @kanta2.validate
+    def check(data: Data):
+        raise ValueError("always inconsistent")
+
+    with pytest.raises(ValueError, match="inconsistent"):
+        await kanta2.open(log=False)
+
+    # The failed open released the file: a fresh instance can open it.
+    kanta3 = make_kanta(tmp_path / "d.kantadb", Data, format_config)
+    async with kanta3:
+        assert kanta3.data.counter == 5
+
+
+async def test_multiple_validators_stop_at_first_failure(tmp_path, format_config):
+    kanta = make_kanta(tmp_path / "d.kantadb", Data, format_config)
+    calls = []
+
+    @kanta.validate
+    def first(data: Data):
+        calls.append("first")
+        if data.counter > 1:
+            raise ValueError("too big")
+
+    @kanta.validate
+    def second(data: Data):
+        calls.append("second")
+
+    await kanta.open(log=False)
+    calls.clear()
+    with pytest.raises(ValueError, match="too big"):
+        with kanta.transaction("bump", log=False) as data:
+            data.counter = 2
+    assert calls == ["first"]
+    await kanta.close()
+
+
+async def test_validate_rejects_async_callback(tmp_path, format_config):
+    kanta = make_kanta(tmp_path / "d.kantadb", Data, format_config)
+
+    with pytest.raises(TypeError, match="must not be async"):
+
+        @kanta.validate
+        async def check(data: Data):
+            pass