Stricter versioning: always store migrate version record, file must be within versions included in migrations. Log only for migrations that made changes. Allow deleting older migration functions when no longer required.

README.md

@@ -53,10 +53,13 @@ asyncio.run(main())
 
 ## Bootstrap and Open Modes
 
-Kanta supports open-time bootstrap callbacks for initializing a brand-new
-database before `open()` returns.
+When `open()` creates a brand-new database, it always writes a single bootstrap
+change record from the initial data object you passed to `Kanta(...)`. The
+simplest bootstrap is therefore the object itself — no extra code is required.
 
-Register bootstrap handlers with a decorator:
+Bootstrap handlers are optional. Use them only when you need to modify the
+initial state at creation time, for example to seed defaults or perform
+expensive/external setup that should happen exactly once:
 
 ```python
 kanta = Kanta("data.kantadb", Data())
@@ -76,9 +79,10 @@ async def bootstrap_async(data) -> None:
     data.counter = 1
 ```
 
-When multiple bootstrap handlers are registered:
+Whether or not handlers are registered, exactly one bootstrap change record is
+written when a new database is created. The record contains the initial object,
+or the state after all bootstrap handlers have run. When handlers are present:
 - they run in registration order,
-- exactly one bootstrap change record is queued,
 - bootstrap metadata (`action`, `user`, `mtime`) is taken from the last
   registration.
 
@@ -94,6 +98,18 @@ await kanta.open(create=False)
 With `create=False`, open fails if the database file does not exist or is
 empty.
 
+Read-only mode opens an existing database without locking it or starting the
+background flush task. This is useful for readers that must not block the
+writer or modify the file:
+
+```python
+await kanta.open(readonly=True)
+```
+
+In read-only mode, records are replayed and migrations are applied in memory,
+but transactions and explicit flushes are rejected and the file is never
+created if missing.
+
 ## Fatal Error Handlers
 
 Fatal background write errors can be observed with a decorator:

docs/database.md

@@ -118,6 +118,12 @@ reloads, while system operations such as migrations leave it unchanged.
 
 - `await kanta.open()` (default) creates the database file if missing.
 - `await kanta.open(create=False)` fails when the file is missing or empty.
+- `await kanta.open(readonly=True)` opens an existing database read-only.
+  - The file is opened without acquiring a lock and without a background flush
+    task.
+  - Existing records are replayed and migrations are still applied in memory.
+  - Transactions and explicit flushes are rejected.
+  - The file is never created if missing.
 
 ### Callbacks
 
@@ -127,7 +133,11 @@ when they have a default value.
 
 #### Bootstrap Callbacks
 
-- Bootstrap callbacks run during `open()` when the database is empty.
+- When `open()` creates a new database, it always writes a single bootstrap
+  `ChangeRecord`.
+- The simplest bootstrap is the initial data object passed to `Kanta(...)`;
+  bootstrap callbacks are optional and only needed when you want to modify or
+  enrich that object at creation time.
 - Register callbacks via:
   - `@kanta.bootstrap`
   - `@kanta.bootstrap(action=..., user=..., mtime=...)`
@@ -140,6 +150,8 @@ when they have a default value.
   - exactly one bootstrap `ChangeRecord` is queued,
   - bootstrap metadata (`action`, `user`, `mtime`) is taken from the last
     callback registration.
+- If no bootstrap callbacks are registered, the bootstrap record still uses
+  `action="bootstrap"` and contains the initial data object.
 - If any bootstrap callback raises, Kanta closes and removes the database file,
   then re-raises the exception.
 

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 types
 from collections.abc import Callable
 from dataclasses import dataclass
 from typing import Annotated, Any, Union, get_args, get_origin
@@ -317,11 +318,7 @@ class CallbackRegistry:
                     f"Allowed: str path, {self._allowed_message('logfmt')}"
                 )
 
-        if sig.return_annotation is inspect.Signature.empty:
-            raise TypeError(
-                f"logfmt callback {callback.__name__} must annotate its "
-                f"return type as str | None"
-            )
+        if sig.return_annotation is not inspect.Signature.empty:
             return_ann = self._resolve_raw_annotation(sig.return_annotation, callback)
             if not self._is_optional_str(return_ann):
                 raise TypeError(
@@ -416,11 +413,7 @@ class CallbackRegistry:
                 f"logfmt class {cls.__name__}.resolve must accept a 'path: str' parameter"
             )
 
-        if resolve_sig.return_annotation is inspect.Signature.empty:
-            raise TypeError(
-                f"logfmt class {cls.__name__}.resolve must annotate its "
-                f"return type as str | None"
-            )
+        if resolve_sig.return_annotation is not inspect.Signature.empty:
             return_ann = self._resolve_raw_annotation(
                 resolve_sig.return_annotation, resolve
             )
@@ -516,7 +509,7 @@ class CallbackRegistry:
     @staticmethod
     def _unwrap_optional(ann: Any) -> Any:
         origin = get_origin(ann)
-        if origin is not Union:
+        if origin not in (Union, types.UnionType):
             return ann
         args = [arg for arg in get_args(ann) if arg is not type(None)]
         return args[0] if len(args) == 1 else ann
@@ -524,7 +517,7 @@ class CallbackRegistry:
     @staticmethod
     def _is_optional_str(ann: Any) -> bool:
         origin = get_origin(ann)
-        if origin is not Union:
+        if origin not in (Union, types.UnionType):
             return ann is str
         args = get_args(ann)
         return type(None) in args and any(arg is str for arg in args)

kanta/filelock.py

@@ -34,6 +34,7 @@ if sys.platform == "win32":
     _GENERIC_READ = 0x80000000
     _GENERIC_WRITE = 0x40000000
     _FILE_SHARE_READ = 0x00000001
+    _FILE_SHARE_WRITE = 0x00000002
     _OPEN_EXISTING = 3
     _OPEN_ALWAYS = 4
     _FILE_ATTRIBUTE_NORMAL = 0x80
@@ -91,12 +92,13 @@ else:
 
 
 class LockedFile:
-    """A file opened with an exclusive write lock.
+    """A file opened for read+write with an optional exclusive lock.
 
     Usage::
 
         f = LockedFile()
         f.open(path)                    # open + lock (read+write)
+        f.open(path, readonly=True)     # open read-only without locking
         content = f.read()              # read entire content
         f.write(data)                   # append data (seeks to end first)
         f.close()                       # release lock + close fd
@@ -108,12 +110,13 @@ class LockedFile:
     def __init__(self) -> None:
         self._fd: int | None = None  # Unix fd or Windows HANDLE
 
-    def open(self, path: Path, *, create: bool = False) -> None:
-        """Open *path* for read+write with an exclusive lock.
+    def open(self, path: Path, *, create: bool = False, readonly: bool = False) -> None:
+        """Open *path* and optionally acquire an exclusive lock.
 
         Args:
             path:   File to open and lock.
             create: If True, create the file if it doesn't exist (bootstrap).
+            readonly: If True, open read-only without acquiring a lock.
 
         Raises:
             FileLockError: If the file is locked by another process or not found.
@@ -122,16 +125,18 @@ class LockedFile:
             return  # Already open (idempotent)
 
         if sys.platform == "win32":
-            self._open_win32(path, create)
+            self._open_win32(path, create, readonly)
         else:
-            self._open_unix(path, create)
+            self._open_unix(path, create, readonly)
 
-    def open_and_read(self, path: Path, create: bool = False) -> bytes:
-        """Open *path* with exclusive lock and read all content.
+    def open_and_read(
+        self, path: Path, create: bool = False, readonly: bool = False
+    ) -> bytes:
+        """Open *path* and read all content.
 
         Combined operation for efficient use with asyncio.to_thread().
         """
-        self.open(path, create=create)
+        self.open(path, create=create, readonly=readonly)
         return self.read()
 
     def read(self) -> bytes:
@@ -188,12 +193,16 @@ class LockedFile:
 
     # -- Unix ----------------------------------------------------------------
 
-    def _open_unix(self, path: Path, create: bool) -> None:
+    def _open_unix(self, path: Path, create: bool, readonly: bool) -> None:
+        if readonly:
+            flags = os.O_RDONLY
+        else:
             flags = os.O_RDWR | (os.O_CREAT if create else 0)
         try:
             fd = os.open(path, flags, 0o666)
         except FileNotFoundError:
             _fatal(f"Database file not found: {path.resolve()}", db_path=path)
+        if not readonly:
             try:
                 fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
             except OSError:
@@ -220,12 +229,19 @@ class LockedFile:
 
     # -- Windows -------------------------------------------------------------
 
-    def _open_win32(self, path: Path, create: bool) -> None:
+    def _open_win32(self, path: Path, create: bool, readonly: bool) -> None:
+        if readonly:
+            disposition = _OPEN_EXISTING
+            access = _GENERIC_READ
+            share = _FILE_SHARE_READ | _FILE_SHARE_WRITE
+        else:
             disposition = _OPEN_ALWAYS if create else _OPEN_EXISTING
+            access = _GENERIC_READ | _GENERIC_WRITE
+            share = _FILE_SHARE_READ
         handle = _kernel32.CreateFileW(
             str(path),
-            _GENERIC_READ | _GENERIC_WRITE,
-            _FILE_SHARE_READ,
+            access,
+            share,
             None,
             disposition,
             _FILE_ATTRIBUTE_NORMAL,

kanta/kanta.py

@@ -3,8 +3,8 @@
 from __future__ import annotations
 from datetime import datetime
 from pathlib import Path
-from types import ModuleType
-from typing import Any, Generic, TypeVar
+from types import ModuleType, SimpleNamespace
+from typing import Generic, TypeVar
 
 from kanta.kantaimpl import KantaImpl
 from kanta.serialization import JsonSerializer, Serializer
@@ -50,7 +50,6 @@ class Kanta(Generic[T]):
         *,
         type: type[T] | None = None,
         migrations: ModuleType | str | None = None,
-        migration_ctx: Any | None = None,
         serializer: Serializer | None = None,
         flush_interval: float = 0.1,
     ):
@@ -61,7 +60,6 @@ class Kanta(Generic[T]):
             data: Caller-owned root msgspec.Struct state instance.
             type: Optional explicit root type. Defaults to ``type(data)``.
             migrations: Optional migrations module object or import path.
-            migration_ctx: Optional context object passed to migration functions.
             flush_interval: Background flush interval in seconds.
             serializer: Optional serializer implementation.
 
@@ -78,7 +76,6 @@ class Kanta(Generic[T]):
             data=data,
             type=data_type,
             migrations=migrations,
-            migration_ctx=migration_ctx,
             flush_interval=flush_interval,
             kanta=self,
         )
@@ -127,6 +124,15 @@ class Kanta(Generic[T]):
         """
         return self._impl.filename
 
+    @property
+    def ctx(self) -> SimpleNamespace:
+        """User-writable context namespace.
+
+        Migration functions receive the ``Kanta`` instance and can read or
+        mutate ``kanta.ctx`` during migrations.
+        """
+        return self._impl.ctx
+
     @property
     def mtime(self) -> datetime | None:
         """Last modification time carried forward from change records.
@@ -138,7 +144,7 @@ class Kanta(Generic[T]):
         """
         return self._impl.mtime
 
-    async def open(self, *, create: bool = True) -> None:
+    async def open(self, *, create: bool = True, readonly: bool = False) -> None:
         """Open the database file and start background persistence.
 
         This loads existing records, applies configured migrations, and starts
@@ -147,6 +153,9 @@ class Kanta(Generic[T]):
         Args:
             create: Whether to create the database file when missing.
                 If False, opening fails when the file does not exist or is empty.
+            readonly: If True, open the database read-only. No lock is acquired,
+                no background flush task is started, and transactions are
+                rejected. The file is not created if missing.
 
         Calling ``open`` more than once on the same instance is not allowed.
 
@@ -154,7 +163,7 @@ class Kanta(Generic[T]):
             kanta.exceptions.DatabaseError: If replay or decoding fails.
             kanta.exceptions.DataIntegrityError: If the instance is already open.
         """
-        await self._impl.open(create=create)
+        await self._impl.open(create=create, readonly=readonly)
 
     async def __aenter__(self) -> Kanta[T]:
         """Enter async context manager and open the database.

kanta/kantaimpl.py

@@ -7,11 +7,12 @@ import copy
 import importlib
 import logging
 from datetime import UTC, datetime
+from types import SimpleNamespace
 from typing import Any, Generic, TypeVar
 
 from kanta.callbacks import CallbackRegistry, InjectionContext
 from kanta.exceptions import DatabaseError, DataIntegrityError, ReplayError
-from kanta.migrate import MigrationRegistry
+from kanta.migrations import Migrations
 from kanta.persistence import PersistenceMixin
 from kanta.serialization import restore_data_in_place, struct_to_dict
 from kanta.serialization.base import replay
@@ -27,22 +28,23 @@ class KantaImpl(PersistenceMixin, Generic[T]):
     def __init__(self, **kwargs: Any):
         self.data_type = kwargs.pop("type")
         self.data: T = kwargs.pop("data")
-        self.migrations = kwargs.pop("migrations", None)
-        self.migration_ctx = kwargs.pop("migration_ctx", None)
         self._kanta = kwargs.pop("kanta", None)
+        migrations = kwargs.pop("migrations", None)
+        self.ctx = SimpleNamespace()
         super().__init__(**kwargs)
-        self.migration_registry: MigrationRegistry | None = None
-        if self.migrations is not None:
+        self.migrations: Migrations | None = None
+        if migrations is not None:
             module = (
-                importlib.import_module(self.migrations)
-                if isinstance(self.migrations, str)
-                else self.migrations
+                importlib.import_module(migrations)
+                if isinstance(migrations, str)
+                else migrations
             )
-            self.migration_registry = MigrationRegistry.from_module(module)
+            self.migrations = Migrations.from_module(module)
 
         self.in_transaction = False
         self.transaction_snapshot: dict[str, Any] | None = None
         self.opened = False
+        self.readonly = False
         self.bootstrap_action = "bootstrap"
         self.bootstrap_user: str | None = None
         self.bootstrap_mtime: bool | datetime = True
@@ -53,9 +55,7 @@ class KantaImpl(PersistenceMixin, Generic[T]):
         )
 
         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
-        )
+        self.version = self.migrations.dbver if self.migrations is not None else 0
 
     def add_bootstrap(
         self,
@@ -75,7 +75,7 @@ class KantaImpl(PersistenceMixin, Generic[T]):
         """Register one transaction logfmt callback."""
         self.callback_registry.register("logfmt", callback, path=path)
 
-    async def open(self, *, create: bool = True) -> None:
+    async def open(self, *, create: bool = True, readonly: bool = False) -> None:
         """Open the database: load from disk, apply migrations, start background task."""
         if self.opened:
             raise DataIntegrityError(
@@ -84,12 +84,17 @@ class KantaImpl(PersistenceMixin, Generic[T]):
                 action="open",
             )
 
+        self.readonly = readonly
         existed_before_open = self.filename.exists()
 
+        # Read-only mode never creates the file.
+        open_create = create and not readonly
+
         content = await asyncio.to_thread(
             self.file.open_and_read,
             self.filename,
-            create=create,
+            create=open_create,
+            readonly=readonly,
         )
 
         if not create and (not existed_before_open or not content):
@@ -134,10 +139,11 @@ class KantaImpl(PersistenceMixin, Generic[T]):
                     cause_type=type(e).__name__,
                 ) from e
 
-            if self.migration_registry is not None:
-                rr.version = self.migration_registry.apply(
-                    rr.state, rr.version, self.migration_ctx
-                )
+            migrations_ran = False
+            if self.migrations is not None:
+                previous_version = rr.version
+                rr.version = self.migrations.apply(rr.state, rr.version, self._kanta)
+                migrations_ran = rr.version != previous_version
 
             self.statedict = copy.deepcopy(rr.state)
             self.data = restore_data_in_place(
@@ -149,25 +155,45 @@ class KantaImpl(PersistenceMixin, Generic[T]):
             self.version = rr.version
             self.mtime = rr.m
             normalized = struct_to_dict(self.data, serializer=self.serializer)
+            if self.readonly:
+                self.statedict = copy.deepcopy(normalized)
+            else:
+                if migrations_ran:
+                    self.queue_change(
+                        f"migrate:v{self.version}",
+                        self.statedict,
+                        mtime=False,
+                        force=True,
+                    )
                 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.callback_registry.has("bootstrap"):
+        elif self.readonly:
+            self.file.close()
+            raise DataIntegrityError(
+                "Cannot open empty database in read-only mode",
+                db_path=self.filename,
+                action="open",
+            )
+        else:
             try:
+                if self.callback_registry.has("bootstrap"):
                     await self.callback_registry.invoke(
                         "bootstrap",
                         InjectionContext(data=self.data, kanta=self._kanta),
                     )
 
+                self.statedict = {}
                 current = struct_to_dict(self.data, serializer=self.serializer)
                 self.queue_change(
                     self.bootstrap_action,
                     current,
                     user=self.bootstrap_user,
                     mtime=self.bootstrap_mtime,
+                    force=True,
                 )
             except Exception:
                 self.file.close()
@@ -179,6 +205,7 @@ class KantaImpl(PersistenceMixin, Generic[T]):
 
         self.opened = True
 
+        if not self.readonly:
             self.background_task = asyncio.create_task(self._background_loop())
 
     async def close(self) -> None:
@@ -196,6 +223,7 @@ class KantaImpl(PersistenceMixin, Generic[T]):
 
         # Always run a final flush in case the background task never reached
         # its cancellation handler.
+        if not self.readonly:
             await self.flush()
 
         self.file.close()

kanta/migrate.py

@@ -1,117 +0,0 @@
-"""Database schema migration framework.
-
-Migrations are numbered functions discovered automatically via a decorator
-or by prefix. Each runs exactly once based on the current version.
-"""
-
-from __future__ import annotations
-
-import importlib
-import logging
-from types import ModuleType
-from typing import Any
-
-import msgspec
-
-_logger = logging.getLogger(__name__)
-
-
-class MigrationCtx(msgspec.Struct, omit_defaults=True):
-    """Context passed to each migration function.
-
-    Subclass or replace this with your own context type.
-    """
-
-    pass
-
-
-class MigrationRegistry:
-    """Registry of schema migration functions.
-
-    Usage::
-
-        registry = MigrationRegistry()
-
-        @registry.register
-        def migrate_v1(d: dict, ctx: MigrationCtx) -> None:
-            d.setdefault("version", 1)
-
-        new_version = registry.apply(state, current_version=0)
-
-    Or load from a module::
-
-        registry = MigrationRegistry.from_module("myapp.migrations")
-        new_version = registry.apply(state, current_version=0)
-    """
-
-    def __init__(self) -> None:
-        self._migrations: dict[int, Any] = {}
-
-    @staticmethod
-    def _migration_version(fn: Any) -> int:
-        name = getattr(fn, "__name__", "")
-        if not name.startswith("migrate_v"):
-            raise ValueError(f"Invalid migration function name: {name!r}")
-        suffix = name.removeprefix("migrate_v")
-        if not suffix.isdigit() or int(suffix) <= 0:
-            raise ValueError(f"Invalid migration version in function name: {name!r}")
-        return int(suffix)
-
-    def register(self, fn):
-        """Decorator to register a migration function."""
-        version = self._migration_version(fn)
-        self._migrations[version] = fn
-        return fn
-
-    @classmethod
-    def from_module(cls, module: str | ModuleType) -> MigrationRegistry:
-        """Create a registry by scanning a module for ``migrate_vN`` functions.
-
-        Args:
-            module: A module name (string) or an imported module object.
-        """
-        reg = cls()
-        if isinstance(module, str):
-            mod = importlib.import_module(module)
-        else:
-            mod = module
-
-        for name in dir(mod):
-            if name.startswith("migrate_v"):
-                fn = getattr(mod, name)
-                if callable(fn):
-                    version = reg._migration_version(fn)
-                    reg._migrations[version] = fn
-        return reg
-
-    @property
-    def dbver(self) -> int:
-        """Current schema version (= highest discovered migration, or 0)."""
-        return max(self._migrations.keys(), default=0)
-
-    def apply(
-        self,
-        data_dict: dict[str, Any],
-        current_version: int,
-        ctx: MigrationCtx | None = None,
-        *,
-        silent: bool = False,
-    ) -> int:
-        """Apply pending migrations to *data_dict* in place.
-
-        Returns the new version after all migrations.
-        """
-        while current_version < self.dbver:
-            next_version = current_version + 1
-            fn = self._migrations.get(next_version)
-            if fn is None:
-                raise ValueError(
-                    f"Missing migration step migrate_v{next_version} "
-                    f"(highest discovered is v{self.dbver})"
-                )
-            fn(data_dict, ctx or MigrationCtx())
-            current_version = next_version
-            if not silent:
-                desc = (fn.__doc__ or fn.__name__).split("\n")[0].rstrip(".")
-                _logger.info("Applied migration %s: %s", fn.__name__, desc)
-        return current_version

kanta/migrations.py

@@ -0,0 +1,154 @@
+"""Database schema migration framework.
+
+Migrations are numbered functions discovered automatically via a decorator
+or by prefix. Each runs exactly once based on the current version.
+"""
+
+from __future__ import annotations
+
+import copy
+import importlib
+import inspect
+import logging
+from types import ModuleType
+from typing import Any
+
+from kanta.exceptions import DatabaseError
+
+_logger = logging.getLogger(__name__)
+
+# Cache registries by imported module object so that many Kanta instances using
+# the same migrations module do not re-scan it each time.
+_module_registry_cache: dict[ModuleType, Migrations] = {}
+
+
+class Migrations:
+    """Registry of schema migration functions.
+
+    Usage::
+
+        migrations = Migrations()
+
+        @migrations.register
+        def migrate_v1(d: dict, kanta) -> None:
+            d.setdefault("version", 1)
+            kanta.ctx.note = "migrated"
+
+        @migrations.register
+        def migrate_v2(d: dict) -> None:
+            d.setdefault("version", 2)
+
+        new_version = migrations.apply(state, current_version=0, kanta=kanta)
+
+    Or load from a module::
+
+        migrations = Migrations.from_module("myapp.migrations")
+        new_version = migrations.apply(state, current_version=0, kanta=kanta)
+    """
+
+    def __init__(self) -> None:
+        self._migrations: dict[int, Any] = {}
+
+    @staticmethod
+    def _migration_version(fn: Any) -> int:
+        name = getattr(fn, "__name__", "")
+        if not name.startswith("migrate_v"):
+            raise ValueError(f"Invalid migration function name: {name!r}")
+        suffix = name.removeprefix("migrate_v")
+        if not suffix.isdigit() or int(suffix) <= 0:
+            raise ValueError(f"Invalid migration version in function name: {name!r}")
+        return int(suffix)
+
+    def register(self, fn):
+        """Decorator to register a migration function."""
+        version = self._migration_version(fn)
+        self._migrations[version] = fn
+        return fn
+
+    @classmethod
+    def from_module(cls, module: str | ModuleType) -> Migrations:
+        """Create or retrieve a cached registry by scanning a module.
+
+        Args:
+            module: A module name (string) or an imported module object.
+        """
+        if isinstance(module, str):
+            mod = importlib.import_module(module)
+        else:
+            mod = module
+
+        try:
+            return _module_registry_cache[mod]
+        except KeyError:
+            pass
+
+        reg = cls()
+        for name in dir(mod):
+            if name.startswith("migrate_v"):
+                fn = getattr(mod, name)
+                if callable(fn):
+                    version = reg._migration_version(fn)
+                    reg._migrations[version] = fn
+        _module_registry_cache[mod] = reg
+        return reg
+
+    @property
+    def dbver(self) -> int:
+        """Current schema version (= highest discovered migration, or 0)."""
+        return max(self._migrations.keys(), default=0)
+
+    @property
+    def minver(self) -> int:
+        """Minimum supported current version (first migration minus 1, or 0)."""
+        return min(self._migrations.keys(), default=1) - 1
+
+    @staticmethod
+    def _call_migration(fn: Any, data_dict: dict[str, Any], kanta: Any) -> None:
+        """Call *fn* with the data dict and, if accepted, the Kanta instance."""
+        try:
+            inspect.signature(fn).bind(data_dict, kanta)
+        except TypeError:
+            fn(data_dict)
+        else:
+            fn(data_dict, kanta)
+
+    def apply(
+        self,
+        data_dict: dict[str, Any],
+        current_version: int,
+        kanta: Any,
+        *,
+        silent: bool = False,
+    ) -> int:
+        """Apply pending migrations to *data_dict* in place.
+
+        Missing intermediate migration steps are silently skipped.
+
+        Raises:
+            DatabaseError: If the database version is newer than the highest
+                supported version or older than the minimum supported version.
+
+        Returns the new version after all migrations.
+        """
+        if current_version > self.dbver:
+            raise DatabaseError(
+                f"Database version v{current_version} is newer than the "
+                f"highest supported version v{self.dbver}"
+            )
+        if current_version < self.minver:
+            raise DatabaseError(
+                f"Database version v{current_version} is older than the "
+                f"minimum supported version v{self.minver}"
+            )
+
+        for version in sorted(self._migrations.keys()):
+            if version <= current_version:
+                continue
+            fn = self._migrations[version]
+            before = copy.deepcopy(data_dict) if not silent else None
+            self._call_migration(fn, data_dict, kanta)
+            current_version = version
+            if not silent and before != data_dict:
+                desc = (fn.__doc__ or fn.__name__).split("\n")[0].rstrip(".")
+                _logger.info("Applied migration %s: %s", fn.__name__, desc)
+        return current_version

kanta/persistence.py

@@ -39,6 +39,7 @@ class PersistenceMixin:
     flush_interval: float
     version: int
     opened: bool
+    readonly: bool
     mtime: datetime | None
 
     def __init__(self, **kwargs: Any) -> None:
@@ -68,6 +69,8 @@ class PersistenceMixin:
 
     async def _background_loop(self) -> None:
         """Background task that periodically flushes changes to disk."""
+        if self.readonly:
+            return
         while True:
             try:
                 await asyncio.sleep(self.flush_interval)
@@ -106,6 +109,7 @@ class PersistenceMixin:
         *,
         user: str | None = None,
         mtime: bool | datetime = True,
+        force: bool = False,
     ) -> ChangeRecord | None:
         """Queue a change record internally (thread-safe).
 
@@ -118,9 +122,11 @@ class PersistenceMixin:
                 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.
+            force: If ``True``, queue the record even when the diff is empty.
 
         Returns:
-            The queued :class:`ChangeRecord`, or ``None`` if the diff was empty.
+            The queued :class:`ChangeRecord`, or ``None`` if the diff was empty
+            and *force* is ``False``.
         """
         now = datetime.now(UTC)
 
@@ -135,7 +141,9 @@ class PersistenceMixin:
 
         diff = compute_diff(self.statedict, current)
         if not diff:
+            if not force:
                 return None
+            diff = {}
 
         record = ChangeRecord(
             ts=now,
@@ -160,6 +168,13 @@ class PersistenceMixin:
                 action="flush_sync",
             )
 
+        if self.readonly:
+            raise DataIntegrityError(
+                "Cannot flush in read-only mode",
+                db_path=self.filename,
+                action="flush_sync",
+            )
+
         if self.flush_failed:
             return
 
@@ -207,6 +222,13 @@ class PersistenceMixin:
                 action="flush",
             )
 
+        if self.readonly:
+            raise DataIntegrityError(
+                "Cannot flush in read-only mode",
+                db_path=self.filename,
+                action="flush",
+            )
+
         if self.flush_failed:
             return
 

kanta/transaction.py

@@ -24,6 +24,13 @@ def transaction(
     mtime: bool | datetime = True,
 ):
     """Wrap writes in a transaction and yield the live db object."""
+    if impl.readonly:
+        raise DataIntegrityError(
+            "Cannot start transaction in read-only mode",
+            db_path=impl.filename,
+            action=action,
+        )
+
     if impl.in_transaction:
         raise RuntimeError(
             "Nested or simultaneous transactions are not supported "

tests/support.py

@@ -70,6 +70,18 @@ def change_actions(path: Path, format_config) -> list[str]:
     return actions
 
 
+def read_changes(path: Path, format_config) -> list[ChangeRecord]:
+    _, serializer_cls = format_config
+    serializer = serializer_cls()
+    framer = serializer.framer_cls()
+    records: list[ChangeRecord] = []
+    for is_snapshot, payload, _, _ in framer.iter_records(path.read_bytes(), 0):
+        if is_snapshot:
+            continue
+        records.append(serializer.decode(payload, type=ChangeRecord))
+    return records
+
+
 def make_migrations_module(name: str, fn_name: str, fn):
     mod = ModuleType(name)
     mod.__dict__[fn_name] = fn

tests/test_callbacks.py

@@ -1,4 +1,4 @@
-from typing import Any
+from typing import Any, Optional, Union
 
 import pytest
 
@@ -49,16 +49,60 @@ def test_logfmt_requires_value_annotation(tmp_path, format_config):
             return None
 
 
-def test_logfmt_requires_return_annotation(tmp_path, format_config):
+def test_logfmt_allows_missing_return_annotation(tmp_path, format_config):
     kanta = make_kanta(tmp_path / "test.db", Data, format_config)
 
-    with pytest.raises(TypeError, match="must annotate its return"):
-
     @kanta.logfmt
     def resolve_names(value: str, current: DictPost):
         return None
 
 
+def test_logfmt_class_allows_missing_return_annotation(tmp_path, format_config):
+    kanta = make_kanta(tmp_path / "test.db", Data, format_config)
+
+    @kanta.logfmt
+    class UserLogFmt(LogFmt):
+        def resolve(self, value: str, path: str):
+            return None
+
+
+# fmt: off
+def test_logfmt_accepts_optional_return_typing_forms(tmp_path, format_config):
+    kanta = make_kanta(tmp_path / "test.db", Data, format_config)
+
+    @kanta.logfmt
+    def resolve_optional(value: str) -> Optional[str]:  # noqa: UP007
+        return value
+
+    @kanta.logfmt
+    def resolve_union(value: str) -> Union[str, None]:  # noqa: UP007
+        return value
+
+    @kanta.logfmt
+    def resolve_pipe(value: "str") -> "str | None":
+        return value
+
+
+def test_logfmt_class_accepts_optional_return_typing_forms(tmp_path, format_config):
+    kanta = make_kanta(tmp_path / "test.db", Data, format_config)
+
+    @kanta.logfmt
+    class OptionalStyle(LogFmt):
+        def resolve(self, value: str, path: str) -> Optional[str]:  # noqa: UP007
+            return value
+
+    @kanta.logfmt
+    class UnionStyle(LogFmt):
+        def resolve(self, value: str, path: str) -> Union[str, None]:  # noqa: UP007
+            return value
+
+    @kanta.logfmt
+    class StringStyle(LogFmt):
+        def resolve(self, value: "str", path: "str") -> "str | None":
+            return value
+# fmt: on
+
+
 def test_logfmt_rejects_async_callback(tmp_path, format_config):
     kanta = make_kanta(tmp_path / "test.db", Data, format_config)
 

tests/test_kanta_integration.py

@@ -17,6 +17,8 @@ from .support import (
     change_actions,
     fixed_change,
     make_kanta,
+    make_migrations_module,
+    read_changes,
     seed_single_change,
 )
 
@@ -30,6 +32,44 @@ async def test_load_empty(tmp_path, format_config):
     await kanta.close()
 
 
+@pytest.mark.asyncio
+async def test_new_file_writes_bootstrap_record_without_handlers(
+    tmp_path, format_config
+):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open()
+    await kanta.close()
+
+    records = read_changes(path, format_config)
+    assert len(records) == 1
+    assert records[0].a == "bootstrap"
+    assert records[0].diff == {"$replace": {"users": {}, "counter": 0}}
+
+
+@pytest.mark.asyncio
+async def test_new_file_persists_initial_state_for_roundtrip(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(
+        path, Data(counter=5, users={"alice": User(name="Alice")}), format_config
+    )
+    await kanta.open()
+    await kanta.close()
+
+    records = read_changes(path, format_config)
+    assert len(records) == 1
+    assert records[0].a == "bootstrap"
+    assert records[0].diff == {
+        "$replace": {"users": {"alice": {"name": "Alice", "age": 0}}, "counter": 5}
+    }
+
+    kanta2 = make_kanta(path, Data, format_config)
+    await kanta2.open()
+    assert kanta2.data.counter == 5
+    assert kanta2.data.users["alice"].name == "Alice"
+    await kanta2.close()
+
+
 @pytest.mark.asyncio
 async def test_open_overwrites_caller_owned_root_data(tmp_path, format_config):
     path = tmp_path / "test.db"
@@ -404,7 +444,7 @@ async def test_migrations_from_module(tmp_path, format_config):
 
     mod = type(sys)("test_migrations")
 
-    def migrate_v1(d, ctx):
+    def migrate_v1(d, kanta):
         d["version"] = 1
 
     mod.__dict__["migrate_v1"] = migrate_v1
@@ -434,6 +474,43 @@ async def test_msgspec_normalization_logs_migration(tmp_path, format_config):
     assert "migrate:msgspec" in change_actions(path, format_config)
 
 
+@pytest.mark.asyncio
+async def test_empty_migration_is_recorded_and_not_reapplied(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    seed_single_change(
+        path, fixed_change("init", {"counter": 0, "users": {}}), format_config
+    )
+
+    def migrate_v1(d, kanta):
+        """No-op migration that only bumps the schema version."""
+        pass
+
+    mod = make_migrations_module("empty_migration_mod", "migrate_v1", migrate_v1)
+
+    try:
+        kanta = make_kanta(path, Data, format_config, migrations=mod)
+        await kanta.open()
+        assert kanta.version == 1
+        await kanta.flush()
+        await kanta.close()
+
+        records = read_changes(path, format_config)
+        migration_records = [r for r in records if r.a.startswith("migrate")]
+        assert len(migration_records) == 1
+        assert migration_records[0].v == 1
+        assert migration_records[0].diff == {}
+
+        kanta2 = make_kanta(path, Data, format_config, migrations=mod)
+        await kanta2.open()
+        assert kanta2.version == 1
+        await kanta2.close()
+
+        records2 = read_changes(path, format_config)
+        assert len([r for r in records2 if r.a.startswith("migrate")]) == 1
+    finally:
+        sys.modules.pop("empty_migration_mod", None)
+
+
 @pytest.mark.asyncio
 async def test_open_locked_file_raises_filelock_error(tmp_path, format_config):
     path = tmp_path / "test.db"
@@ -514,7 +591,7 @@ async def test_migrations_from_module_path(tmp_path, format_config):
     module_name = "test_migrations_path"
     mod = type(sys)(module_name)
 
-    def migrate_v1(d, ctx):
+    def migrate_v1(d, kanta):
         d["counter"] = 2
 
     mod.__dict__["migrate_v1"] = migrate_v1

tests/test_migrations.py

@@ -1,53 +1,204 @@
-from types import ModuleType
+import logging
+from types import ModuleType, SimpleNamespace
 
-from kanta.migrate import MigrationRegistry
+import pytest
+
+from kanta.exceptions import DatabaseError
+from kanta.migrations import Migrations
+
+
+class _DummyKanta:
+    def __init__(self):
+        self.ctx = SimpleNamespace()
 
 
 def test_register_and_apply():
-    reg = MigrationRegistry()
+    reg = Migrations()
+    kanta = _DummyKanta()
 
     @reg.register
-    def migrate_v1(d, ctx):
+    def migrate_v1(d, kanta):
         d["version"] = 1
 
     @reg.register
-    def migrate_v2(d, ctx):
+    def migrate_v2(d, kanta):
         d["version"] = 2
 
     state = {}
-    new_ver = reg.apply(state, current_version=0, silent=True)
+    new_ver = reg.apply(state, current_version=0, kanta=kanta, silent=True)
     assert new_ver == 2
     assert state["version"] == 2
 
 
 def test_no_migrations_needed():
-    reg = MigrationRegistry()
+    reg = Migrations()
+    kanta = _DummyKanta()
 
     @reg.register
-    def migrate_v1(d, ctx):
+    def migrate_v1(d, kanta):
         d["x"] = 1
 
     state = {"x": 1}
-    new_ver = reg.apply(state, current_version=1, silent=True)
+    new_ver = reg.apply(state, current_version=1, kanta=kanta, silent=True)
     assert new_ver == 1
 
 
 def test_from_module():
     mod = ModuleType("fake_migrations")
+    kanta = _DummyKanta()
 
-    def migrate_v1(d, ctx):
+    def migrate_v1(d, kanta):
         d["v"] = 1
 
-    def migrate_v2(d, ctx):
+    def migrate_v2(d, kanta):
         d["v"] = 2
 
     mod.__dict__["migrate_v1"] = migrate_v1
     mod.__dict__["migrate_v2"] = migrate_v2
 
-    reg = MigrationRegistry.from_module(mod)
+    reg = Migrations.from_module(mod)
     assert reg.dbver == 2
 
     state = {}
-    new_ver = reg.apply(state, current_version=0, silent=True)
+    new_ver = reg.apply(state, current_version=0, kanta=kanta, silent=True)
     assert new_ver == 2
     assert state["v"] == 2
+
+
+def test_migrations_can_use_kanta_ctx():
+    reg = Migrations()
+    kanta = _DummyKanta()
+
+    @reg.register
+    def migrate_v1(d, kanta):
+        kanta.ctx.source = "migration"
+        d["source"] = kanta.ctx.source
+
+    state = {}
+    new_ver = reg.apply(state, current_version=0, kanta=kanta, silent=True)
+    assert new_ver == 1
+    assert state["source"] == "migration"
+    assert kanta.ctx.source == "migration"
+
+
+def test_migration_can_omit_kanta_argument():
+    reg = Migrations()
+    kanta = _DummyKanta()
+
+    @reg.register
+    def migrate_v1(d):
+        d["x"] = 1
+
+    state = {}
+    new_ver = reg.apply(state, current_version=0, kanta=kanta, silent=True)
+    assert new_ver == 1
+    assert state["x"] == 1
+
+
+def test_version_too_new():
+    reg = Migrations()
+    kanta = _DummyKanta()
+
+    @reg.register
+    def migrate_v1(d):
+        d["x"] = 1
+
+    with pytest.raises(
+        DatabaseError,
+        match="Database version v2 is newer than the highest supported version v1",
+    ):
+        reg.apply({}, current_version=2, kanta=kanta, silent=True)
+
+
+def test_version_too_old():
+    reg = Migrations()
+    kanta = _DummyKanta()
+
+    @reg.register
+    def migrate_v3(d):
+        d["x"] = 3
+
+    with pytest.raises(
+        DatabaseError,
+        match="Database version v1 is older than the minimum supported version v2",
+    ):
+        reg.apply({}, current_version=1, kanta=kanta, silent=True)
+
+
+def test_missing_middle_migration_is_skipped():
+    reg = Migrations()
+    kanta = _DummyKanta()
+
+    @reg.register
+    def migrate_v1(d):
+        d["x"] = 1
+
+    @reg.register
+    def migrate_v3(d):
+        d["y"] = 3
+
+    state = {"x": 1}
+    new_ver = reg.apply(state, current_version=1, kanta=kanta, silent=True)
+    assert new_ver == 3
+    assert state["x"] == 1
+    assert state["y"] == 3
+
+
+def test_old_migrations_deleted_current_supported():
+    reg = Migrations()
+    kanta = _DummyKanta()
+
+    @reg.register
+    def migrate_v3(d):
+        d["x"] = 3
+
+    state = {"x": 2}
+    new_ver = reg.apply(state, current_version=2, kanta=kanta, silent=True)
+    assert new_ver == 3
+    assert state["x"] == 3
+
+
+def test_migration_log_only_when_changed(caplog):
+    reg = Migrations()
+    kanta = _DummyKanta()
+
+    @reg.register
+    def migrate_v1(d):
+        """Set x."""
+        d["x"] = 1
+
+    @reg.register
+    def migrate_v2(d):
+        """No-op."""
+        pass
+
+    @reg.register
+    def migrate_v3(d):
+        """Set y."""
+        d["y"] = 3
+
+    with caplog.at_level(logging.INFO, logger="kanta.migrations"):
+        reg.apply({}, current_version=0, kanta=kanta)
+
+    messages = [r.message for r in caplog.records if r.levelno == logging.INFO]
+    assert len(messages) == 2
+    assert "migrate_v1" in messages[0]
+    assert "Set x" in messages[0]
+    assert "migrate_v3" in messages[1]
+    assert "Set y" in messages[1]
+
+
+def test_no_op_migration_produces_no_log(caplog):
+    reg = Migrations()
+    kanta = _DummyKanta()
+
+    @reg.register
+    def migrate_v1(d):
+        """No-op."""
+        pass
+
+    with caplog.at_level(logging.INFO, logger="kanta.migrations"):
+        reg.apply({}, current_version=0, kanta=kanta)
+
+    info_messages = [r for r in caplog.records if r.levelno == logging.INFO]
+    assert not info_messages

tests/test_mtime.py

@@ -82,8 +82,9 @@ async def test_transaction_mtime_false_preserves_mtime(tmp_path, format_config):
             continue
         records.append(serializer.decode(payload, type=ChangeRecord))
 
-    assert records[0].m == first_m
-    assert records[1].m is None
+    assert records[0].a == "bootstrap"
+    assert records[1].m == first_m
+    assert records[2].m is None
     assert kanta.mtime == first_m
 
 

tests/test_readonly.py

@@ -0,0 +1,153 @@
+"""Tests for Kanta read-only mode."""
+
+import pytest
+
+from kanta.exceptions import DataIntegrityError, FileLockError
+from kanta.serialization import struct_to_dict
+
+from .support import (
+    Data,
+    EvolvableDataV2,
+    fixed_change,
+    make_kanta,
+    make_migrations_module,
+    seed_single_change,
+)
+
+
+@pytest.mark.asyncio
+async def test_readonly_opens_existing_database(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    seed_single_change(path, fixed_change("seed", {"counter": 7}), format_config)
+
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open(readonly=True)
+
+    assert isinstance(kanta.data, Data)
+    assert kanta.data.counter == 7
+    assert kanta._impl.readonly is True
+    assert kanta._impl.background_task is None
+
+    await kanta.close()
+
+
+@pytest.mark.asyncio
+async def test_readonly_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(readonly=True)
+
+    assert not path.exists()
+
+
+@pytest.mark.asyncio
+async def test_readonly_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(readonly=True)
+
+
+@pytest.mark.asyncio
+async def test_readonly_transaction_fails(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    seed_single_change(path, fixed_change("seed", {"counter": 1}), format_config)
+
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open(readonly=True)
+
+    with pytest.raises(DataIntegrityError, match="read-only"):
+        with kanta.transaction(action="inc") as data:
+            data.counter = 2
+
+    # In-memory state must remain unchanged.
+    assert kanta.data.counter == 1
+    await kanta.close()
+
+
+@pytest.mark.asyncio
+async def test_readonly_flush_fails(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    seed_single_change(path, fixed_change("seed", {"counter": 1}), format_config)
+
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open(readonly=True)
+
+    with pytest.raises(DataIntegrityError, match="read-only"):
+        await kanta.flush()
+
+    await kanta.close()
+
+
+@pytest.mark.asyncio
+async def test_readonly_create_true_does_not_create_file(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    with pytest.raises(FileLockError):
+        await kanta.open(create=True, readonly=True)
+
+    assert not path.exists()
+
+
+@pytest.mark.asyncio
+async def test_readonly_does_not_persist_changes(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    seed_single_change(path, fixed_change("seed", {"counter": 1}), format_config)
+    original_content = path.read_bytes()
+
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open(readonly=True)
+    await kanta.close()
+
+    assert path.read_bytes() == original_content
+
+
+@pytest.mark.asyncio
+async def test_readonly_runs_migrations(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    seed_single_change(
+        path,
+        fixed_change("seed", {"counter": 1}, version=0),
+        format_config,
+    )
+
+    def migrate_v1(data, kanta):
+        data.setdefault("enabled", True)
+
+    migrations = make_migrations_module("readonly_migrations", "migrate_v1", migrate_v1)
+
+    kanta = make_kanta(path, EvolvableDataV2, format_config, migrations=migrations)
+    await kanta.open(readonly=True)
+
+    assert kanta.data.counter == 1
+    # Migration ran in memory even though no change was persisted.
+    assert struct_to_dict(kanta.data, serializer=kanta._impl.serializer) == {
+        "counter": 1,
+        "enabled": True,
+    }
+    assert not kanta._impl.pending_changes
+
+    await kanta.close()
+
+
+@pytest.mark.asyncio
+async def test_readwrite_and_readonly_can_open_together(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    seed_single_change(path, fixed_change("seed", {"counter": 1}), format_config)
+
+    rw = make_kanta(path, Data, format_config)
+    await rw.open()
+
+    ro = make_kanta(path, Data, format_config)
+    await ro.open(readonly=True)
+
+    assert rw.data.counter == 1
+    assert ro.data.counter == 1
+
+    await ro.close()
+    await rw.close()