Implement proper bootstrap and fatal error handling via decorators. Breaks compatibility.

README.md

@@ -51,6 +51,64 @@ asyncio.run(main())
 3. Let Kanta flush queued changes to disk in the background.
 4. Use snapshots and replay for fast startup and full history.
 
+## Bootstrap and Open Modes
+
+Kanta supports open-time bootstrap callbacks for initializing a brand-new
+database before `open()` returns.
+
+Register bootstrap handlers with a decorator:
+
+```python
+kanta = Kanta("data.kantadb", Data())
+
+@kanta.bootstrap(action="seed", user="system")
+def seed_defaults(data) -> None:
+    data.users["admin"] = User(name="Admin")
+
+await kanta.open()
+```
+
+You can also use `@kanta.bootstrap` with no arguments and async handlers:
+
+```python
+@kanta.bootstrap
+async def bootstrap_async(data) -> None:
+    data.counter = 1
+```
+
+When multiple bootstrap handlers are registered:
+- they run in registration order,
+- exactly one bootstrap change record is queued,
+- bootstrap metadata (`action`, `user`, `mtime`) is taken from the last
+  registration.
+
+If any bootstrap handler raises, Kanta closes and removes the database file,
+then re-raises the error.
+
+`open()` also supports strict open mode:
+
+```python
+await kanta.open(create=False)
+```
+
+With `create=False`, open fails if the database file does not exist or is
+empty.
+
+## Fatal Error Handlers
+
+Fatal background write errors can be observed with a decorator:
+
+```python
+import os
+import signal
+
+@kanta.fatal_error
+async def on_fatal(err):
+    os.kill(os.getpid(), signal.SIGTERM)  # Die
+```
+
+Multiple fatal handlers are supported and run in registration order.
+
 ## Migrations
 
 Adding or removing a field and other such simple operations are automatic, but when the time comes to really change your data model, implement a `migrate_v1` function that converts your old data to the new form. This works on plain built-in dict and other types, to avoid needing to preserve old versions of your structs.

__init__.py

@@ -1,16 +0,0 @@
-"""Project-root shim package for local development layout.
-
-This forwards imports to the inner `kanta/` package directory so
-`from kanta import ...` works when running tests from the workspace root.
-"""
-
-import importlib
-from pathlib import Path
-
-_inner_pkg = Path(__file__).with_name("kanta")
-if str(_inner_pkg) not in __path__:
-    __path__.append(str(_inner_pkg))
-
-_pkg = importlib.import_module(".kanta", __name__)
-__all__ = list(getattr(_pkg, "__all__", ()))
-globals().update({name: getattr(_pkg, name) for name in __all__})

docs/database.md

@@ -82,15 +82,30 @@ history.
 ## Transaction Semantics
 
 - `kanta.transaction(action=...)` captures a pre-transaction snapshot dict.
+- By default a transaction updates the modification time `m` to the current UTC
+  time.
+- `mtime=True|False|datetime` controls the modification time `m`:
+  - `True` (default) sets `m` to the current UTC time.
+  - `False` omits `m`, leaving the previous modification time in effect.
+  - A `datetime` sets `m` to that explicit value.
+- System operations such as `migrate:msgspec` use `mtime=False` so they are not
+  considered modifications and do not advance `m`.
 - On success:
   - compute diff between previous builtins and current builtins,
-  - queue a `ChangeRecord` if non-empty.
+  - queue a `ChangeRecord` if non-empty,
+  - update `kanta.mtime` when the change carries an `m` value.
 - On exception:
   - restore in-memory data from snapshot,
   - re-raise the exception.
 
 Nested transactions are rejected.
 
+## Modification Time
+
+`kanta.mtime` exposes the last modification time carried forward from change
+records. It is updated by normal transactions and preserved across snapshots and
+reloads, while system operations such as migrations leave it unchanged.
+
 ## Flush and Lifecycle
 
 - Writes are queued in memory.
@@ -99,6 +114,33 @@ Nested transactions are rejected.
 - `kanta.close()` performs final flush and releases file resources.
 - `async with Kanta(...)` guarantees open/close lifecycle management.
 
+### Open Modes
+
+- `await kanta.open()` (default) creates the database file if missing.
+- `await kanta.open(create=False)` fails when the file is missing or empty.
+
+### Bootstrap Callbacks
+
+- Bootstrap callbacks run during `open()` when the database is empty.
+- Register callbacks via:
+  - `@kanta.bootstrap`
+  - `@kanta.bootstrap(action=..., user=..., mtime=...)`
+- Bootstrap callbacks may be sync or async and receive the live root data
+  object.
+- Multiple bootstrap callbacks are supported:
+  - callbacks execute in registration order,
+  - exactly one bootstrap `ChangeRecord` is queued,
+  - bootstrap metadata (`action`, `user`, `mtime`) is taken from the last
+    callback registration.
+- If any bootstrap callback raises, Kanta closes and removes the database file,
+  then re-raises the exception.
+
+### Fatal Error Handlers
+
+- Fatal background persistence errors can be handled with `@kanta.fatal_error`.
+- Handlers may be sync or async.
+- Multiple handlers are supported and invoked in registration order.
+
 ## Migrations
 
 - Migration source is configured on `Kanta(...)` via `migrations=`.

kanta/diff.py

@@ -2,7 +2,7 @@
 
 import jsondiff
 
-from kanta.kanta.structs import ChangeRecord
+from kanta.structs import ChangeRecord
 from kanta.serialization.base import ReplayResult, replay
 from kanta.serialization.framing import LineFramer
 from kanta.serialization.json import JsonSerializer

kanta/kanta.py

@@ -1,14 +1,13 @@
 """JSONL persistence layer with background flush task."""
 
 from __future__ import annotations
-
-from collections.abc import Callable
+from datetime import datetime
 from pathlib import Path
 from types import ModuleType
 from typing import Any, Generic, TypeVar
 
 from kanta.exceptions import DatabaseError
-from kanta.kanta.kantaimpl import KantaImpl
+from kanta.kantaimpl import KantaImpl
 from kanta.serialization import JsonSerializer, Serializer
 from kanta.transaction import transaction as _transaction
 
@@ -54,7 +53,6 @@ class Kanta(Generic[T]):
         migrations: ModuleType | str | None = None,
         migration_ctx: Any | None = None,
         serializer: Serializer | None = None,
-        fatal_error: Callable[[DatabaseError], None] | None = None,
         flush_interval: float = 0.1,
     ):
         """Initialize a Kanta persistence instance.
@@ -67,8 +65,6 @@ class Kanta(Generic[T]):
             migration_ctx: Optional context object passed to migration functions.
             flush_interval: Background flush interval in seconds.
             serializer: Optional serializer implementation.
-            fatal_error: Optional callback invoked immediately when the
-                background writer encounters a DatabaseError.
 
         Raises:
             ImportError: If ``migrations`` is a string path that cannot be imported.
@@ -79,7 +75,6 @@ class Kanta(Generic[T]):
 
         self._impl = KantaImpl(
             serializer=active_serializer,
-            fatal_error=fatal_error,
             filename=filename,
             data=data,
             type=data_type,
@@ -132,19 +127,34 @@ class Kanta(Generic[T]):
         """
         return self._impl.filename
 
-    async def open(self) -> None:
+    @property
+    def mtime(self) -> datetime | None:
+        """Last modification time carried forward from change records.
+
+        Returns:
+            The latest ``m`` value, or ``None`` if no modification time has
+            been set yet. System operations such as migrations do not update
+            this value.
+        """
+        return self._impl.mtime
+
+    async def open(self, *, create: bool = True) -> None:
         """Open the database file and start background persistence.
 
         This loads existing records, applies configured migrations, and starts
         the background flush task.
 
+        Args:
+            create: Whether to create the database file when missing.
+                If False, opening fails when the file does not exist or is empty.
+
         Calling ``open`` more than once on the same instance is not allowed.
 
         Raises:
             kanta.exceptions.DatabaseError: If replay or decoding fails.
             kanta.exceptions.DataIntegrityError: If the instance is already open.
         """
-        await self._impl.open()
+        await self._impl.open(create=create)
 
     async def __aenter__(self) -> Kanta[T]:
         """Enter async context manager and open the database.
@@ -177,6 +187,53 @@ class Kanta(Generic[T]):
         """Stop background task, flush pending changes, and close file lock."""
         await self._impl.close()
 
+    def bootstrap(
+        self,
+        fn=None,
+        *,
+        action: str = "bootstrap",
+        user: str | None = None,
+        mtime: bool | datetime = True,
+    ):
+        """Register a bootstrap callback executed during :meth:`open`.
+
+        Can be used as ``@kanta.bootstrap`` or ``@kanta.bootstrap(...)``.
+        The callback receives the live ``data`` object and may be sync or async.
+        """
+
+        def _register(callback):
+            if not callable(callback):
+                raise TypeError("bootstrap callback must be callable")
+            self._impl.add_bootstrap(
+                callback=callback,
+                action=action,
+                user=user,
+                mtime=mtime,
+            )
+            return callback
+
+        if fn is None:
+            return _register
+        return _register(fn)
+
+    def fatal_error(self, fn=None):
+        """Register fatal error handler callback.
+
+        Can be used as ``@kanta.fatal_error``.
+        The callback receives a :class:`kanta.exceptions.DatabaseError` and may
+        be sync or async.
+        """
+
+        def _register(callback):
+            if not callable(callback):
+                raise TypeError("fatal error callback must be callable")
+            self._impl.add_fatal_error(callback)
+            return callback
+
+        if fn is None:
+            return _register
+        return _register(fn)
+
     def transaction(
         self,
         action: str,
@@ -184,6 +241,7 @@ class Kanta(Generic[T]):
         user: str | None = None,
         user_display: str | None = None,
         resolver: Any = None,
+        mtime: bool | datetime = True,
     ):
         """Create a transactional mutation context manager.
 
@@ -192,6 +250,12 @@ class Kanta(Generic[T]):
             user: Optional user identifier stored in metadata.
             user_display: Optional display name used for logging/resolution.
             resolver: Optional callable for resolving identifiers in logs.
+            mtime: Controls the modification time ``m``. ``True`` (default)
+                sets ``m`` to the current UTC time. ``False`` omits ``m`` so the
+                previous modification time remains in effect; this is used for
+                system operations that are not considered modifications. A
+                :class:`~datetime.datetime` value sets ``m`` to that explicit
+                time.
 
         Returns:
             A context manager yielding the live state object for mutation.
@@ -202,5 +266,10 @@ class Kanta(Generic[T]):
             rolled back.
         """
         return _transaction(
-            self._impl, action, user=user, user_display=user_display, resolver=resolver
+            self._impl,
+            action,
+            user=user,
+            user_display=user_display,
+            resolver=resolver,
+            mtime=mtime,
         )

kanta/kantaimpl.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import asyncio
 import copy
 import importlib
+import inspect
 import logging
 from datetime import UTC, datetime
 from typing import Any, Generic, TypeVar
@@ -41,13 +42,31 @@ class KantaImpl(PersistenceMixin, Generic[T]):
         self.in_transaction = False
         self.transaction_snapshot: dict[str, Any] | None = None
         self.opened = False
+        self.bootstrap_callbacks: list[Any] = []
+        self.bootstrap_action = "bootstrap"
+        self.bootstrap_user: str | None = None
+        self.bootstrap_mtime: bool | datetime = True
 
         self.statedict = struct_to_dict(self.data, serializer=self.serializer)
         self.version = (
             self.migration_registry.dbver if self.migration_registry is not None else 0
         )
 
-    async def open(self) -> None:
+    def add_bootstrap(
+        self,
+        *,
+        callback,
+        action: str,
+        user: str | None,
+        mtime: bool | datetime,
+    ) -> None:
+        """Add bootstrap callback and update bootstrap metadata."""
+        self.bootstrap_callbacks.append(callback)
+        self.bootstrap_action = action
+        self.bootstrap_user = user
+        self.bootstrap_mtime = mtime
+
+    async def open(self, *, create: bool = True) -> None:
         """Open the database: load from disk, apply migrations, start background task."""
         if self.opened:
             raise DataIntegrityError(
@@ -56,12 +75,27 @@ class KantaImpl(PersistenceMixin, Generic[T]):
                 action="open",
             )
 
+        existed_before_open = self.filename.exists()
+
         content = await asyncio.to_thread(
             self.file.open_and_read,
             self.filename,
-            create=True,
+            create=create,
         )
 
+        if not create and (not existed_before_open or not content):
+            self.file.close()
+            reason = (
+                "database file did not exist"
+                if not existed_before_open
+                else "database file is empty"
+            )
+            raise DataIntegrityError(
+                f"Cannot open database: {reason}",
+                db_path=self.filename,
+                action="open",
+            )
+
         if content:
             try:
                 rr = replay(
@@ -104,13 +138,35 @@ class KantaImpl(PersistenceMixin, Generic[T]):
                 serializer=self.serializer,
             )
             self.version = rr.version
+            self.mtime = rr.m
             normalized = struct_to_dict(self.data, serializer=self.serializer)
-            self.queue_change("migrate:msgspec", normalized)
+            self.queue_change("migrate:msgspec", normalized, mtime=False)
             self.snapshot.ts = (
                 datetime.fromtimestamp(rr.last_snapshot_mtime, UTC)
                 if rr.last_snapshot_mtime is not None
                 else None
             )
+        elif self.bootstrap_callbacks:
+            try:
+                for callback in self.bootstrap_callbacks:
+                    callback_result = callback(self.data)
+                    if inspect.isawaitable(callback_result):
+                        await callback_result
+
+                current = struct_to_dict(self.data, serializer=self.serializer)
+                self.queue_change(
+                    self.bootstrap_action,
+                    current,
+                    user=self.bootstrap_user,
+                    mtime=self.bootstrap_mtime,
+                )
+            except Exception:
+                self.file.close()
+                try:
+                    await asyncio.to_thread(self.filename.unlink, missing_ok=True)
+                except FileNotFoundError:
+                    pass
+                raise
 
         self.opened = True
 

kanta/persistence.py

@@ -4,18 +4,18 @@ from __future__ import annotations
 
 import asyncio
 import copy
+import inspect
 import logging
-import threading
 from collections import deque
 from collections.abc import Callable
-from datetime import datetime
+from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
 
 from kanta.diff import compute_diff
 from kanta.exceptions import DatabaseError, DataIntegrityError
 from kanta.filelock import LockedFile
-from kanta.kanta.structs import ChangeRecord
+from kanta.structs import ChangeRecord
 from kanta.serialization import JsonSerializer, Serializer
 from kanta.serialization.framing import Framer
 from kanta.snapshot import SnapshotState
@@ -31,38 +31,41 @@ class PersistenceMixin:
     flush_failed: bool
     statedict: dict[str, Any]
     pending_changes: deque[ChangeRecord]
-    pending_lock: threading.Lock
     snapshot: SnapshotState
     serializer: Serializer
     framer: Framer
     background_task: asyncio.Task | None
-    fatal_error: Callable[[DatabaseError], None] | None
+    fatal_error_handlers: list[Callable[[DatabaseError], Any]]
     background_error: DatabaseError | None
     flush_interval: float
     version: int
     opened: bool
+    mtime: datetime | None
 
     def __init__(self, **kwargs: Any) -> None:
         """Initialize persistence-owned state used by mixin methods."""
         filename = kwargs.pop("filename")
         flush_interval = kwargs.pop("flush_interval", 0.1)
         serializer = kwargs.pop("serializer", None)
-        fatal_error = kwargs.pop("fatal_error", None)
         super().__init__(**kwargs)
         self.filename = Path(filename)
         self.file = LockedFile()
         self.flush_failed = False
         self.statedict = {}
         self.pending_changes = deque()
-        self.pending_lock = threading.Lock()
         self.serializer = serializer or JsonSerializer()
         self.framer = self.serializer.framer_cls()
         self.snapshot = SnapshotState(serializer=self.serializer, framer=self.framer)
         self.background_task = None
-        self.fatal_error = fatal_error
+        self.fatal_error_handlers = []
         self.background_error = None
         self.flush_interval = flush_interval
         self.version = 0
+        self.mtime: datetime | None = None
+
+    def add_fatal_error(self, callback: Callable[[DatabaseError], Any]) -> None:
+        """Register one fatal error callback in call order."""
+        self.fatal_error_handlers.append(callback)
 
     async def _background_loop(self) -> None:
         """Background task that periodically flushes changes to disk."""
@@ -77,9 +80,11 @@ class PersistenceMixin:
                 break
             except DatabaseError as e:
                 self.background_error = e
-                if self.fatal_error is not None:
+                for callback in self.fatal_error_handlers:
                     try:
-                        self.fatal_error(e)
+                        callback_result = callback(e)
+                        if inspect.isawaitable(callback_result):
+                            await callback_result
                     except Exception as callback_error:
                         _logger.exception(
                             "Background error callback failed: %s", callback_error
@@ -89,30 +94,59 @@ class PersistenceMixin:
 
     def maybe_snapshot(self) -> None:
         """Evaluate and possibly write a snapshot from current state."""
-        self.snapshot.maybe_write(self.file, self.version, self.statedict)
+        self.snapshot.maybe_write(self.file, self.version, self.statedict, m=self.mtime)
 
     def queue_change(
         self,
         action: str,
         current: dict,
+        *,
         user: str | None = None,
-        m: datetime | None = None,
-    ) -> None:
-        """Queue a change record internally (thread-safe)."""
+        mtime: bool | datetime = True,
+    ) -> ChangeRecord | None:
+        """Queue a change record internally (thread-safe).
+
+        Args:
+            action: Action label stored in the change record.
+            current: New serialized state after the change.
+            user: Optional actor identifier.
+            mtime: Controls the modification timestamp. ``True`` (default)
+                sets ``m`` to the current UTC time. ``False`` omits ``m`` so the
+                previous modification time remains in effect; this is used for
+                system operations that are not considered modifications. A
+                :class:`~datetime.datetime` value sets ``m`` to that explicit time.
+
+        Returns:
+            The queued :class:`ChangeRecord`, or ``None`` if the diff was empty.
+        """
+        now = datetime.now(UTC)
+
+        if mtime is True:
+            m = now
+        elif mtime is False:
+            m = None
+        elif isinstance(mtime, datetime):
+            m = mtime
+        else:
+            raise TypeError("mtime must be True, False, or a datetime")
+
         diff = compute_diff(self.statedict, current)
         if not diff:
-            return
-        with self.pending_lock:
-            self.pending_changes.append(
-                ChangeRecord(
-                    a=action,
-                    v=self.version,
-                    u=user,
-                    m=m,
-                    diff=diff,
-                )
-            )
+            return None
+
+        record = ChangeRecord(
+            ts=now,
+            a=action,
+            v=self.version,
+            u=user,
+            m=m,
+            diff=diff,
+        )
+        self.pending_changes.append(record)
         self.statedict = copy.deepcopy(current)
+        if m is not None:
+            self.mtime = m
+        return record
 
     def flush_sync(self) -> None:
         """Synchronously flush all pending changes to disk."""
@@ -126,10 +160,9 @@ class PersistenceMixin:
         if self.flush_failed:
             return
 
-        with self.pending_lock:
-            if not self.pending_changes:
-                return
-            changes_to_write = list(self.pending_changes)
+        if not self.pending_changes:
+            return
+        changes_to_write = list(self.pending_changes)
 
         if not self.file.is_open:
             self.file.open(self.filename, create=True)
@@ -146,15 +179,13 @@ class PersistenceMixin:
                 records.append(framed)
                 running_size += len(framed)
             if not records:
-                with self.pending_lock:
-                    self.pending_changes.clear()
+                self.pending_changes.clear()
                 return
 
             self.file.write(b"".join(records))
             self.snapshot.record_changes(len(records))
-            with self.pending_lock:
-                for _ in changes_to_write:
-                    self.pending_changes.popleft()
+            for _ in changes_to_write:
+                self.pending_changes.popleft()
         except OSError as e:
             _logger.error("Failed to flush database: %s", e)
             self.flush_failed = True
@@ -176,10 +207,9 @@ class PersistenceMixin:
         if self.flush_failed:
             return
 
-        with self.pending_lock:
-            if not self.pending_changes:
-                return
-            changes_to_write = list(self.pending_changes)
+        if not self.pending_changes:
+            return
+        changes_to_write = list(self.pending_changes)
 
         if not self.file.is_open:
             await asyncio.to_thread(self.file.open, self.filename, create=True)
@@ -196,15 +226,13 @@ class PersistenceMixin:
                 records.append(framed)
                 running_size += len(framed)
             if not records:
-                with self.pending_lock:
-                    self.pending_changes.clear()
+                self.pending_changes.clear()
                 return
 
             await asyncio.to_thread(self.file.write, b"".join(records))
             self.snapshot.record_changes(len(records))
-            with self.pending_lock:
-                for _ in changes_to_write:
-                    self.pending_changes.popleft()
+            for _ in changes_to_write:
+                self.pending_changes.popleft()
         except OSError as e:
             _logger.error("Failed to flush database: %s", e)
             self.flush_failed = True

kanta/serialization/base.py

@@ -9,7 +9,7 @@ from typing import Any, Protocol, TypeVar
 import msgspec
 
 from kanta.exceptions import ReplayError
-from kanta.kanta.structs import ChangeRecord, Snapshot
+from kanta.structs import ChangeRecord, Snapshot
 from kanta.serialization.framing import Framer
 
 T = TypeVar("T")
@@ -23,14 +23,12 @@ class ReplayResult:
         state: dict[str, Any],
         version: int = 0,
         has_migration: bool = False,
-        last_patch_mtime: float | None = None,
         last_snapshot_mtime: float | None = None,
         m: datetime | None = None,
     ):
         self.state = state
         self.version = version
         self.has_migration = has_migration
-        self.last_patch_mtime = last_patch_mtime
         self.last_snapshot_mtime = last_snapshot_mtime
         self.m = m
 
@@ -63,7 +61,6 @@ def replay(
     last_snapshot_mtime: float | None = None
     m: datetime | None = None
     has_migration = False
-    last_patch_mtime: float | None = None
 
     if snap_payload is not None:
         try:
@@ -112,14 +109,12 @@ def replay(
             has_migration = True
         if change.m is not None:
             m = change.m
-        last_patch_mtime = change.ts.timestamp()
         version = change.v
         state = _patch_state(state, change.diff)
     return ReplayResult(
         state=state,
         version=version,
         has_migration=has_migration,
-        last_patch_mtime=last_patch_mtime,
         last_snapshot_mtime=last_snapshot_mtime,
         m=m,
     )

kanta/serialization/json.py

@@ -6,7 +6,7 @@ from typing import Any, TypeVar
 
 import msgspec
 
-from kanta.serialization.framing import LineFramer
+from kanta.serialization.framing import Framer, LineFramer
 
 T = TypeVar("T")
 
@@ -14,7 +14,7 @@ T = TypeVar("T")
 class JsonSerializer:
     """Line-based JSON serializer."""
 
-    framer_cls = LineFramer
+    framer_cls: type[Framer] = LineFramer
 
     def encode(self, obj: Any) -> bytes:
         return msgspec.json.encode(obj)

kanta/serialization/msgpack.py

@@ -6,7 +6,7 @@ from typing import Any, TypeVar
 
 import msgspec
 
-from kanta.serialization.framing import BinFramer
+from kanta.serialization.framing import BinFramer, Framer
 
 T = TypeVar("T")
 
@@ -14,7 +14,7 @@ T = TypeVar("T")
 class MsgPackSerializer:
     """Binary serializer using MessagePack format."""
 
-    framer_cls = BinFramer
+    framer_cls: type[Framer] = BinFramer
 
     def encode(self, obj: Any) -> bytes:
         return msgspec.msgpack.encode(obj)

kanta/snapshot.py

@@ -5,7 +5,7 @@ from __future__ import annotations
 import logging
 from datetime import UTC, datetime
 
-from kanta.kanta.structs import Snapshot
+from kanta.structs import Snapshot
 from kanta.serialization import JsonSerializer, Serializer
 from kanta.serialization.framing import Framer, LineFramer
 
@@ -37,7 +37,9 @@ class SnapshotState:
     def record_changes(self, count: int) -> None:
         self.changes += count
 
-    def maybe_write(self, file, version: int, state: dict) -> None:
+    def maybe_write(
+        self, file, version: int, state: dict, m: datetime | None = None
+    ) -> None:
         """Write snapshot when thresholds/time policy allows it."""
         if self.changes < self._min_diffs:
             return
@@ -51,14 +53,16 @@ class SnapshotState:
         if not file.is_open:
             return
         try:
-            self._write(file, version, state, now)
+            self._write(file, version, state, now, m=m)
             self._force_pending = False
         except Exception as exc:
             _logger.error("snapshot: failed to write snapshot: %r", exc)
 
-    def _write(self, file, version: int, state: dict, now: datetime) -> None:
+    def _write(
+        self, file, version: int, state: dict, now: datetime, m: datetime | None = None
+    ) -> None:
         """Write a snapshot and update internal state."""
-        payload = self._serializer.encode(Snapshot(ts=now, v=version, state=state))
+        payload = self._serializer.encode(Snapshot(ts=now, v=version, state=state, m=m))
         record_offset = file.size() if hasattr(file, "size") else 0
         file.write(self._framer.frame_snapshot(payload, record_offset=record_offset))
         self.changes = 0

kanta/transaction.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import logging
 from contextlib import contextmanager
+from datetime import datetime
 from typing import Any
 
 from kanta.diff import compute_diff
@@ -22,6 +23,7 @@ def transaction(
     user: str | None = None,
     user_display: str | None = None,
     resolver: Any = None,
+    mtime: bool | datetime = True,
 ):
     """Wrap writes in a transaction and yield the live db object."""
     if impl.in_transaction:
@@ -58,9 +60,10 @@ def transaction(
         new_dict = struct_to_dict(impl.data, serializer=impl.serializer)
         diff = compute_diff(impl.statedict, new_dict)
         if diff:
-            impl.queue_change(action, new_dict, user=user)
-            log_change(action, diff, user_display, impl.statedict, resolver)
-            impl.statedict = new_dict
+            previous = impl.statedict
+            record = impl.queue_change(action, new_dict, user=user, mtime=mtime)
+            if record is not None:
+                log_change(action, record.diff, user_display, previous, resolver)
     except Exception:
         _logger.warning("Transaction '%s' failed, rolling back changes", action)
         if impl.transaction_snapshot is not None:

tests/test_kanta_integration.py

@@ -100,6 +100,202 @@ async def test_bootstrap_creates_file(tmp_path, format_config):
     assert path.exists()
 
 
+@pytest.mark.asyncio
+async def test_bootstrap_decorator_with_args(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    @kanta.bootstrap(action="seed_init", user="system")
+    def seed(data):
+        data.counter = 3
+
+    await kanta.open()
+    await kanta.close()
+
+    assert change_actions(path, format_config) == ["seed_init"]
+
+
+@pytest.mark.asyncio
+async def test_bootstrap_decorator_without_args(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    @kanta.bootstrap
+    def seed(data):
+        data.counter = 4
+
+    await kanta.open()
+    await kanta.close()
+
+    assert change_actions(path, format_config) == ["bootstrap"]
+
+
+@pytest.mark.asyncio
+async def test_bootstrap_decorator_async(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    @kanta.bootstrap(action="async_seed")
+    async def seed(data):
+        await asyncio.sleep(0)
+        data.counter = 5
+
+    await kanta.open()
+    await kanta.close()
+
+    assert change_actions(path, format_config) == ["async_seed"]
+
+
+@pytest.mark.asyncio
+async def test_bootstrap_decorator_multiple_handlers_in_order(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    @kanta.bootstrap(action="boot_1")
+    def seed_one(data):
+        data.counter = 1
+
+    @kanta.bootstrap(action="boot_2")
+    async def seed_two(data):
+        await asyncio.sleep(0)
+        data.counter = 2
+
+    await kanta.open()
+    await kanta.close()
+
+    assert change_actions(path, format_config) == ["boot_2"]
+
+
+@pytest.mark.asyncio
+async def test_bootstrap_failure_removes_database_file(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    @kanta.bootstrap(action="boot_fail")
+    def seed_fail(data):
+        data.counter = 10
+        raise RuntimeError("bootstrap failed")
+
+    with pytest.raises(RuntimeError, match="bootstrap failed"):
+        await kanta.open()
+
+    assert not path.exists()
+
+
+@pytest.mark.asyncio
+async def test_bootstrap_async_failure_removes_database_file(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    @kanta.bootstrap(action="boot_fail_async")
+    async def seed_fail(data):
+        await asyncio.sleep(0)
+        data.counter = 10
+        raise RuntimeError("bootstrap async failed")
+
+    with pytest.raises(RuntimeError, match="bootstrap async failed"):
+        await kanta.open()
+
+    assert not path.exists()
+
+
+@pytest.mark.asyncio
+async def test_open_create_false_missing_file_fails(tmp_path, format_config):
+    path = tmp_path / "missing.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    with pytest.raises(FileLockError):
+        await kanta.open(create=False)
+
+
+@pytest.mark.asyncio
+async def test_open_create_false_empty_file_fails(tmp_path, format_config):
+    path = tmp_path / "empty.db"
+    path.touch()
+    kanta = make_kanta(path, Data, format_config)
+
+    with pytest.raises(DataIntegrityError, match="empty"):
+        await kanta.open(create=False)
+
+
+@pytest.mark.asyncio
+async def test_background_write_failure_notifies_decorator_callback(
+    tmp_path, format_config, monkeypatch
+):
+    path = tmp_path / "test.db"
+    errors: list[DatabaseError] = []
+    signaled = asyncio.Event()
+
+    kanta = make_kanta(
+        path,
+        Data,
+        format_config,
+        flush_interval=0.01,
+    )
+
+    @kanta.fatal_error
+    async def on_fatal_error(err: DatabaseError) -> None:
+        errors.append(err)
+        signaled.set()
+
+    await kanta.open()
+
+    with kanta.transaction(action="inc") as data:
+        data.counter = 1
+
+    def fail_write(_data: bytes) -> None:
+        raise OSError("simulated background write failure")
+
+    monkeypatch.setattr(kanta._impl.file, "write", fail_write)
+
+    await asyncio.wait_for(signaled.wait(), timeout=1.0)
+    assert errors
+    assert "Failed to flush database" in str(errors[0])
+
+    await kanta.close()
+
+
+@pytest.mark.asyncio
+async def test_background_write_failure_notifies_multiple_callbacks_in_order(
+    tmp_path, format_config, monkeypatch
+):
+    path = tmp_path / "test.db"
+    calls: list[str] = []
+    signaled = asyncio.Event()
+
+    kanta = make_kanta(
+        path,
+        Data,
+        format_config,
+        flush_interval=0.01,
+    )
+
+    @kanta.fatal_error
+    def on_fatal_error_sync(err: DatabaseError) -> None:
+        calls.append("sync")
+
+    @kanta.fatal_error
+    async def on_fatal_error_async(err: DatabaseError) -> None:
+        await asyncio.sleep(0)
+        calls.append("async")
+        signaled.set()
+
+    await kanta.open()
+
+    with kanta.transaction(action="inc") as data:
+        data.counter = 1
+
+    def fail_write(_data: bytes) -> None:
+        raise OSError("simulated background write failure")
+
+    monkeypatch.setattr(kanta._impl.file, "write", fail_write)
+
+    await asyncio.wait_for(signaled.wait(), timeout=1.0)
+    assert calls == ["sync", "async"]
+
+    await kanta.close()
+
+
 @pytest.mark.asyncio
 async def test_snapshot(tmp_path, format_config):
     path = tmp_path / "test.db"
@@ -281,17 +477,18 @@ async def test_background_write_failure_notifies_callback(
     errors: list[DatabaseError] = []
     signaled = asyncio.Event()
 
-    def on_fatal_error(err: DatabaseError) -> None:
-        errors.append(err)
-        signaled.set()
-
     kanta = make_kanta(
         path,
         Data,
         format_config,
         flush_interval=0.01,
-        fatal_error=on_fatal_error,
     )
+
+    @kanta.fatal_error
+    def on_fatal_error(err: DatabaseError) -> None:
+        errors.append(err)
+        signaled.set()
+
     await kanta.open()
 
     with kanta.transaction(action="inc") as data:

tests/test_mtime.py

@@ -0,0 +1,143 @@
+"""Tests for mtime handling and the public ``kanta.mtime`` property."""
+
+from datetime import UTC, datetime
+
+import pytest
+
+from kanta import ChangeRecord
+
+from .support import Data, make_kanta, seed_single_change
+
+
+def _read_last_change(path, format_config):
+    name, serializer_cls = format_config
+    serializer = serializer_cls()
+    framer = serializer.framer_cls()
+    last = None
+    for is_snapshot, payload, _, _ in framer.iter_records(path.read_bytes(), 0):
+        if is_snapshot:
+            continue
+        last = serializer.decode(payload, type=ChangeRecord)
+    assert last is not None
+    return last
+
+
+@pytest.mark.asyncio
+async def test_default_transaction_updates_mtime(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open()
+
+    before = datetime.now(UTC)
+    with kanta.transaction(action="inc") as data:
+        data.counter = 1
+    await kanta.flush()
+    await kanta.close()
+
+    rec = _read_last_change(path, format_config)
+    assert rec.ts == rec.m
+    assert before <= rec.m <= datetime.now(UTC)
+    assert kanta.mtime == rec.m
+
+
+@pytest.mark.asyncio
+async def test_transaction_custom_mtime(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open()
+
+    custom_m = datetime(2026, 1, 1, 8, 0, tzinfo=UTC)
+    with kanta.transaction(action="inc", mtime=custom_m) as data:
+        data.counter = 1
+    await kanta.flush()
+    await kanta.close()
+
+    rec = _read_last_change(path, format_config)
+    assert rec.m == custom_m
+    assert kanta.mtime == custom_m
+
+
+@pytest.mark.asyncio
+async def test_transaction_mtime_false_preserves_mtime(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open()
+
+    first_m = datetime(2026, 1, 1, 10, 0, tzinfo=UTC)
+    with kanta.transaction(action="first", mtime=first_m) as data:
+        data.counter = 1
+
+    with kanta.transaction(action="second", mtime=False) as data:
+        data.counter = 2
+
+    await kanta.flush()
+    await kanta.close()
+
+    records = []
+    name, serializer_cls = format_config
+    serializer = serializer_cls()
+    framer = serializer.framer_cls()
+    for is_snapshot, payload, _, _ in framer.iter_records(path.read_bytes(), 0):
+        if is_snapshot:
+            continue
+        records.append(serializer.decode(payload, type=ChangeRecord))
+
+    assert records[0].m == first_m
+    assert records[1].m is None
+    assert kanta.mtime == first_m
+
+
+@pytest.mark.asyncio
+async def test_migration_does_not_update_mtime(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    seed_m = datetime(2025, 12, 31, 23, 0, tzinfo=UTC)
+    seed_single_change(
+        path,
+        ChangeRecord(
+            ts=seed_m,
+            m=seed_m,
+            a="seed",
+            v=0,
+            diff={"counter": 0},
+        ),
+        format_config,
+    )
+
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open()
+
+    assert kanta.mtime == seed_m
+
+    new_m = datetime(2026, 1, 5, 10, 0, tzinfo=UTC)
+    with kanta.transaction(action="inc", mtime=new_m) as data:
+        data.counter = 5
+    await kanta.flush()
+
+    assert kanta.mtime == new_m
+    await kanta.close()
+
+
+@pytest.mark.asyncio
+async def test_rollback_does_not_update_mtime(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open()
+
+    seed_m = datetime(2026, 1, 1, 10, 0, tzinfo=UTC)
+    with kanta.transaction(action="seed", mtime=seed_m) as data:
+        data.counter = 1
+
+    before = kanta.mtime
+
+    try:
+        with kanta.transaction(
+            action="boom", mtime=datetime(2099, 1, 1, tzinfo=UTC)
+        ) as data:
+            data.counter = 99
+            raise RuntimeError("fail")
+    except RuntimeError:
+        pass
+
+    assert kanta.data.counter == 1
+    assert kanta.mtime == before
+    await kanta.close()