Initial commit

.gitignore

@@ -0,0 +1,14 @@
+# Python cache/artifacts
+__pycache__/
+*.py[cod]
+*.so
+
+# Hidden things
+.*
+!.gitignore
+
+# Build/distribution artifacts
+build/
+dist/
+*.egg-info/
+*.lock

README.md

@@ -0,0 +1,76 @@
+# Kanta database
+
+Kanta is a small embedded NoSQL store for async Python apps. It keeps live state in memory, writes transactional diffs to an append-only log, and supports versioned schema migrations.
+
+## Why Kanta
+
+- Fast synchronous reads and modifications on native Python objects
+- Durable writes with append-only file and periodic snapshots
+- Transaction semantics with rollback on failure
+- Explicit schema evolution via `migrate_vN` functions
+- Line-based JSON or binary MessagePack, or bring your own serializer
+- Even in JSON we can use non-string keys, bytes, datetimes, UUID and other types
+
+This design is often preferable when you want low-latency local persistence without operating a separate database service. You get straightforward deployment, auditable history, and deterministic replay while keeping application state ergonomic to work with.
+
+Queries and updates on native data are far faster than over an SQL server connection, and we can can provide fully synchronous operation. The limitation is that you can only use the same database within a single process at a time, but this is well suited for async programming.
+
+## Quick Start
+
+```python
+import asyncio
+import msgspec
+from uuid import UUID, uuid7
+
+from kanta import Kanta
+
+
+class User(msgspec.Struct):
+    name: str = ""
+    email: str | None = None
+
+
+class Data(msgspec.Struct):
+    users: dict[UUID, User] = {}
+
+
+async def main() -> None:
+    async with Kanta("data.kantadb", Data()) as kanta:
+        user_id = uuid7()
+        with kanta.transaction(action="create_user") as data:
+            data.users[user_id] = User(name="Alice")
+
+
+asyncio.run(main())
+```
+
+## Core Concepts
+
+1. Define your schema as a `msgspec.Struct` root object.
+2. Mutate data inside `with kanta.transaction(...):`.
+3. Let Kanta flush queued changes to disk in the background.
+4. Use snapshots and replay for fast startup and full history.
+
+## 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.
+
+Pass a module (or import path) containing `migrate_vN` functions:
+
+```python
+kanta = Kanta("data.kantadb", Data(), migrations="myapp.migrations")
+await kanta.open()
+```
+
+Kanta tracks migration version metadata automatically, and fast forwards your database to current version by running all the migrations needed while opening the database.
+
+## On-Disk Format
+
+Kanta in JSON mode (default) stores newline-delimited records. Transaction history is viewable by any simple text editor, and rollbacks to prior state are done by simply removing final lines (one per transaction)
+
+MsgPack mode uses binary records with length and checksum to avoid data corruption.
+
+- Change line: JSON object with metadata + `diff`
+- Snapshot line: `SNAPSHOT { ... full state ... }`
+
+See `docs/database.md` for format details and invariants.

__init__.py

@@ -0,0 +1,16 @@
+"""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

@@ -0,0 +1,116 @@
+# Kanta Database Format and Design Principles
+
+This document describes the on-disk format and design principles of Kanta.
+It is intentionally focused on the current standalone package behavior.
+
+## Core Principles
+
+1. Append-only durability
+- State changes are persisted as appended JSON lines.
+- Existing lines are never edited in place.
+
+2. Differential persistence
+- Kanta stores diffs (patches), not full state, for normal writes.
+- This keeps write volume small and preserves a clear change history.
+
+3. Deterministic replay
+- Current state is reconstructed by replaying log records in order.
+- Snapshot records accelerate replay while preserving deterministic results.
+
+4. Transactional in-memory writes
+- Application code mutates in-memory data inside `kanta.transaction(...)`.
+- On success, Kanta computes and queues a diff record.
+- On failure, in-memory data is rolled back.
+
+5. Explicit schema evolution
+- Schema migration functions are versioned (`migrate_vN`).
+- Migrations run at open time and advance the stored version.
+
+## On-Disk Record Types
+
+Kanta uses a newline-delimited stream where each line is either a change
+record or a snapshot record.
+
+### Change record
+
+One JSON object per line:
+
+```json
+{"ts":"2026-06-10T02:55:00Z","a":"update","v":5,"u":"user-id","m":"2026-06-10T02:55:00Z","diff":{"users":{"alice":{"age":31}}}}
+```
+
+Fields:
+- `ts`: UTC timestamp of the record.
+- `a`: action name.
+- `v`: schema version after this change.
+- `u`: optional actor identifier.
+- `m`: optional domain modification timestamp.
+- `diff`: jsondiff patch payload.
+
+### Snapshot record
+
+Snapshot lines are prefixed with `SNAPSHOT `, followed by JSON:
+
+```text
+SNAPSHOT {"ts":"2026-06-10T00:00:00Z","v":5,"state":{"users":{}},"m":"2026-06-10T00:00:00Z"}
+```
+
+Fields:
+- `ts`: snapshot creation time.
+- `v`: schema version represented by the snapshot.
+- `state`: full state dictionary.
+- `m`: optional domain modification timestamp.
+
+## Replay Model
+
+1. Find the last snapshot in the file, if present.
+2. Initialize replay state from snapshot state (or `{}` if none).
+3. Replay subsequent change records in order using patch application.
+4. The final replay state becomes in-memory `kanta.data`.
+
+This model provides fast startup for large logs while retaining append-only
+history.
+
+## Serialization Semantics
+
+- In-memory data is defined by an application `msgspec.Struct` type.
+- Kanta round-trips through plain builtins for persistence and diffing.
+- Dict keys are serialized as strings (`str_keys=True`) for stable JSON form.
+- Normalization changes introduced by struct decode/encode are logged as
+  `migrate:msgspec` when they produce a diff.
+
+## Transaction Semantics
+
+- `kanta.transaction(action=...)` captures a pre-transaction snapshot dict.
+- On success:
+  - compute diff between previous builtins and current builtins,
+  - queue a `ChangeRecord` if non-empty.
+- On exception:
+  - restore in-memory data from snapshot,
+  - re-raise the exception.
+
+Nested transactions are rejected.
+
+## Flush and Lifecycle
+
+- Writes are queued in memory.
+- `kanta.flush()` appends queued records to disk.
+- A background async task can flush periodically.
+- `kanta.close()` performs final flush and releases file resources.
+- `async with Kanta(...)` guarantees open/close lifecycle management.
+
+## Migrations
+
+- Migration source is configured on `Kanta(...)` via `migrations=`.
+- Accepted values:
+  - imported module object,
+  - import path string.
+- Migrations mutate replayed dict state in-place and return the new version.
+
+## Safety Invariants
+
+- Any detected out-of-transaction mutation is treated as a fatal consistency
+  violation.
+- Flush failures mark the instance as failed and trigger shutdown behavior.
+- Object identity of `kanta.data` is preserved across rollback when possible,
+  minimizing stale-reference hazards for callers.

kanta/__init__.py

@@ -0,0 +1,26 @@
+from .diff import compute_diff
+from .diff import replay_jsonl as replay
+from .exceptions import DatabaseError, DataIntegrityError, FileLockError, ReplayError
+from .filelock import LockedFile
+from .kanta import Kanta
+from .logging import configure_logging, format_diff, log_change
+from .serialization import JsonSerializer, MsgPackSerializer
+from .structs import ChangeRecord, Snapshot
+
+__all__ = [
+    "ChangeRecord",
+    "compute_diff",
+    "configure_logging",
+    "DataIntegrityError",
+    "DatabaseError",
+    "FileLockError",
+    "format_diff",
+    "JsonSerializer",
+    "Kanta",
+    "LockedFile",
+    "log_change",
+    "MsgPackSerializer",
+    "ReplayError",
+    "replay",
+    "Snapshot",
+]

kanta/diff.py

@@ -0,0 +1,81 @@
+"""Diff computation and replay utilities."""
+
+import jsondiff
+
+from kanta.kanta.structs import ChangeRecord
+from kanta.serialization.base import ReplayResult, replay
+from kanta.serialization.framing import LineFramer
+from kanta.serialization.json import JsonSerializer
+
+
+def compute_diff(previous: dict, current: dict) -> dict | None:
+    """Compute a jsondiff patch between two dicts.
+
+    Returns None if there is no difference.
+    """
+    return jsondiff.diff(previous, current, marshal=True) or None
+
+
+def _apply_diff(state: dict, diff: dict) -> dict:
+    """Apply a jsondiff patch manually, handling ``$replace`` and ``$delete``.
+
+    jsondiff.patch does not handle nested ``$replace`` commands when the
+    parent key is missing from the state.  This function recursively applies
+    diffs, treating ``$replace`` as full replacement and ``$delete`` as
+    key removal.
+    """
+    if not isinstance(diff, dict):
+        return diff
+
+    result = dict(state) if isinstance(state, dict) else state
+    if not isinstance(result, dict):
+        result = {}
+
+    for key, value in diff.items():
+        if key == "$replace":
+            return value
+        elif key == "$delete":
+            if isinstance(value, list):
+                for k in value:
+                    result.pop(k, None)
+            else:
+                result.pop(value, None)
+        elif isinstance(value, dict):
+            old = result.get(key, {})
+            if not isinstance(old, dict):
+                old = {}
+            result[key] = _apply_diff(old, value)
+        else:
+            result[key] = value
+
+    return result
+
+
+def patch_state(state: dict, diff: dict) -> dict:
+    """Apply a jsondiff patch to a state dict.
+
+    The diff was produced with ``marshal=True`` (string keys like
+    ``"$replace"`` and ``"$delete"``) and decoded from JSON.
+    """
+    return _apply_diff(state, diff)
+
+
+# Backward-compatible JSONL replay using the default serializer.
+_default_serializer = JsonSerializer()
+_default_framer = LineFramer()
+
+
+def replay_jsonl(
+    data: bytes,
+    *,
+    type: type[ChangeRecord] = ChangeRecord,
+) -> ReplayResult:
+    """Replay database state from JSONL file data.
+
+    This is the legacy public API that hard-codes JSON/JSONL handling.
+    """
+    return replay(
+        data,
+        framer=_default_framer,
+        decode=_default_serializer.decode,
+    )

kanta/exceptions.py

@@ -0,0 +1,61 @@
+"""Custom exception types for Kanta."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+
+class DatabaseError(ValueError):
+    """Exception raised for database loading errors."""
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        db_path: Path | None = None,
+        line_number: int | None = None,
+        byte_pos: int | None = None,
+        cause_type: str | None = None,
+    ):
+        self.db_path = db_path
+        self.line_number = line_number
+        self.byte_pos = byte_pos
+        self.cause_type = cause_type
+        super().__init__(message)
+
+
+class ReplayError(DatabaseError):
+    """Structured replay error with source location metadata."""
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        line_number: int | None = None,
+        byte_pos: int | None = None,
+        record_type: str | None = None,
+    ):
+        self.record_type = record_type
+        super().__init__(message, line_number=line_number, byte_pos=byte_pos)
+
+
+class FileLockError(DatabaseError):
+    """Raised when database file open/lock operations fail."""
+
+
+class DataIntegrityError(RuntimeError):
+    """Raised when in-memory data integrity invariants are violated."""
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        db_path: Path | None = None,
+        action: str | None = None,
+        diff: dict[str, Any] | None = None,
+    ):
+        self.db_path = db_path
+        self.action = action
+        self.diff = diff
+        super().__init__(message)

kanta/filelock.py

@@ -0,0 +1,274 @@
+"""Cross-platform locked file for the database (no separate .lock files).
+
+Unix:    open() + fcntl.flock (advisory, cooperative among processes that flock).
+Windows: CreateFileW with FILE_SHARE_READ (OS-enforced, allows readers, blocks writers).
+
+A single file descriptor is opened once for both reading and writing.
+The lock is acquired atomically (on Windows) or immediately after open (on Unix),
+and the same descriptor is used for the lifetime of the process: first to read
+the existing content, then to append new writes.
+"""
+
+import logging
+import os
+import sys
+from pathlib import Path
+
+from kanta.exceptions import FileLockError
+
+_logger = logging.getLogger(__name__)
+
+
+def _fatal(msg: str, *, db_path: Path | None = None) -> None:
+    """Log a fatal error and raise a typed exception."""
+    _logger.critical(msg)
+    raise FileLockError(msg, db_path=db_path)
+
+
+if sys.platform == "win32":
+    import ctypes
+    from ctypes import wintypes
+
+    _kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
+
+    _GENERIC_READ = 0x80000000
+    _GENERIC_WRITE = 0x40000000
+    _FILE_SHARE_READ = 0x00000001
+    _OPEN_EXISTING = 3
+    _OPEN_ALWAYS = 4
+    _FILE_ATTRIBUTE_NORMAL = 0x80
+    _FILE_BEGIN = 0
+    _FILE_END = 2
+    _ERROR_SHARING_VIOLATION = 32
+    _INVALID_FILE_SIZE = 0xFFFFFFFF
+
+    _kernel32.CreateFileW.restype = wintypes.HANDLE
+    _kernel32.CreateFileW.argtypes = [
+        wintypes.LPCWSTR,
+        wintypes.DWORD,
+        wintypes.DWORD,
+        ctypes.c_void_p,
+        wintypes.DWORD,
+        wintypes.DWORD,
+        wintypes.HANDLE,
+    ]
+    _kernel32.ReadFile.restype = wintypes.BOOL
+    _kernel32.ReadFile.argtypes = [
+        wintypes.HANDLE,
+        ctypes.c_void_p,
+        wintypes.DWORD,
+        ctypes.POINTER(wintypes.DWORD),
+        ctypes.c_void_p,
+    ]
+    _kernel32.WriteFile.restype = wintypes.BOOL
+    _kernel32.WriteFile.argtypes = [
+        wintypes.HANDLE,
+        ctypes.c_void_p,
+        wintypes.DWORD,
+        ctypes.POINTER(wintypes.DWORD),
+        ctypes.c_void_p,
+    ]
+    _kernel32.GetFileSize.restype = wintypes.DWORD
+    _kernel32.GetFileSize.argtypes = [
+        wintypes.HANDLE,
+        ctypes.POINTER(wintypes.DWORD),
+    ]
+    _kernel32.SetFilePointer.restype = wintypes.DWORD
+    _kernel32.SetFilePointer.argtypes = [
+        wintypes.HANDLE,
+        wintypes.LONG,
+        ctypes.POINTER(wintypes.LONG),
+        wintypes.DWORD,
+    ]
+    _kernel32.CloseHandle.restype = wintypes.BOOL
+    _kernel32.CloseHandle.argtypes = [wintypes.HANDLE]
+
+    def _is_invalid_handle(handle) -> bool:
+        return ctypes.c_void_p(handle).value == ctypes.c_void_p(-1).value
+
+else:
+    import fcntl
+
+
+class LockedFile:
+    """A file opened with an exclusive write lock.
+
+    Usage::
+
+        f = LockedFile()
+        f.open(path)          # open + lock (read+write)
+        content = f.read()    # read entire content
+        f.write(data)         # append data (seeks to end first)
+        f.close()             # release lock + close fd
+
+    Unix:    fcntl.flock (advisory) — read-only callers that don't flock are unaffected.
+    Windows: CreateFileW with FILE_SHARE_READ — OS blocks other writers.
+    """
+
+    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.
+
+        Args:
+            path:   File to open and lock.
+            create: If True, create the file if it doesn't exist (bootstrap).
+
+        Raises:
+            FileLockError: If the file is locked by another process or not found.
+        """
+        if self._fd is not None:
+            return  # Already open (idempotent)
+
+        if sys.platform == "win32":
+            self._open_win32(path, create)
+        else:
+            self._open_unix(path, create)
+
+    def open_and_read(self, path: Path, create: bool = False) -> bytes:
+        """Open *path* with exclusive lock and read all content.
+
+        Combined operation for efficient use with asyncio.to_thread().
+        """
+        self.open(path, create=create)
+        return self.read()
+
+    def read(self) -> bytes:
+        """Read the entire file content from the beginning."""
+        if self._fd is None:
+            raise RuntimeError("LockedFile.read() called on a closed file")
+
+        if sys.platform == "win32":
+            return self._read_win32()
+        else:
+            return self._read_unix()
+
+    def write(self, data: bytes) -> None:
+        """Append *data* to the end of the file."""
+        if self._fd is None:
+            raise RuntimeError("LockedFile.write() called on a closed file")
+
+        if sys.platform == "win32":
+            self._write_win32(data)
+        else:
+            self._write_unix(data)
+
+    def size(self) -> int:
+        """Return current file size in bytes."""
+        if self._fd is None:
+            raise RuntimeError("LockedFile.size() called on a closed file")
+
+        if sys.platform == "win32":
+            size = _kernel32.GetFileSize(self._fd, None)
+            if size == _INVALID_FILE_SIZE:
+                raise OSError(
+                    f"GetFileSize failed: Windows error {ctypes.get_last_error()}"
+                )
+            return int(size)
+
+        current = os.lseek(self._fd, 0, os.SEEK_CUR)
+        end = os.lseek(self._fd, 0, os.SEEK_END)
+        os.lseek(self._fd, current, os.SEEK_SET)
+        return end
+
+    def close(self) -> None:
+        """Release the lock and close the file."""
+        if self._fd is None:
+            return
+        if sys.platform == "win32":
+            _kernel32.CloseHandle(self._fd)
+        else:
+            os.close(self._fd)
+        self._fd = None
+
+    @property
+    def is_open(self) -> bool:
+        return self._fd is not None
+
+    # -- Unix ----------------------------------------------------------------
+
+    def _open_unix(self, path: Path, create: bool) -> None:
+        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)
+        try:
+            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except OSError:
+            os.close(fd)
+            _fatal(
+                f"{path.resolve()}: database already locked by another instance",
+                db_path=path,
+            )
+        self._fd = fd
+
+    def _read_unix(self) -> bytes:
+        os.lseek(self._fd, 0, os.SEEK_SET)
+        chunks = []
+        while True:
+            chunk = os.read(self._fd, 1 << 20)  # 1 MiB
+            if not chunk:
+                break
+            chunks.append(chunk)
+        return b"".join(chunks)
+
+    def _write_unix(self, data: bytes) -> None:
+        os.lseek(self._fd, 0, os.SEEK_END)
+        os.write(self._fd, data)
+
+    # -- Windows -------------------------------------------------------------
+
+    def _open_win32(self, path: Path, create: bool) -> None:
+        disposition = _OPEN_ALWAYS if create else _OPEN_EXISTING
+        handle = _kernel32.CreateFileW(
+            str(path),
+            _GENERIC_READ | _GENERIC_WRITE,
+            _FILE_SHARE_READ,
+            None,
+            disposition,
+            _FILE_ATTRIBUTE_NORMAL,
+            None,
+        )
+        if _is_invalid_handle(handle):
+            err = ctypes.get_last_error()
+            if err == _ERROR_SHARING_VIOLATION:
+                _fatal(
+                    f"{path.resolve()}: database already locked by another instance",
+                    db_path=path,
+                )
+            _fatal(
+                f"Failed to open database {path.resolve()}: Windows error {err}",
+                db_path=path,
+            )
+        self._fd = handle
+
+    def _read_win32(self) -> bytes:
+        _kernel32.SetFilePointer(self._fd, 0, None, _FILE_BEGIN)
+        size = _kernel32.GetFileSize(self._fd, None)
+        if size == _INVALID_FILE_SIZE:
+            raise OSError(
+                f"GetFileSize failed: Windows error {ctypes.get_last_error()}"
+            )
+        if size == 0:
+            return b""
+        buf = ctypes.create_string_buffer(size)
+        bytes_read = wintypes.DWORD()
+        ok = _kernel32.ReadFile(self._fd, buf, size, ctypes.byref(bytes_read), None)
+        if not ok:
+            raise OSError(f"ReadFile failed: Windows error {ctypes.get_last_error()}")
+        return buf.raw[: bytes_read.value]
+
+    def _write_win32(self, data: bytes) -> None:
+        _kernel32.SetFilePointer(self._fd, 0, None, _FILE_END)
+        written = wintypes.DWORD()
+        ok = _kernel32.WriteFile(
+            self._fd,
+            data,
+            len(data),
+            ctypes.byref(written),
+            None,
+        )
+        if not ok:
+            raise OSError(f"WriteFile failed: Windows error {ctypes.get_last_error()}")

kanta/kanta.py

@@ -0,0 +1,206 @@
+"""JSONL persistence layer with background flush task."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+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.serialization import JsonSerializer, Serializer
+from kanta.transaction import transaction as _transaction
+
+T = TypeVar("T")
+
+
+class Kanta(Generic[T]):
+    """JSONL persistence layer for a msgspec.Struct database state.
+
+    The application defines its schema as a msgspec.Struct (e.g. ``Data``,
+    ``Project``).  The Kanta instance holds the live state as that struct type.
+    Internally it round-trips through plain dicts for diffing, replay,
+    and serialization.
+
+    A background task periodically flushes pending changes to disk.
+    Call `await kanta.open()` to start the background task,
+    and `await kanta.close()` to stop it.
+
+    All transactions are synchronous — they immediately affect the
+    in-memory ``kanta.data``. Persistence happens asynchronously in
+    the background (or via explicit ``await kanta.flush()``).
+
+    Usage::
+
+        class Data(msgspec.Struct):
+            users: dict[str, User] = {}
+
+        kanta = Kanta("data.db", Data())
+        await kanta.open()
+
+        with kanta.transaction(action="create_user") as data:
+            data.users["alice"] = User(name="Alice")
+
+        await kanta.close()
+    """
+
+    def __init__(
+        self,
+        filename: Path | str,
+        data: T,
+        *,
+        type: type[T] | None = None,
+        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.
+
+        Args:
+            filename: Path to the database file.
+            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.
+            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.
+            ValueError: If migration definitions are invalid.
+        """
+        active_serializer = serializer if serializer is not None else JsonSerializer()
+        data_type = type if type is not None else data.__class__
+
+        self._impl = KantaImpl(
+            serializer=active_serializer,
+            fatal_error=fatal_error,
+            filename=filename,
+            data=data,
+            type=data_type,
+            migrations=migrations,
+            migration_ctx=migration_ctx,
+            flush_interval=flush_interval,
+        )
+
+    @property
+    def data(self) -> T:
+        """Current in-memory state object.
+
+        Returns:
+            The live state instance of the configured ``type``.
+
+        Notes:
+            Mutations should only be performed via :meth:`transaction`
+            to ensure proper diffing and persistence.
+        """
+        return self._impl.data
+
+    @data.setter
+    def data(self, value: T) -> None:
+        """Replace the in-memory state object.
+
+        Notes:
+            Mutations should only be performed via :meth:`transaction`
+            to ensure proper diffing and persistence.
+
+        Args:
+            value: New state object instance.
+        """
+        self._impl.data = value
+
+    @property
+    def version(self) -> int:
+        """Current schema/database version.
+
+        Returns:
+            Integer version derived from migrations/replay state.
+        """
+        return self._impl.version
+
+    @property
+    def filename(self) -> Path:
+        """Database file path.
+
+        Returns:
+            Filesystem path used for persistence.
+        """
+        return self._impl.filename
+
+    async def open(self) -> None:
+        """Open the database file and start background persistence.
+
+        This loads existing records, applies configured migrations, and starts
+        the background flush task.
+
+        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()
+
+    async def __aenter__(self) -> Kanta[T]:
+        """Enter async context manager and open the database.
+
+        Returns:
+            The current ``Kanta`` instance itself.
+        """
+        await self.open()
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb) -> None:
+        """Exit async context manager and close the database.
+
+        Args:
+            exc_type: Exception type raised inside the context, if any.
+            exc: Exception instance raised inside the context, if any.
+            tb: Traceback for the exception, if any.
+        """
+        await self.close()
+
+    async def flush(self) -> None:
+        """Asynchronously flush pending change records to disk."""
+        await self._impl.flush()
+
+    def request_snapshot(self) -> None:
+        """Request a snapshot to be written on the next background iteration."""
+        self._impl.snapshot.request_force()
+
+    async def close(self) -> None:
+        """Stop background task, flush pending changes, and close file lock."""
+        await self._impl.close()
+
+    def transaction(
+        self,
+        action: str,
+        *,
+        user: str | None = None,
+        user_display: str | None = None,
+        resolver: Any = None,
+    ):
+        """Create a transactional mutation context manager.
+
+        Args:
+            action: Action label stored in the change record.
+            user: Optional user identifier stored in metadata.
+            user_display: Optional display name used for logging/resolution.
+            resolver: Optional callable for resolving identifiers in logs.
+
+        Returns:
+            A context manager yielding the live state object for mutation.
+
+        Notes:
+            On successful exit, a diff is queued for persistence.
+            If an exception is raised inside the context, in-memory changes are
+            rolled back.
+        """
+        return _transaction(
+            self._impl, action, user=user, user_display=user_display, resolver=resolver
+        )

kanta/kantaimpl.py

@@ -0,0 +1,137 @@
+"""Internal implementation for Kanta."""
+
+from __future__ import annotations
+
+import asyncio
+import copy
+import importlib
+import logging
+from datetime import UTC, datetime
+from typing import Any, Generic, TypeVar
+
+from kanta.exceptions import DatabaseError, DataIntegrityError, ReplayError
+from kanta.migrate import MigrationRegistry
+from kanta.persistence import PersistenceMixin
+from kanta.serialization import restore_data_in_place, struct_to_dict
+from kanta.serialization.base import replay
+
+_logger = logging.getLogger(__name__)
+
+T = TypeVar("T")
+
+
+class KantaImpl(PersistenceMixin, Generic[T]):
+    """Internal state and logic for Kanta."""
+
+    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)
+        super().__init__(**kwargs)
+        self.migration_registry: MigrationRegistry | None = None
+        if self.migrations is not None:
+            module = (
+                importlib.import_module(self.migrations)
+                if isinstance(self.migrations, str)
+                else self.migrations
+            )
+            self.migration_registry = MigrationRegistry.from_module(module)
+
+        self.in_transaction = False
+        self.transaction_snapshot: dict[str, Any] | None = None
+        self.opened = False
+
+        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:
+        """Open the database: load from disk, apply migrations, start background task."""
+        if self.opened:
+            raise DataIntegrityError(
+                "Kanta instance is already open",
+                db_path=self.filename,
+                action="open",
+            )
+
+        content = await asyncio.to_thread(
+            self.file.open_and_read,
+            self.filename,
+            create=True,
+        )
+
+        if content:
+            try:
+                rr = replay(
+                    content,
+                    framer=self.framer,
+                    decode=self.serializer.decode,
+                )
+            except ReplayError as e:
+                raise DatabaseError(
+                    f"{e}",
+                    db_path=self.filename,
+                    line_number=e.line_number,
+                    byte_pos=e.byte_pos,
+                    cause_type=type(e).__name__,
+                ) from e
+            except (OSError, ValueError, DatabaseError) as e:
+                raise DatabaseError(
+                    f"{e}",
+                    db_path=self.filename,
+                    cause_type=type(e).__name__,
+                ) from e
+            except Exception as e:
+                _logger.exception("Unexpected error loading database")
+                raise DatabaseError(
+                    f"{e}",
+                    db_path=self.filename,
+                    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
+                )
+
+            self.statedict = copy.deepcopy(rr.state)
+            self.data = restore_data_in_place(
+                self.data,
+                rr.state,
+                self.data_type,
+                serializer=self.serializer,
+            )
+            self.version = rr.version
+            normalized = struct_to_dict(self.data, serializer=self.serializer)
+            self.queue_change("migrate:msgspec", normalized)
+            self.snapshot.ts = (
+                datetime.fromtimestamp(rr.last_snapshot_mtime, UTC)
+                if rr.last_snapshot_mtime is not None
+                else None
+            )
+
+        self.opened = True
+
+        self.background_task = asyncio.create_task(self._background_loop())
+
+    async def close(self) -> None:
+        """Stop the background task, flush pending changes, and release the file lock."""
+        if not self.opened:
+            return
+
+        if self.background_task is not None:
+            self.background_task.cancel()
+            try:
+                await self.background_task
+            except asyncio.CancelledError:
+                pass
+            self.background_task = None
+
+        # Always run a final flush in case the background task never reached
+        # its cancellation handler.
+        await self.flush()
+
+        self.file.close()
+        self.opened = False

kanta/logging.py

@@ -0,0 +1,289 @@
+"""Database change logging with pretty-printed diffs.
+
+Provides a logger for JSONL database changes that formats diffs
+in a human-readable path.notation style with color coding.
+"""
+
+import logging
+import re
+import sys
+from collections.abc import Callable
+from typing import Any
+
+logger = logging.getLogger("kanta.changes")
+
+# Pattern to match control characters and bidirectional overrides
+_UNSAFE_CHARS = re.compile(
+    r"[\x00-\x1f\x7f-\x9f"
+    r"\u200e\u200f"
+    r"\u202a-\u202e"
+    r"\u2066-\u2069"
+    r"]"
+)
+
+# ANSI color codes
+_RESET = "\033[0m"
+_SEP = "\033[38;5;242m"  # Dark grey for separators
+_PATH_PREFIX = "\033[38;5;242m"  # Dark grey for path prefix
+_PATH_FINAL = "\033[38;5;250m"  # Default for final element
+_DELETE = "\033[1;31m"  # Red for deletions
+_ADD = "\033[0;32m"  # Green for additions
+_ACTION = "\033[1;34m"  # Bold blue for action name
+_USER = "\033[0;34m"  # Blue for user display
+
+
+def _format_value(
+    value: Any, max_len: int = 60, resolver: Callable[[str], str] | None = None
+) -> str:
+    """Format a value for display, truncating if needed."""
+    if value is None:
+        return "null"
+    if isinstance(value, bool):
+        return "true" if value else "false"
+    if isinstance(value, (int, float)):
+        return str(value)
+    if isinstance(value, str):
+        value = _UNSAFE_CHARS.sub("", value)
+        if resolver is not None:
+            resolved = resolver(value)
+            if resolved != value:
+                return resolved
+        if len(value) > max_len:
+            return value[: max_len - 3] + "..."
+        return value
+    if isinstance(value, dict):
+        if not value:
+            return "{}"
+        all_true = all(v is True for v in value.values())
+        parts = []
+        for k, v in value.items():
+            key_display = resolver(k) if resolver is not None else k
+            if all_true:
+                parts.append(key_display)
+            else:
+                val_display = _format_value(v, max_len=30, resolver=resolver)
+                parts.append(f"{key_display}: {val_display}")
+        return "{" + ", ".join(parts) + "}"
+    if isinstance(value, list):
+        if not value:
+            return "[]"
+        parts = [_format_value(v, max_len=30, resolver=resolver) for v in value]
+        return "[" + ", ".join(parts) + "]"
+    text = str(value)
+    if len(text) > max_len:
+        text = text[: max_len - 3] + "..."
+    return text
+
+
+def _format_path(path: list[str], resolver: Callable[[str], str] | None = None) -> str:
+    """Format a path as dot notation with prefix in dark grey, final in default."""
+    if not path:
+        return ""
+    if resolver is not None:
+        path = [resolver(p) for p in path]
+    if len(path) == 1:
+        return f"{_PATH_FINAL}{path[0]}{_RESET}"
+    prefix = ".".join(path[:-1])
+    final = path[-1]
+    return f"{_PATH_PREFIX}{prefix}.{_RESET}{_PATH_FINAL}{final}{_RESET}"
+
+
+def _get_nested(data: dict | None, path: list[str]) -> Any:
+    """Get a nested value from a dict by path, or None if not found."""
+    if data is None:
+        return None
+    current = data
+    for key in path:
+        if not isinstance(current, dict) or key not in current:
+            return None
+        current = current[key]
+    return current
+
+
+def _collect_changes(
+    diff: dict,
+    path: list[str],
+    changes: list[tuple[str, list[str], Any]],
+    previous: dict | None,
+) -> None:
+    """Recursively collect changes from a diff into a flat list.
+
+    Each change is a tuple of (change_type, path, new_value).
+    change_type is one of: 'add', 'update', 'delete'
+    """
+    if not isinstance(diff, dict):
+        existed = _get_nested(previous, path) is not None
+        changes.append(("update" if existed else "add", path, diff))
+        return
+
+    for key, value in diff.items():
+        if key == "$delete":
+            if isinstance(value, list):
+                for deleted_key in value:
+                    changes.append(("delete", path + [str(deleted_key)], None))
+            else:
+                changes.append(("delete", path + [str(value)], None))
+        elif key == "$replace":
+            old_collection = _get_nested(previous, path)
+            old_keys = (
+                set(old_collection.keys())
+                if isinstance(old_collection, dict)
+                else set()
+            )
+            new_keys = set(value.keys()) if isinstance(value, dict) else set()
+            for deleted_key in old_keys - new_keys:
+                changes.append(("delete", path + [str(deleted_key)], None))
+            if isinstance(value, dict):
+                for rkey, rval in value.items():
+                    existed = rkey in old_keys
+                    changes.append(
+                        ("update" if existed else "add", path + [str(rkey)], rval)
+                    )
+            elif value or not old_keys:
+                changes.append(
+                    ("update" if old_collection is not None else "add", path, value)
+                )
+        elif isinstance(key, str) and key.startswith("$"):
+            changes.append(("add", path, {key: value}))
+        else:
+            new_path = path + [str(key)]
+            existed = _get_nested(previous, new_path) is not None
+            if existed:
+                _collect_changes(value, new_path, changes, previous)
+            else:
+                changes.append(("add", new_path, value))
+
+
+def _format_change_lines(
+    change_type: str,
+    path: list[str],
+    value: Any,
+    resolver: Callable[[str], str] | None = None,
+) -> list[str]:
+    """Format a single change as one or more lines."""
+
+    def fmt_value(v: Any, child_path: list[str]) -> str:
+        return _format_value(v, resolver=resolver)
+
+    formatted_path = list(path)
+    if resolver is not None:
+        formatted_path = [resolver(p) for p in formatted_path]
+
+    if change_type == "delete":
+        if len(formatted_path) == 1:
+            return [f"  {_DELETE}{formatted_path[0]} ✗{_RESET}"]
+        prefix = ".".join(formatted_path[:-1])
+        final = formatted_path[-1]
+        return [f"  {_PATH_PREFIX}{prefix}.{_RESET}{_DELETE}{final} ✗{_RESET}"]
+
+    if change_type == "add":
+        if isinstance(value, dict) and value:
+            lines = []
+            if len(formatted_path) == 1:
+                lines.append(f"  {_ADD}{formatted_path[0]}{_RESET} {_SEP}={_RESET}")
+            else:
+                prefix = ".".join(formatted_path[:-1])
+                final = formatted_path[-1]
+                lines.append(
+                    f"  {_PATH_PREFIX}{prefix}.{_RESET}{_ADD}{final}{_RESET} {_SEP}={_RESET}"
+                )
+            formatted_items = []
+            for k, v in value.items():
+                k_display = resolver(k) if resolver is not None else k
+                v_str = fmt_value(v, path + [k])
+                formatted_items.append((k_display, v_str))
+            max_key_len = max(len(k) for k, _ in formatted_items)
+            field_width = max(max_key_len, 12)
+            for k_display, v_str in formatted_items:
+                padding = " " * (field_width - len(k_display))
+                lines.append(f"    {k_display}{_SEP}:{_RESET}{padding} {v_str}")
+            return lines
+        else:
+            value_str = fmt_value(value, path)
+            if len(formatted_path) == 1:
+                return [
+                    f"  {_ADD}{formatted_path[0]}{_RESET} {_SEP}={_RESET} {value_str}"
+                ]
+            prefix = ".".join(formatted_path[:-1])
+            final = formatted_path[-1]
+            return [
+                f"  {_PATH_PREFIX}{prefix}.{_RESET}{_ADD}{final}{_RESET} {_SEP}={_RESET} {value_str}"
+            ]
+
+    value_str = fmt_value(value, path)
+    path_str = _format_path(path, resolver=resolver)
+    return [f"  {path_str} {_SEP}={_RESET} {value_str}"]
+
+
+def format_diff(
+    diff: dict,
+    previous: dict | None = None,
+    resolver: Callable[[str], str] | None = None,
+) -> list[str]:
+    """Format a JSON diff as human-readable lines.
+
+    Args:
+        diff: The JSON diff dict.
+        previous: The previous state dict (for determining add vs update).
+        resolver: Optional callable to resolve path components (e.g. UUID→name).
+
+    Returns a list of formatted lines (without newlines).
+    """
+    changes: list[tuple[str, list[str], Any]] = []
+    _collect_changes(diff, [], changes, previous)
+    if not changes:
+        return []
+    lines = []
+    for change_type, path, value in changes:
+        lines.extend(_format_change_lines(change_type, path, value, resolver))
+    return lines
+
+
+def format_action_header(action: str, user_display: str | None = None) -> str:
+    """Format the action header line."""
+    action_str = f"{_ACTION}{action}{_RESET}"
+    if user_display:
+        user_str = f"{_USER}{user_display}{_RESET}"
+        return f"{action_str} by {user_str}"
+    return action_str
+
+
+def log_change(
+    action: str,
+    diff: dict,
+    user_display: str | None = None,
+    previous: dict | None = None,
+    resolver: Callable[[str], str] | None = None,
+) -> None:
+    """Log a database change with pretty-printed diff.
+
+    Args:
+        action: The action name (e.g., "login", "admin:delete_user").
+        diff: The JSON diff dict.
+        user_display: Optional display name of the user who performed the action.
+        previous: The previous state dict (for determining add vs update).
+        resolver: Optional callable to resolve path components (e.g. UUID→name).
+    """
+    header = format_action_header(action, user_display)
+    diff_lines = format_diff(diff, previous, resolver)
+
+    if not diff_lines:
+        logger.info(header)
+        return
+
+    if len(diff_lines) == 1:
+        logger.info(f"{header}{diff_lines[0]}")
+    else:
+        logger.info(header)
+        for line in diff_lines:
+            logger.info(line)
+
+
+def configure_logging() -> None:
+    """Configure the database logger to output to stderr without prefix."""
+    if not logger.handlers:
+        handler = logging.StreamHandler(sys.stderr)
+        handler.setFormatter(logging.Formatter("%(message)s"))
+        logger.addHandler(handler)
+    logger.setLevel(logging.INFO)
+    logger.propagate = False

kanta/migrate.py

@@ -0,0 +1,117 @@
+"""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/persistence.py

@@ -0,0 +1,215 @@
+"""Persistence mixin for KantaImpl."""
+
+from __future__ import annotations
+
+import asyncio
+import copy
+import logging
+import threading
+from collections import deque
+from collections.abc import Callable
+from datetime import 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.serialization import JsonSerializer, Serializer
+from kanta.serialization.framing import Framer
+from kanta.snapshot import SnapshotState
+
+_logger = logging.getLogger(__name__)
+
+
+class PersistenceMixin:
+    """Persistence-related behavior for Kanta implementations."""
+
+    filename: Path
+    file: LockedFile
+    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
+    background_error: DatabaseError | None
+    flush_interval: float
+    version: int
+    opened: bool
+
+    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.background_error = None
+        self.flush_interval = flush_interval
+        self.version = 0
+
+    async def _background_loop(self) -> None:
+        """Background task that periodically flushes changes to disk."""
+        while True:
+            try:
+                await asyncio.sleep(self.flush_interval)
+                await self.flush()
+                self.maybe_snapshot()
+            except asyncio.CancelledError:
+                await self.flush()
+                self.maybe_snapshot()
+                break
+            except DatabaseError as e:
+                self.background_error = e
+                if self.fatal_error is not None:
+                    try:
+                        self.fatal_error(e)
+                    except Exception as callback_error:
+                        _logger.exception(
+                            "Background error callback failed: %s", callback_error
+                        )
+                _logger.error("Background flush loop stopped: %s", e)
+                break
+
+    def maybe_snapshot(self) -> None:
+        """Evaluate and possibly write a snapshot from current state."""
+        self.snapshot.maybe_write(self.file, self.version, self.statedict)
+
+    def queue_change(
+        self,
+        action: str,
+        current: dict,
+        user: str | None = None,
+        m: datetime | None = None,
+    ) -> None:
+        """Queue a change record internally (thread-safe)."""
+        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,
+                )
+            )
+        self.statedict = copy.deepcopy(current)
+
+    def flush_sync(self) -> None:
+        """Synchronously flush all pending changes to disk."""
+        if not self.opened:
+            raise DataIntegrityError(
+                "Kanta instance must be opened before flush_sync",
+                db_path=self.filename,
+                action="flush_sync",
+            )
+
+        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.file.is_open:
+            self.file.open(self.filename, create=True)
+
+        try:
+            base_offset = self.file.size()
+            records = []
+            running_size = 0
+            for change in changes_to_write:
+                framed = self.framer.frame_change(
+                    self.serializer.encode(change),
+                    record_offset=base_offset + running_size,
+                )
+                records.append(framed)
+                running_size += len(framed)
+            if not records:
+                with self.pending_lock:
+                    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()
+        except OSError as e:
+            _logger.error("Failed to flush database: %s", e)
+            self.flush_failed = True
+            raise DatabaseError(
+                f"Failed to flush database: {e}",
+                db_path=self.filename,
+                cause_type=type(e).__name__,
+            ) from e
+
+    async def flush(self) -> None:
+        """Write all pending changes to disk via threadpool-backed file I/O."""
+        if not self.opened:
+            raise DataIntegrityError(
+                "Kanta instance must be opened before flush",
+                db_path=self.filename,
+                action="flush",
+            )
+
+        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.file.is_open:
+            await asyncio.to_thread(self.file.open, self.filename, create=True)
+
+        try:
+            base_offset = await asyncio.to_thread(self.file.size)
+            records = []
+            running_size = 0
+            for change in changes_to_write:
+                framed = self.framer.frame_change(
+                    self.serializer.encode(change),
+                    record_offset=base_offset + running_size,
+                )
+                records.append(framed)
+                running_size += len(framed)
+            if not records:
+                with self.pending_lock:
+                    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()
+        except OSError as e:
+            _logger.error("Failed to flush database: %s", e)
+            self.flush_failed = True
+            raise DatabaseError(
+                f"Failed to flush database: {e}",
+                db_path=self.filename,
+                cause_type=type(e).__name__,
+            ) from e

kanta/serialization/__init__.py

@@ -0,0 +1,59 @@
+"""Serializer package with pluggable format classes."""
+
+from __future__ import annotations
+
+from typing import Any, TypeVar
+
+from .base import ReplayResult, Serializer
+from .framing import BinFramer, Framer, LineFramer
+from .json import JsonSerializer
+from .msgpack import MsgPackSerializer
+
+T = TypeVar("T")
+
+_DEFAULT_SERIALIZER = JsonSerializer()
+
+
+def struct_to_dict(obj: T, serializer: Serializer | None = None) -> dict[str, Any]:
+    """Convert a struct instance to plain builtins with the selected serializer."""
+    active = serializer or _DEFAULT_SERIALIZER
+    return active.decode(active.encode(obj), type=dict[str, Any])
+
+
+def dict_to_struct(
+    d: dict[str, Any],
+    data_type: type[T],
+    serializer: Serializer | None = None,
+) -> T:
+    """Decode plain builtins back into the configured struct type."""
+    active = serializer or _DEFAULT_SERIALIZER
+    return active.decode(active.encode(d), type=data_type)
+
+
+def restore_data_in_place(
+    data: T,
+    snapshot_dict: dict[str, Any],
+    data_type: type[T],
+    serializer: Serializer | None = None,
+) -> T:
+    """Restore data from snapshot while preserving object identity when possible."""
+    restored = dict_to_struct(snapshot_dict, data_type, serializer=serializer)
+    if type(restored) is not type(data):
+        return restored
+    for field_name in getattr(restored, "__struct_fields__", ()):
+        setattr(data, field_name, getattr(restored, field_name))
+    return data
+
+
+__all__ = [
+    "Framer",
+    "JsonSerializer",
+    "BinFramer",
+    "LineFramer",
+    "MsgPackSerializer",
+    "ReplayResult",
+    "Serializer",
+    "dict_to_struct",
+    "restore_data_in_place",
+    "struct_to_dict",
+]

kanta/serialization/base.py

@@ -0,0 +1,158 @@
+"""Serializer interfaces for Kanta journal formats."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from datetime import datetime
+from typing import Any, Protocol, TypeVar
+
+import msgspec
+
+from kanta.exceptions import ReplayError
+from kanta.kanta.structs import ChangeRecord, Snapshot
+from kanta.serialization.framing import Framer
+
+T = TypeVar("T")
+
+
+class ReplayResult:
+    """Result of replaying serialized database bytes."""
+
+    def __init__(
+        self,
+        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
+
+
+class Serializer(Protocol):
+    """Format-specific serializer contract used by persistence and loading."""
+
+    framer_cls: type[Framer]
+
+    def encode(self, obj: Any) -> bytes:
+        """Encode one record payload."""
+        raise NotImplementedError
+
+    def decode(self, payload: bytes, *, type: type[T]) -> T:
+        """Decode one record payload to the provided type."""
+        raise NotImplementedError
+
+
+def replay(
+    data: bytes,
+    *,
+    framer: Framer,
+    decode: Callable[..., Any],
+) -> ReplayResult:
+    """Rebuild database state from raw file bytes using the supplied framer."""
+    snap_payload, offset, snap_byte_pos = framer.scan_last_snapshot(data)
+
+    state: dict[str, Any] = {}
+    version = 0
+    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:
+            snap = decode(snap_payload, type=Snapshot)
+        except msgspec.DecodeError as e:
+            raise ReplayError(
+                "invalid snapshot record",
+                line_number=0,
+                byte_pos=snap_byte_pos,
+                record_type="snapshot",
+            ) from e
+        state = snap.state
+        version = snap.v
+        last_snapshot_mtime = snap.ts.timestamp()
+        m = snap.m
+
+    for is_snapshot, payload, line_number, byte_pos in framer.iter_records(
+        data, offset
+    ):
+        if is_snapshot:
+            try:
+                snap = decode(payload, type=Snapshot)
+            except Exception as e:
+                raise ReplayError(
+                    "invalid snapshot record",
+                    line_number=line_number,
+                    byte_pos=byte_pos,
+                    record_type="snapshot",
+                ) from e
+            last_snapshot_mtime = snap.ts.timestamp()
+            if snap.m is not None:
+                m = snap.m
+            continue
+
+        try:
+            change = decode(payload, type=ChangeRecord)
+        except msgspec.DecodeError:
+            raise ReplayError(
+                "invalid record",
+                line_number=line_number,
+                byte_pos=byte_pos,
+                record_type="change",
+            ) from None
+
+        if change.a.startswith("migrate"):
+            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,
+    )
+
+
+def _patch_state(state: dict, diff: dict) -> dict:
+    return _apply_diff(state, diff)
+
+
+def _apply_diff(state: dict, diff: dict) -> dict:
+    if not isinstance(diff, dict):
+        return diff
+
+    result = dict(state) if isinstance(state, dict) else state
+    if not isinstance(result, dict):
+        result = {}
+
+    for key, value in diff.items():
+        if key == "$replace":
+            return value
+        if key == "$delete":
+            if isinstance(value, list):
+                for k in value:
+                    result.pop(k, None)
+            else:
+                result.pop(value, None)
+            continue
+        if isinstance(value, dict):
+            old = result.get(key, {})
+            if not isinstance(old, dict):
+                old = {}
+            result[key] = _apply_diff(old, value)
+            continue
+        result[key] = value
+
+    return result

kanta/serialization/framing.py

@@ -0,0 +1,304 @@
+"""Framing strategies for on-disk record storage.
+
+Framers handle how encoded payloads are delimited and scanned on disk,
+independent of the payload encoding (JSON, MessagePack, etc.).
+"""
+
+from __future__ import annotations
+
+import secrets
+from collections.abc import Iterator
+from typing import Protocol
+
+from kanta.exceptions import ReplayError
+
+try:
+    from blake3 import blake3
+except ImportError:
+    raise ImportError("Install kanta[bin] for binary framing / msgpack support.")
+
+
+class Framer(Protocol):
+    """Handles on-disk record framing and scanning, independent of payload encoding."""
+
+    def frame_change(self, payload: bytes, *, record_offset: int = 0) -> bytes:
+        """Wrap an encoded change payload for writing."""
+        raise NotImplementedError
+
+    def frame_snapshot(self, payload: bytes, *, record_offset: int = 0) -> bytes:
+        """Wrap an encoded snapshot payload for writing."""
+        raise NotImplementedError
+
+    def scan_last_snapshot(self, data: bytes) -> tuple[bytes | None, int, int]:
+        """Find the last snapshot payload and the byte offset to resume iteration from.
+
+        Returns:
+            (snapshot_payload_or_None, resume_offset, snapshot_byte_pos)
+        """
+        raise NotImplementedError
+
+    def iter_records(
+        self, data: bytes, offset: int = 0
+    ) -> Iterator[tuple[bool, bytes, int, int]]:
+        """Iterate records from *offset* onward.
+
+        Yields (is_snapshot, payload, line_number, byte_pos) tuples.
+        ``line_number`` is 1-based for text framers and 0 for binary framers.
+        Raises ReplayError on corruption.
+        """
+        raise NotImplementedError
+
+
+class LineFramer:
+    """Line-delimited framer for text-based formats such as JSONL."""
+
+    SNAPSHOT_PREFIX = b"SNAPSHOT "
+
+    def frame_change(self, payload: bytes, *, record_offset: int = 0) -> bytes:
+        _ = record_offset
+        return payload + b"\n"
+
+    def frame_snapshot(self, payload: bytes, *, record_offset: int = 0) -> bytes:
+        _ = record_offset
+        return self.SNAPSHOT_PREFIX + payload + b"\n"
+
+    def scan_last_snapshot(self, data: bytes) -> tuple[bytes | None, int, int]:
+        marker = b"\n" + self.SNAPSHOT_PREFIX
+        pos = data.rfind(marker)
+        if pos != -1:
+            pos += 1
+        elif data.startswith(self.SNAPSHOT_PREFIX):
+            pos = 0
+        else:
+            return None, 0, 0
+
+        end = data.find(b"\n", pos)
+        if end == -1:
+            line_number = data[:pos].count(b"\n") + 1
+            raise ReplayError(
+                "incomplete snapshot line at end of file",
+                line_number=line_number,
+                byte_pos=pos,
+                record_type="snapshot",
+            )
+
+        payload = data[pos + len(self.SNAPSHOT_PREFIX) : end]
+        return payload, end + 1, pos
+
+    def iter_records(
+        self, data: bytes, offset: int = 0
+    ) -> Iterator[tuple[bool, bytes, int, int]]:
+        idx = offset
+        line_number = data[:offset].count(b"\n") + 1
+        while idx < len(data):
+            line_start = idx
+            end = data.find(b"\n", idx)
+            if end == -1:
+                raw = data[idx:]
+                idx = len(data)
+            else:
+                raw = data[idx:end]
+                idx = end + 1
+
+            line = raw.strip()
+            if line:
+                is_snapshot = line.startswith(self.SNAPSHOT_PREFIX)
+                payload = line[len(self.SNAPSHOT_PREFIX) :] if is_snapshot else line
+                yield is_snapshot, payload, line_number, line_start
+            line_number += 1
+
+
+class BinFramer:
+    """Length-prefixed binary framer for formats such as MessagePack.
+
+    First 4 bytes of the file are a per-file random sync seed.
+
+    Change frame::
+
+        <4-byte bitwise-not sync seed>
+        <4-byte little-endian length>
+        <8-byte checksum>
+        <payload>
+
+    Snapshot frame::
+
+        <4-byte sync seed>
+        <4-byte little-endian length>
+        <8-byte checksum>
+        <payload>
+
+    The checksum is BLAKE3 keyed by the per-file sync seed (zero-padded to
+    32 bytes), truncated to 8 bytes.
+    """
+
+    _SYNC_SIZE = 4
+    _LEN_SIZE = 4
+    _OFFSET_SIZE = 8
+    _CHECKSUM_SIZE = 8
+
+    def __init__(self) -> None:
+        self.set_sync(secrets.token_bytes(self._SYNC_SIZE))
+
+    def set_sync(self, value: bytes) -> None:
+        if len(value) != self._SYNC_SIZE:
+            raise ValueError("binary sync seed must be exactly 4 bytes")
+        self._sync_snapshot = value
+        self._sync_change = bytes((~b) & 0xFF for b in value)
+
+    def _ensure_sync_seed_for_write(self, record_offset: int) -> tuple[bytes, int]:
+        if record_offset == 0:
+            return self._sync_snapshot, self._SYNC_SIZE
+        return b"", record_offset
+
+    def _load_sync_seed_from_data(self, data: bytes) -> None:
+        if not data:
+            return
+        if len(data) < self._SYNC_SIZE:
+            raise ReplayError("missing binary sync header", line_number=0, byte_pos=0)
+        self.set_sync(data[: self._SYNC_SIZE])
+
+    def _checksum(
+        self, *, is_snapshot: bool, record_offset: int, payload_len: int, payload: bytes
+    ) -> bytes:
+        domain = b"SNAPSHOT" if is_snapshot else b"DIFFRECD"
+        offset_part = record_offset.to_bytes(self._OFFSET_SIZE, "little")
+        h = blake3(
+            payload, key=b"Kanta blake3" + domain + self._sync_snapshot + offset_part
+        )
+        return h.digest(length=self._CHECKSUM_SIZE)
+
+    def frame_change(self, payload: bytes, *, record_offset: int = 0) -> bytes:
+        header, effective_offset = self._ensure_sync_seed_for_write(record_offset)
+        payload_len = len(payload)
+        checksum = self._checksum(
+            is_snapshot=False,
+            record_offset=effective_offset,
+            payload_len=payload_len,
+            payload=payload,
+        )
+        framed = (
+            self._sync_change
+            + payload_len.to_bytes(self._LEN_SIZE, "little")
+            + checksum
+            + payload
+        )
+        return header + framed
+
+    def frame_snapshot(self, payload: bytes, *, record_offset: int = 0) -> bytes:
+        header, effective_offset = self._ensure_sync_seed_for_write(record_offset)
+        payload_len = len(payload)
+        checksum = self._checksum(
+            is_snapshot=True,
+            record_offset=effective_offset,
+            payload_len=payload_len,
+            payload=payload,
+        )
+        framed = (
+            self._sync_snapshot
+            + payload_len.to_bytes(self._LEN_SIZE, "little")
+            + checksum
+            + payload
+        )
+        return header + framed
+
+    def scan_last_snapshot(self, data: bytes) -> tuple[bytes | None, int, int]:
+        if not data:
+            return None, 0, 0
+        self._load_sync_seed_from_data(data)
+        snapshot_payload: bytes | None = None
+        snapshot_resume_offset = self._SYNC_SIZE
+        snapshot_byte_pos = 0
+        for (
+            is_snapshot,
+            payload,
+            next_offset,
+            record_offset,
+        ) in self._iter_records_internal(data, self._SYNC_SIZE):
+            if is_snapshot:
+                snapshot_payload = payload
+                snapshot_resume_offset = next_offset
+                snapshot_byte_pos = record_offset
+        return snapshot_payload, snapshot_resume_offset, snapshot_byte_pos
+
+    def _iter_records_internal(
+        self, data: bytes, start_offset: int
+    ) -> Iterator[tuple[bool, bytes, int, int]]:
+        if not data:
+            return
+        self._load_sync_seed_from_data(data)
+
+        idx = start_offset
+        while idx < len(data):
+            record_offset = idx
+            if idx + self._SYNC_SIZE > len(data):
+                raise ReplayError(
+                    "incomplete frame at end of file",
+                    line_number=0,
+                    byte_pos=record_offset,
+                )
+
+            frame_seed = data[idx : idx + self._SYNC_SIZE]
+            if frame_seed == self._sync_snapshot:
+                is_snapshot = True
+            elif frame_seed == self._sync_change:
+                is_snapshot = False
+            else:
+                raise ReplayError(
+                    "invalid frame marker",
+                    line_number=0,
+                    byte_pos=record_offset,
+                )
+            idx += self._SYNC_SIZE
+
+            min_record = self._LEN_SIZE + self._CHECKSUM_SIZE
+            if idx + min_record > len(data):
+                raise ReplayError(
+                    "incomplete frame at end of file",
+                    line_number=0,
+                    byte_pos=record_offset,
+                    record_type="snapshot" if is_snapshot else "change",
+                )
+
+            payload_len = int.from_bytes(data[idx : idx + self._LEN_SIZE], "little")
+            idx += self._LEN_SIZE
+
+            checksum = data[idx : idx + self._CHECKSUM_SIZE]
+            idx += self._CHECKSUM_SIZE
+
+            if idx + payload_len > len(data):
+                raise ReplayError(
+                    "incomplete frame at end of file",
+                    line_number=0,
+                    byte_pos=record_offset,
+                    record_type="snapshot" if is_snapshot else "change",
+                )
+
+            payload = data[idx : idx + payload_len]
+            expected = self._checksum(
+                is_snapshot=is_snapshot,
+                record_offset=record_offset,
+                payload_len=payload_len,
+                payload=payload,
+            )
+            if checksum != expected:
+                raise ReplayError(
+                    "invalid frame checksum",
+                    line_number=0,
+                    byte_pos=record_offset,
+                    record_type="snapshot" if is_snapshot else "change",
+                )
+
+            idx += payload_len
+            yield is_snapshot, payload, idx, record_offset
+
+    def iter_records(
+        self, data: bytes, offset: int = 0
+    ) -> Iterator[tuple[bool, bytes, int, int]]:
+        if not data:
+            return
+        self._load_sync_seed_from_data(data)
+        start = self._SYNC_SIZE if offset == 0 else offset
+        for is_snapshot, payload, _, record_offset in self._iter_records_internal(
+            data, start
+        ):
+            yield is_snapshot, payload, 0, record_offset

kanta/serialization/json.py

@@ -0,0 +1,23 @@
+"""JSON serializer implementation."""
+
+from __future__ import annotations
+
+from typing import Any, TypeVar
+
+import msgspec
+
+from kanta.serialization.framing import LineFramer
+
+T = TypeVar("T")
+
+
+class JsonSerializer:
+    """Line-based JSON serializer."""
+
+    framer_cls = LineFramer
+
+    def encode(self, obj: Any) -> bytes:
+        return msgspec.json.encode(obj)
+
+    def decode(self, payload: bytes, *, type: type[T]) -> T:
+        return msgspec.json.decode(payload, type=type)

kanta/serialization/msgpack.py

@@ -0,0 +1,23 @@
+"""MessagePack serializer implementation."""
+
+from __future__ import annotations
+
+from typing import Any, TypeVar
+
+import msgspec
+
+from kanta.serialization.framing import BinFramer
+
+T = TypeVar("T")
+
+
+class MsgPackSerializer:
+    """Binary serializer using MessagePack format."""
+
+    framer_cls = BinFramer
+
+    def encode(self, obj: Any) -> bytes:
+        return msgspec.msgpack.encode(obj)
+
+    def decode(self, payload: bytes, *, type: type[T]) -> T:
+        return msgspec.msgpack.decode(payload, type=type)

kanta/snapshot.py

@@ -0,0 +1,65 @@
+"""Internal snapshot behavior and state."""
+
+from __future__ import annotations
+
+import logging
+from datetime import UTC, datetime
+
+from kanta.kanta.structs import Snapshot
+from kanta.serialization import JsonSerializer, Serializer
+from kanta.serialization.framing import Framer, LineFramer
+
+_logger = logging.getLogger(__name__)
+
+MINDIFFS = 100
+
+
+class SnapshotState:
+    """Internal snapshot counters and write policy."""
+
+    def __init__(
+        self,
+        min_diffs: int = MINDIFFS,
+        serializer: Serializer | None = None,
+        framer: Framer | None = None,
+    ) -> None:
+        self.ts: datetime | None = None
+        self.changes: int = 0
+        self._force_pending: bool = False
+        self._min_diffs = min_diffs
+        self._serializer = serializer if serializer is not None else JsonSerializer()
+        self._framer = framer if framer is not None else LineFramer()
+
+    def request_force(self) -> None:
+        """Force snapshot write on next check."""
+        self._force_pending = True
+
+    def record_changes(self, count: int) -> None:
+        self.changes += count
+
+    def maybe_write(self, file, version: int, state: dict) -> None:
+        """Write snapshot when thresholds/time policy allows it."""
+        if self.changes < self._min_diffs:
+            return
+        force = self._force_pending
+        now = datetime.now(UTC)
+        if not force and now.weekday() != 6:  # 6 = Sunday
+            return
+        sunday_midnight = now.replace(hour=0, minute=0, second=0, microsecond=0)
+        if not force and self.ts is not None and self.ts >= sunday_midnight:
+            return
+        if not file.is_open:
+            return
+        try:
+            self._write(file, version, state, now)
+            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:
+        """Write a snapshot and update internal state."""
+        payload = self._serializer.encode(Snapshot(ts=now, v=version, state=state))
+        record_offset = file.size() if hasattr(file, "size") else 0
+        file.write(self._framer.frame_snapshot(payload, record_offset=record_offset))
+        self.changes = 0
+        self.ts = now

kanta/structs.py

@@ -0,0 +1,35 @@
+"""On-disk record structures for the JSONL log."""
+
+from datetime import UTC, datetime
+from typing import Any
+
+import msgspec
+
+
+class ChangeRecord(msgspec.Struct, omit_defaults=True, kw_only=True):
+    """A single change record in the JSONL log.
+
+    Attributes:
+        ts: Timestamp of the change.
+        a: Action name (e.g., "sync", "migrate", "create_user").
+        v: Schema version after this change.
+        u: User/actor identifier (None for system operations).
+        m: Last real (non-migration) modification time, carried forward.
+        diff: The jsondiff patch representing the change.
+    """
+
+    ts: datetime = msgspec.field(default_factory=lambda: datetime.now(UTC))
+    a: str = ""
+    v: int = 0
+    u: str | None = None
+    m: datetime | None = None
+    diff: dict
+
+
+class Snapshot(msgspec.Struct, omit_defaults=True):
+    """Full state snapshot embedded in the JSONL log."""
+
+    ts: datetime
+    v: int
+    state: dict[str, Any]
+    m: datetime | None = None

kanta/transaction.py

@@ -0,0 +1,76 @@
+"""Transaction context manager for Kanta."""
+
+from __future__ import annotations
+
+import logging
+from contextlib import contextmanager
+from typing import Any
+
+from kanta.diff import compute_diff
+from kanta.exceptions import DataIntegrityError
+from kanta.logging import log_change
+from kanta.serialization import restore_data_in_place, struct_to_dict
+
+_logger = logging.getLogger(__name__)
+
+
+@contextmanager
+def transaction(
+    impl,
+    action: str,
+    *,
+    user: str | None = None,
+    user_display: str | None = None,
+    resolver: Any = None,
+):
+    """Wrap writes in a transaction and yield the live db object."""
+    if impl.in_transaction:
+        raise RuntimeError(
+            "Nested or simultaneous transactions are not supported "
+            "(don't await inside transactions)."
+        )
+
+    current_dict = struct_to_dict(impl.data, serializer=impl.serializer)
+
+    if current_dict != impl.statedict:
+        is_bootstrap = action in {"bootstrap"}
+        if not (is_bootstrap and not impl.statedict):
+            diff = compute_diff(impl.statedict, current_dict)
+            if diff:
+                _logger.critical(
+                    "Database state modified outside of transaction! "
+                    "This indicates a bug where changes occurred without a transaction wrapper.\n"
+                    "Changes detected: %s",
+                    diff,
+                )
+                raise DataIntegrityError(
+                    "Database state modified outside of transaction",
+                    db_path=impl.db_path,
+                    action=action,
+                    diff=diff,
+                )
+
+    impl.in_transaction = True
+    impl.transaction_snapshot = current_dict
+
+    try:
+        yield impl.data
+        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
+    except Exception:
+        _logger.warning("Transaction '%s' failed, rolling back changes", action)
+        if impl.transaction_snapshot is not None:
+            impl.data = restore_data_in_place(
+                impl.data,
+                impl.transaction_snapshot,
+                impl.data_type,
+                serializer=impl.serializer,
+            )
+        raise
+    finally:
+        impl.in_transaction = False
+        impl.transaction_snapshot = None

main.py

@@ -0,0 +1,6 @@
+def main():
+    print("Hello from kanta!")
+
+
+if __name__ == "__main__":
+    main()

pyproject.toml

@@ -0,0 +1,42 @@
+[build-system]
+requires = [
+    "hatchling>=1.25.0",
+    "hatch-vcs>=0.4.0",
+]
+build-backend = "hatchling.build"
+
+[project]
+name = "kanta"
+dynamic = ["version"]
+description = "Kanta No-SQL database for async Python and frameworks such as FastAPI and Sanic"
+authors = [
+    { name = "Leo Vasanko"},
+]
+keywords = ["kantadb"]
+readme = "README.md"
+requires-python = ">=3.11"
+dependencies = [
+    "blake3>=1.0.8",
+    "jsondiff>=2.2.1",
+    "msgspec>=0.20.0",
+]
+
+[project.optional-dependencies]
+bin = [
+    "blake3>=1.0.8",
+]
+
+[project.urls]
+Repository = "https://git.zi.fi/LeoVasanko/kanta"
+
+[dependency-groups]
+dev = [
+    "pytest>=9.0.2",
+    "pytest-asyncio>=1.3.0",
+]
+
+[tool.hatch.version]
+source = "vcs"
+
+[tool.hatch.build.targets.wheel]
+packages = ["kanta"]

tests/conftest.py

@@ -0,0 +1,14 @@
+import pytest
+
+from kanta import JsonSerializer, MsgPackSerializer
+
+
+@pytest.fixture(
+    params=[
+        ("json", JsonSerializer),
+        ("msgpack", MsgPackSerializer),
+    ],
+    ids=["json", "msgpack"],
+)
+def format_config(request):
+    return request.param

tests/support.py

@@ -0,0 +1,82 @@
+import sys
+from datetime import UTC, datetime
+from pathlib import Path
+from types import ModuleType
+from uuid import UUID
+
+import msgspec
+
+from kanta import ChangeRecord, Kanta
+
+
+class User(msgspec.Struct):
+    name: str = ""
+    age: int = 0
+
+
+class Data(msgspec.Struct):
+    users: dict[str, User] = {}
+    counter: int = 0
+
+
+class ExoticData(msgspec.Struct, omit_defaults=False):
+    uuid_values: dict[str, UUID] = {}
+    uuid_keys: dict[UUID, int] = {}
+    datetime_values: dict[str, datetime] = {}
+    datetime_keys: dict[datetime, int] = {}
+    bytes_values: dict[str, bytes] = {}
+    bytes_keys: dict[bytes, int] = {}
+
+
+class EvolvableDataV1(msgspec.Struct, omit_defaults=False):
+    counter: int = 0
+
+
+class EvolvableDataV2(msgspec.Struct, omit_defaults=False):
+    counter: int = 0
+    enabled: bool = True
+
+
+def make_kanta(path: Path, data_or_type, format_config, **kwargs):
+    _, serializer_cls = format_config
+    root = data_or_type() if isinstance(data_or_type, type) else data_or_type
+    return Kanta(
+        str(path),
+        root,
+        serializer=serializer_cls(),
+        **kwargs,
+    )
+
+
+def seed_single_change(path: Path, change: ChangeRecord, format_config) -> None:
+    _, serializer_cls = format_config
+    serializer = serializer_cls()
+    framer = serializer.framer_cls()
+    payload = serializer.encode(change)
+    path.write_bytes(framer.frame_change(payload, record_offset=0))
+
+
+def change_actions(path: Path, format_config) -> list[str]:
+    _, serializer_cls = format_config
+    serializer = serializer_cls()
+    framer = serializer.framer_cls()
+    actions: list[str] = []
+    for is_snapshot, payload, _, _ in framer.iter_records(path.read_bytes(), 0):
+        if is_snapshot:
+            continue
+        rec = serializer.decode(payload, type=ChangeRecord)
+        actions.append(rec.a)
+    return actions
+
+
+def make_migrations_module(name: str, fn_name: str, fn):
+    mod = ModuleType(name)
+    mod.__dict__[fn_name] = fn
+    sys.modules[name] = mod
+    return mod
+
+
+def fixed_change(action: str, diff: dict, *, version: int = 0) -> ChangeRecord:
+    return ChangeRecord(
+        ts=datetime(2026, 1, 1, tzinfo=UTC), a=action, v=version, diff=diff
+    )

tests/test_diff.py

@@ -0,0 +1,16 @@
+from kanta import compute_diff
+
+
+def test_no_diff():
+    assert compute_diff({"a": 1}, {"a": 1}) is None
+
+
+def test_simple_diff():
+    diff = compute_diff({"a": 1}, {"a": 2})
+    assert diff is not None
+    assert diff == {"a": 2}
+
+
+def test_nested_diff():
+    diff = compute_diff({"x": {"y": 1}}, {"x": {"y": 2}})
+    assert diff == {"x": {"y": 2}}

tests/test_format_diff.py

@@ -0,0 +1,25 @@
+from kanta import format_diff
+
+
+def test_add():
+    lines = format_diff({"name": "Alice"}, previous={})
+    assert any("name" in line for line in lines)
+
+
+def test_update():
+    lines = format_diff({"name": "Bob"}, previous={"name": "Alice"})
+    assert any("Bob" in line for line in lines)
+
+
+def test_delete():
+    lines = format_diff({"$delete": ["old_key"]}, previous={"old_key": 1})
+    assert any("old_key" in line for line in lines)
+
+
+def test_resolver():
+    lines = format_diff(
+        {"users": {"uuid-1": {"name": "Alice"}}},
+        previous={},
+        resolver=lambda x: "Alice" if x == "uuid-1" else x,
+    )
+    assert any("Alice" in line for line in lines)

tests/test_framing.py

@@ -0,0 +1,58 @@
+import pytest
+
+from kanta.exceptions import ReplayError
+from kanta.serialization.framing import BinFramer
+
+
+def test_roundtrip_with_sync_header_and_checksum():
+    framer = BinFramer()
+    first = framer.frame_change(b"c1", record_offset=0)
+    second = framer.frame_snapshot(b"snap", record_offset=len(first))
+    third = framer.frame_change(b"c2", record_offset=len(first) + len(second))
+    data = first + second + third
+
+    assert framer._sync_snapshot is not None
+    assert data[:4] == framer._sync_snapshot
+
+    snap_payload, resume_offset, snap_pos = framer.scan_last_snapshot(data)
+    assert snap_payload == b"snap"
+    assert snap_pos == len(first)
+    assert list(framer.iter_records(data, resume_offset)) == [
+        (False, b"c2", 0, len(first) + len(second))
+    ]
+    assert list(framer.iter_records(data, 0)) == [
+        (False, b"c1", 0, 4),
+        (True, b"snap", 0, len(first)),
+        (False, b"c2", 0, len(first) + len(second)),
+    ]
+
+
+def test_no_serialized_offset_in_change_frame():
+    framer = BinFramer()
+    data = framer.frame_change(b"payload", record_offset=0)
+    assert data[4:8] == bytes((~b) & 0xFF for b in framer._sync_snapshot)
+    payload_len = int.from_bytes(data[8:12], "little")
+    assert len(data) == 4 + 4 + 4 + 8 + payload_len
+
+
+def test_detects_tampered_checksum():
+    framer = BinFramer()
+    data = bytearray(framer.frame_change(b"payload", record_offset=0))
+    checksum_start = 4 + 4 + 4
+    data[checksum_start] ^= 0x01
+
+    with pytest.raises(ReplayError, match="invalid frame checksum") as exc_info:
+        list(framer.iter_records(bytes(data), 0))
+    assert exc_info.value.line_number == 0
+    assert exc_info.value.byte_pos == 4
+
+
+def test_detects_invalid_frame_marker():
+    framer = BinFramer()
+    data = bytearray(framer.frame_change(b"payload", record_offset=0))
+    data[4:8] = b"BAD!"
+
+    with pytest.raises(ReplayError, match="invalid frame marker") as exc_info:
+        list(framer.iter_records(bytes(data), 0))
+    assert exc_info.value.line_number == 0
+    assert exc_info.value.byte_pos == 4

tests/test_kanta_integration.py

@@ -0,0 +1,391 @@
+import asyncio
+import sys
+from datetime import UTC, datetime
+from uuid import uuid4
+
+import pytest
+
+from kanta.exceptions import DatabaseError, DataIntegrityError, FileLockError
+from kanta.serialization import struct_to_dict
+
+from .support import (
+    Data,
+    EvolvableDataV1,
+    EvolvableDataV2,
+    ExoticData,
+    User,
+    change_actions,
+    fixed_change,
+    make_kanta,
+    seed_single_change,
+)
+
+
+@pytest.mark.asyncio
+async def test_load_empty(tmp_path, format_config):
+    kanta = make_kanta(tmp_path / "test.db", Data, format_config)
+    await kanta.open()
+    assert isinstance(kanta.data, Data)
+    assert kanta.data.users == {}
+    await kanta.close()
+
+
+@pytest.mark.asyncio
+async def test_open_overwrites_caller_owned_root_data(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    seed_single_change(path, fixed_change("seed", {"counter": 7}), format_config)
+
+    root = Data(counter=99, users={"stale": User(name="Stale", age=1)})
+    kanta = make_kanta(path, root, format_config)
+    await kanta.open()
+
+    assert kanta.data is root
+    assert root.counter == 7
+    assert root.users == {}
+
+    await kanta.close()
+
+
+@pytest.mark.asyncio
+async def test_roundtrip(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open()
+
+    with kanta.transaction(action="inc") as data:
+        data.counter = 1
+
+    await kanta.flush()
+    await kanta.close()
+
+    kanta2 = make_kanta(path, Data, format_config)
+    await kanta2.open()
+    assert isinstance(kanta2.data, Data)
+    assert kanta2.data.counter == 1
+    await kanta2.close()
+
+
+@pytest.mark.asyncio
+async def test_rollback_on_error(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open()
+
+    try:
+        with kanta.transaction(action="inc") as data:
+            data.counter = 1
+            raise ValueError("boom")
+    except ValueError:
+        pass
+
+    assert kanta.data.counter == 0
+    assert isinstance(kanta.data, Data)
+    await kanta.close()
+
+
+@pytest.mark.asyncio
+async def test_bootstrap_creates_file(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open()
+
+    kanta.data = Data(counter=1)
+    kanta._impl.statedict = {}
+
+    with kanta.transaction(action="bootstrap") as data:
+        data.counter = 1
+
+    await kanta.flush()
+    await kanta.close()
+    assert path.exists()
+
+
+@pytest.mark.asyncio
+async def test_snapshot(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config, flush_interval=0.01)
+    await kanta.open()
+
+    kanta.data = Data(counter=1)
+    kanta._impl.statedict = struct_to_dict(kanta.data)
+    kanta._impl.snapshot._min_diffs = 1
+    kanta._impl.snapshot.request_force()
+
+    with kanta.transaction(action="inc") as data:
+        data.counter = 2
+
+    await kanta.flush()
+    await asyncio.sleep(0.05)
+    await kanta.close()
+
+    data = path.read_bytes()
+    _, serializer_cls = format_config
+    framer = serializer_cls().framer_cls()
+    snap_payload, _, _ = framer.scan_last_snapshot(data)
+    assert snap_payload is not None
+
+
+@pytest.mark.asyncio
+async def test_nested_struct_roundtrip(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open()
+
+    with kanta.transaction(action="create_user") as data:
+        data.users["alice"] = User(name="Alice", age=30)
+
+    await kanta.flush()
+    await kanta.close()
+
+    kanta2 = make_kanta(path, Data, format_config)
+    await kanta2.open()
+    assert isinstance(kanta2.data, Data)
+    assert kanta2.data.users["alice"].name == "Alice"
+    assert kanta2.data.users["alice"].age == 30
+
+    with kanta2.transaction(action="update_user") as data:
+        data.users["alice"].age = 31
+
+    await kanta2.flush()
+    await kanta2.close()
+
+    kanta3 = make_kanta(path, Data, format_config)
+    await kanta3.open()
+    assert isinstance(kanta3.data, Data)
+    assert kanta3.data.users["alice"].name == "Alice"
+    assert kanta3.data.users["alice"].age == 31
+    await kanta3.close()
+
+
+@pytest.mark.asyncio
+async def test_background_flush(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config, flush_interval=0.01)
+    await kanta.open()
+
+    with kanta.transaction(action="inc") as data:
+        data.counter = 1
+
+    await asyncio.sleep(0.05)
+    await kanta.close()
+
+    assert path.exists()
+    reloaded = make_kanta(path, Data, format_config)
+    await reloaded.open()
+    assert reloaded.data.counter == 1
+    await reloaded.close()
+
+
+@pytest.mark.asyncio
+async def test_async_with_open_close(tmp_path, format_config):
+    path = tmp_path / "test.db"
+
+    async with make_kanta(path, Data, format_config) as kanta:
+        with kanta.transaction(action="inc") as data:
+            data.counter = 1
+
+    assert path.exists()
+    reloaded = make_kanta(path, Data, format_config)
+    await reloaded.open()
+    assert reloaded.data.counter == 1
+    await reloaded.close()
+
+
+@pytest.mark.asyncio
+async def test_open_twice_raises(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+
+    await kanta.open()
+    with pytest.raises(DataIntegrityError, match="already open"):
+        await kanta.open()
+    await kanta.close()
+
+
+@pytest.mark.asyncio
+async def test_migrations_from_module(tmp_path, format_config):
+    path = tmp_path / "test.db"
+
+    mod = type(sys)("test_migrations")
+
+    def migrate_v1(d, ctx):
+        d["version"] = 1
+
+    mod.__dict__["migrate_v1"] = migrate_v1
+
+    seed_single_change(path, fixed_change("init", {"counter": 0}), format_config)
+
+    kanta = make_kanta(path, Data, format_config, migrations=mod)
+    await kanta.open()
+    assert kanta.version == 1
+    await kanta.close()
+
+
+@pytest.mark.asyncio
+async def test_msgspec_normalization_logs_migration(tmp_path, format_config):
+    path = tmp_path / "test.db"
+
+    seed_single_change(
+        path,
+        fixed_change("seed", {"users": {"alice": {"name": "Alice", "age": 30}}}),
+        format_config,
+    )
+
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open()
+    await kanta.close()
+
+    assert "migrate:msgspec" in change_actions(path, format_config)
+
+
+@pytest.mark.asyncio
+async def test_open_locked_file_raises_filelock_error(tmp_path, format_config):
+    path = tmp_path / "test.db"
+    kanta1 = make_kanta(path, Data, format_config)
+    await kanta1.open()
+    kanta2 = make_kanta(path, Data, format_config)
+    try:
+        with pytest.raises(FileLockError):
+            await kanta2.open()
+    finally:
+        await kanta1.close()
+
+
+@pytest.mark.asyncio
+async def test_flush_write_failure_bubbles_database_error(
+    tmp_path, format_config, monkeypatch
+):
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, Data, format_config)
+    await kanta.open()
+
+    with kanta.transaction(action="inc") as data:
+        data.counter = 1
+
+    def fail_write(_data: bytes) -> None:
+        raise OSError("simulated write failure")
+
+    monkeypatch.setattr(kanta._impl.file, "write", fail_write)
+
+    with pytest.raises(DatabaseError, match="Failed to flush database"):
+        await kanta.flush()
+
+    await kanta.close()
+
+
+@pytest.mark.asyncio
+async def test_background_write_failure_notifies_callback(
+    tmp_path, format_config, monkeypatch
+):
+    path = tmp_path / "test.db"
+    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,
+    )
+    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])
+    assert kanta._impl.background_error is not None
+
+    await kanta.close()
+
+
+@pytest.mark.asyncio
+async def test_migrations_from_module_path(tmp_path, format_config):
+    path = tmp_path / "test.db"
+
+    module_name = "test_migrations_path"
+    mod = type(sys)(module_name)
+
+    def migrate_v1(d, ctx):
+        d["counter"] = 2
+
+    mod.__dict__["migrate_v1"] = migrate_v1
+    sys.modules[module_name] = mod
+
+    seed_single_change(path, fixed_change("init", {"counter": 0}), format_config)
+
+    try:
+        kanta = make_kanta(path, Data, format_config, migrations=module_name)
+        await kanta.open()
+        assert kanta.version == 1
+        assert kanta.data.counter == 2
+        await kanta.close()
+    finally:
+        sys.modules.pop(module_name, None)
+
+
+@pytest.mark.asyncio
+async def test_uuid_datetime_bytes_keys_and_values_roundtrip(tmp_path, format_config):
+
+    path = tmp_path / "test.db"
+    kanta = make_kanta(path, ExoticData, format_config)
+    await kanta.open()
+
+    u = uuid4()
+    dt = datetime(2026, 1, 1, 12, 0, tzinfo=UTC)
+    bkey = b"blob-key"
+    bval = b"blob-value"
+
+    with kanta.transaction(action="set_exotic") as data:
+        data.uuid_values["u"] = u
+        data.uuid_keys[u] = 1
+        data.datetime_values["ts"] = dt
+        data.datetime_keys[dt] = 2
+        data.bytes_values["blob"] = bval
+        data.bytes_keys[bkey] = 3
+
+    await kanta.flush()
+    await kanta.close()
+
+    reloaded = make_kanta(path, ExoticData, format_config)
+    await reloaded.open()
+    assert reloaded.data.uuid_values["u"] == u
+    assert reloaded.data.uuid_keys[u] == 1
+    assert reloaded.data.datetime_values["ts"] == dt
+    assert reloaded.data.datetime_keys[dt] == 2
+    assert reloaded.data.bytes_values["blob"] == bval
+    assert reloaded.data.bytes_keys[bkey] == 3
+    await reloaded.close()
+
+
+@pytest.mark.asyncio
+async def test_schema_evolution_add_default_field_logs_migration(
+    tmp_path, format_config
+):
+    path = tmp_path / "test.db"
+
+    kanta_v1 = make_kanta(path, EvolvableDataV1, format_config)
+    await kanta_v1.open()
+    with kanta_v1.transaction(action="seed") as data:
+        data.counter = 1
+    await kanta_v1.flush()
+    await kanta_v1.close()
+
+    kanta_v2 = make_kanta(path, EvolvableDataV2, format_config)
+    await kanta_v2.open()
+    assert kanta_v2.data.counter == 1
+    assert kanta_v2.data.enabled is True
+    await kanta_v2.close()
+
+    assert "migrate:msgspec" in change_actions(path, format_config)

tests/test_logging.py

@@ -0,0 +1,17 @@
+import logging
+
+from kanta import configure_logging, log_change
+from kanta.logging import logger
+
+
+def test_configure_logging():
+    configure_logging()
+    assert logger.level == logging.INFO
+
+
+def test_log_change_no_diff(capsys):
+    logger.handlers.clear()
+    configure_logging()
+    log_change("test", {})
+    captured = capsys.readouterr()
+    assert "test" in captured.err

tests/test_migrations.py

@@ -0,0 +1,53 @@
+from types import ModuleType
+
+from kanta.migrate import MigrationRegistry
+
+
+def test_register_and_apply():
+    reg = MigrationRegistry()
+
+    @reg.register
+    def migrate_v1(d, ctx):
+        d["version"] = 1
+
+    @reg.register
+    def migrate_v2(d, ctx):
+        d["version"] = 2
+
+    state = {}
+    new_ver = reg.apply(state, current_version=0, silent=True)
+    assert new_ver == 2
+    assert state["version"] == 2
+
+
+def test_no_migrations_needed():
+    reg = MigrationRegistry()
+
+    @reg.register
+    def migrate_v1(d, ctx):
+        d["x"] = 1
+
+    state = {"x": 1}
+    new_ver = reg.apply(state, current_version=1, silent=True)
+    assert new_ver == 1
+
+
+def test_from_module():
+    mod = ModuleType("fake_migrations")
+
+    def migrate_v1(d, ctx):
+        d["v"] = 1
+
+    def migrate_v2(d, ctx):
+        d["v"] = 2
+
+    mod.__dict__["migrate_v1"] = migrate_v1
+    mod.__dict__["migrate_v2"] = migrate_v2
+
+    reg = MigrationRegistry.from_module(mod)
+    assert reg.dbver == 2
+
+    state = {}
+    new_ver = reg.apply(state, current_version=0, silent=True)
+    assert new_ver == 2
+    assert state["v"] == 2

tests/test_replay.py

@@ -0,0 +1,34 @@
+from datetime import UTC, datetime
+
+from kanta import ChangeRecord, Snapshot, replay
+from kanta.serialization.framing import LineFramer
+
+
+def test_empty_data():
+    rr = replay(b"")
+    assert rr.state == {}
+    assert rr.version == 0
+
+
+def test_single_change():
+    rec = ChangeRecord(a="test", v=1, diff={"name": "Alice"})
+    data = b"" + __import__("msgspec").json.encode(rec) + b"\n"
+    rr = replay(data)
+    assert rr.state == {"name": "Alice"}
+    assert rr.version == 1
+
+
+def test_snapshot_then_change():
+    snap = Snapshot(ts=datetime.now(UTC), v=1, state={"counter": 5})
+    line = LineFramer.SNAPSHOT_PREFIX + __import__("msgspec").json.encode(snap) + b"\n"
+    rec = ChangeRecord(a="inc", v=1, diff={"counter": 6})
+    line += __import__("msgspec").json.encode(rec) + b"\n"
+    rr = replay(line)
+    assert rr.state == {"counter": 6}
+
+
+def test_migration_flag():
+    rec = ChangeRecord(a="migrate:v1", v=1, diff={"x": 1})
+    data = __import__("msgspec").json.encode(rec) + b"\n"
+    rr = replay(data)
+    assert rr.has_migration is True

tests/test_snapshot.py

@@ -0,0 +1,34 @@
+from kanta.snapshot import SnapshotState
+
+
+def test_no_write_below_min_diffs():
+    class FakeFile:
+        def __init__(self):
+            self.written = []
+            self.is_open = True
+
+        def write(self, data: bytes):
+            self.written.append(data)
+
+    ss = SnapshotState(min_diffs=10)
+    ss.record_changes(5)
+    f = FakeFile()
+    ss.maybe_write(f, 1, {"x": 1})
+    assert len(f.written) == 0
+
+
+def test_force_writes():
+    class FakeFile:
+        def __init__(self):
+            self.written = []
+            self.is_open = True
+
+        def write(self, data: bytes):
+            self.written.append(data)
+
+    ss = SnapshotState(min_diffs=10)
+    ss.record_changes(15)
+    ss.request_force()
+    f = FakeFile()
+    ss.maybe_write(f, 1, {"x": 1})
+    assert len(f.written) == 1