From d33f3f9c2fc77790e265155cf9df35ea18d1d22c Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Wed, 2 Sep 2026 17:02:14 +0000 Subject: [PATCH] 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 | 22 ++++++-- kanta/kanta.py | 21 ++++++++ kanta/kantaimpl.py | 20 +++++++ kanta/transaction.py | 5 ++ tests/test_validate.py | 116 +++++++++++++++++++++++++++++++++++++++++ 5 files changed, 181 insertions(+), 3 deletions(-) create mode 100644 tests/test_validate.py diff --git a/kanta/callbacks.py b/kanta/callbacks.py index c12a5be..0663165 100644 --- a/kanta/callbacks.py +++ b/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": diff --git a/kanta/kanta.py b/kanta/kanta.py index 5641a67..40789e9 100644 --- a/kanta/kanta.py +++ b/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. diff --git a/kanta/kantaimpl.py b/kanta/kantaimpl.py index caa02f7..089a9ba 100644 --- a/kanta/kantaimpl.py +++ b/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( diff --git a/kanta/transaction.py b/kanta/transaction.py index bea48a5..1747eeb 100644 --- a/kanta/transaction.py +++ b/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: diff --git a/tests/test_validate.py b/tests/test_validate.py new file mode 100644 index 0000000..5c5b181 --- /dev/null +++ b/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