Fortify logging: no logging failure may break functionality

emit_event now swallows and reports any failure, including crashes in the
built-in default_emit formatting itself; log_change routes through it.
logfmt chain callbacks that raise are logged and treated as fall-through,
and logmigr callbacks get on_error reporting like fatal_error handlers,
so a broken logging callback can no longer abort a transaction or open.

Demo: raw user ids in v0 logs, logfmt-resolved names in v1 logs.

demo/main.py

@@ -46,7 +46,6 @@ def migrate_v1(d: dict) -> None:
 kanta_v1 = Kanta(filename, Data(), migrations=sys.modules[__name__])
 
 
-@kanta_v0.logfmt
 @kanta_v1.logfmt
 def resolve_user(
     value: str, path: str, previous: DictPre, current: DictPost
@@ -75,9 +74,6 @@ async def main() -> None:
             data.users["userid002"]["role"] = "editor"
             data.counter = 1
 
-        with kanta.transaction(action="delete", user="userid002") as data:
-            del data.users["userid001"]
-
         # Display-only extra string, appended after the action.
         with kanta.transaction(
             action="export", user="userid002", extra="extra info"
@@ -91,11 +87,18 @@ async def main() -> None:
         except ValueError:
             print(f"# Reading does not need transaction: {data.counter=}", flush=True)
 
-        with kanta.transaction(action="import", logdiff=False) as data:
-            data.counter = 3
-
-    print("\n# A later version of our application with new data model and migrations")
+    print(
+        "\n# A later version of our application with new data model, migrations and logfmt"
+    )
     async with kanta_v1 as kanta:
+        with kanta.transaction(
+            action="import", user="userid001", logdiff=False
+        ) as data:
+            data.total = 3
+
+        with kanta.transaction(action="delete", user="userid002") as data:
+            del data.users["userid001"]
+
         with kanta.transaction(
             action="update", user="userid002", extra=filename.name
         ) as data:

docs/database.md

@@ -244,6 +244,11 @@ def resolve_user_key(value: str) -> str | None:
   also call `default_emit(event)` itself to delegate events it does not
   customize.  Operational diagnostics (rollback warnings, integrity errors)
   do not go through this mechanism.
+- Logging never breaks functionality: a crashing `logemit` callback is
+  reported with `logger.exception` and the event falls back to the built-in
+  formatting; if the built-in formatting itself fails, the error is reported
+  and swallowed.  The same applies to `logfmt` callbacks (a failing one is
+  treated as a fall-through) and `logmigr` callbacks.
 
 ```python
 @kanta.logemit

kanta/callbacks.py

@@ -12,6 +12,7 @@ and receive the value plus an optional ``path`` string.  They return
 from __future__ import annotations
 
 import inspect
+import logging
 import types
 from collections.abc import Callable
 from dataclasses import dataclass
@@ -23,6 +24,8 @@ from kanta.migrations import MigrationResult
 DictPre = Annotated[dict, "pre"]
 DictPost = Annotated[dict, "post"]
 
+_logger = logging.getLogger(__name__)
+
 
 class LogFmt:
     """Base class for stateful logfmt callbacks.
@@ -226,7 +229,13 @@ class CallbackRegistry:
             for fn, pattern in formatters:
                 if pattern is not None and path != pattern:
                     continue
+                try:
                     resolved = fn(value, path)
+                except Exception:
+                    # Formatting must never break functionality; a failing
+                    # callback is reported and treated as a fall-through.
+                    _logger.exception("logfmt callback %r failed", fn)
+                    continue
                 if resolved is not None:
                     return resolved
             return None

kanta/kantaimpl.py

@@ -29,6 +29,11 @@ _logger = logging.getLogger(__name__)
 T = TypeVar("T")
 
 
+def _log_callback_error(callback_error, callback):
+    """Report a failing logging callback and continue with the next one."""
+    _logger.exception("Log callback %r failed: %s", callback, callback_error)
+
+
 class KantaImpl(PersistenceMixin, Generic[T]):
     """Internal state and logic for Kanta."""
 
@@ -106,6 +111,7 @@ class KantaImpl(PersistenceMixin, Generic[T]):
                     kanta=self._kanta,
                     migration_result=migration_result,
                 ),
+                on_error=_log_callback_error,
             )
             return
 

kanta/logging.py

@@ -94,7 +94,12 @@ def emit_event(
     A truthy return value passes the event — possibly modified — to the next
     handler.  When all handlers pass, :func:`default_emit` renders the event
     with the built-in formatting.
+
+    Logging must never break functionality: a crashing handler is reported
+    and the chain falls back to the built-in formatting, and a failure in
+    the built-in formatting itself is reported and swallowed.
     """
+    try:
         for handler in handlers:
             try:
                 proceed = handler(ev)
@@ -104,6 +109,8 @@ def emit_event(
             if not proceed:
                 return
         default_emit(ev)
+    except Exception:
+        _logger.exception("failed to emit %s log event", ev.kind)
 
 
 def default_emit(ev: LogEvent) -> None:
@@ -406,9 +413,9 @@ def log_change(
 ) -> None:
     """Log a database change with the built-in formatting.
 
-    Compatibility wrapper around :func:`default_emit`; Kanta itself builds a
-    :class:`LogEvent` and dispatches it through :func:`emit_event` so logemit
-    callbacks see it.
+    Compatibility wrapper around :func:`emit_event` with no handlers; Kanta
+    itself builds a :class:`LogEvent` and dispatches it with the registered
+    logemit callbacks.
 
     Args:
         action: The action name (e.g., "login", "admin:delete_user").
@@ -423,7 +430,7 @@ def log_change(
         log_diff: Whether to build and emit the diff lines.  ``False`` skips
             diff formatting entirely and only the header is logged.
     """
-    default_emit(
+    emit_event(
         LogEvent(
             kind="change",
             logger=logger,

tests/test_logemit.py

@@ -1,4 +1,5 @@
 import logging
+import sys
 
 import pytest
 
@@ -7,10 +8,17 @@ from kanta.logging import (
     bootstrap_logger,
     configure_logging,
     emit_event,
+    log_change,
     migration_logger,
     transaction_logger,
 )
-from tests.support import Data, make_kanta
+from kanta.migrations import MigrationResult
+from tests.support import (
+    Data,
+    fixed_change,
+    make_kanta,
+    seed_single_change,
+)
 
 
 @pytest.fixture(autouse=True)
@@ -155,3 +163,79 @@ def test_logemit_rejects_classes_and_async(tmp_path, format_config):
 
     with pytest.raises(TypeError):
         kanta.logemit(ahandler)
+
+
+def _raise(*args, **kwargs):
+    raise RuntimeError("formatting broken")
+
+
+def test_log_change_never_raises(monkeypatch):
+    monkeypatch.setattr("kanta.logging.format_action_header", _raise)
+    log_change("update", {"counter": 1}, previous={})  # must not raise
+
+
+@pytest.mark.asyncio
+async def test_logging_failure_does_not_break_transaction(
+    tmp_path, format_config, monkeypatch
+):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+    kanta.logemit(_raise)
+    monkeypatch.setattr("kanta.logging.format_action_header", _raise)
+    await kanta.open()
+
+    with kanta.transaction(action="inc") as data:
+        data.counter = 1
+
+    await kanta.close()
+
+    kanta2 = make_kanta(path, Data, format_config)
+    kanta2.logemit(_raise)
+    monkeypatch.setattr("kanta.logging.format_action_header", _raise)
+    await kanta2.open()
+    assert kanta2.data.counter == 1
+    await kanta2.close()
+
+
+@pytest.mark.asyncio
+async def test_logfmt_failure_falls_back_to_default(tmp_path, format_config, caplog):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    @kanta.logfmt
+    def bad(value: str, path: str) -> str | None:
+        raise RuntimeError("broken")
+
+    await kanta.open()
+    with caplog.at_level(logging.INFO, logger="kanta.transaction"):
+        with kanta.transaction(action="inc", user="alice") as data:
+            data.counter = 1
+    await kanta.close()
+
+    assert kanta.data.counter == 1
+    assert "alice" in caplog.text  # raw rendering used despite the failure
+    assert "counter" in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_logmigr_failure_does_not_break_open(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    seed_single_change(path, fixed_change("init", {"counter": 0}), format_config)
+
+    mod = type(sys)("test_migrations_broken_logmigr")
+
+    def migrate_v1(d, kanta):
+        """Bump counter."""
+        d["counter"] = 2
+
+    mod.__dict__["migrate_v1"] = migrate_v1
+
+    kanta = make_kanta(path, Data, format_config, migrations=mod)
+
+    @kanta.logmigr
+    def bad(summary: MigrationResult) -> None:
+        raise RuntimeError("broken")
+
+    await kanta.open()
+    assert kanta.data.counter == 2
+    await kanta.close()