Implement database rotation with n days retention (#2)

- Automatically erase history, keep as separate timestamped files. - Full history may be recovered by concatenation of the files. - Only applied at database opening time. - Enable by Kanta(retention=...), use with care (experimental feature).
2026-09-02 16:46:19 +00:00
parent 94c5ddeaba
commit aee6c13996
15 changed files with 1261 additions and 75 deletions
@@ -0,0 +1,267 @@
+# Database Rotation
+
+Goal: bound the on-disk history of a kanta database to a configurable retention
+window (e.g. the last 30 days) by *rotating* the database file: the old content
+is copied to a timestamped sibling file and the main file is truncated and
+rewritten in place with only the retained history plus fresh snapshots. Normal
+operation stays append-only under the exclusive lock; rotation is the only
+operation that rewrites the file.
+
+## Current facts the design must respect
+
+- The writer holds an exclusive `flock` on the file from `open()` until
+  `close()` (`kanta/filelock.py`). No other process can safely touch the file
+  while a writer has it open.
+- Records are append-only frames. Each `ChangeRecord` carries `ts` (record time)
+  and `m` (modification time); snapshots carry `ts`, `v` (schema version) and
+  `state` (`kanta/structs.py`).
+- Replay reads the whole file, then starts from the **last snapshot**
+  (`framer.scan_last_snapshot`, `serialization/base.py:replay`). Anything before
+  the last snapshot is already logically dead.
+- Snapshot state is validated in tooling: replayed state must equal snapshot
+  state (`kanta/replaylog.py`). A snapshot is therefore a consistency
+  checkpoint, not just an accelerator.
+- `BinFramer` checksums are **offset-keyed** (checksum includes the absolute
+  `record_offset`). A binary frame copied to a different byte offset is
+  corrupted. `LineFramer` (JSONL) has no checksums.
+- There is **no fsync/fdatasync** anywhere; durability currently relies on the
+  OS page cache. Rotation must not make this worse, and should fix it for the
+  rotation path at minimum.
+- Migrations run on open, after replay, against the snapshot/replay version.
+  A snapshot records the version it was written at, so "db already migrated"
+  survives in the snapshot even if the migrations produced no change records.
+
+## Rotated file naming
+
+The history that aged out is preserved at:
+
+```
+{stem}@{ISO-8601 timestamp}.kantadb
+```
+
+- `{stem}` is the original filename with its extension stripped
+  (`Path(filename).stem`).
+- The timestamp is the **ts of the last record dropped by the rotation** (see
+  step 4 — the leading snapshot of the rewritten main file carries the same
+  ts), not the current time. The name tells you exactly which point in history
+  the rotated file ends at. Rendered in ISO 8601 basic format at second
+  precision (e.g. `20260902T143000Z`). The exact microsecond timestamp of the
+  cutoff remains available inside the file (it is the ``ts`` of the final
+  line of the rotated file and of the snapshot at the start of the new file);
+  a second rotation within the same second cannot occur because rotation
+  requires history to have aged past the cutoff.
+- The rotated name always ends in `.kantadb`, regardless of the original
+  extension. Users may name their databases with no extension, `.kantadb`, or
+  anything else (`.db`, …). Since the rotated name is derived from the *stem*,
+  all of these work uniformly: `data` → `data@20260902T143000Z.kantadb`,
+  `data.kantadb` → `data@….kantadb`, `data.db` → `data@….kantadb`.
+- Rotated files live in the same directory.
+- Collision: if a rotated file with the same name already exists (rotation
+  rerun over identical history — should be prevented by the eligibility check
+  below, but be defensive), append a disambiguating suffix rather than
+  overwriting.
+
+## Why in-place rewrite (and not rename-and-recreate)
+
+An earlier draft renamed the locked file away and created a fresh file at the
+main path. That opens a race: between the rename and the creation of the new
+file, a second instance can open the (now missing) main path with `O_CREAT`,
+acquire its own lock on the fresh inode, and bootstrap an empty database. The
+rotating instance then cannot lock the path it needs, and two divergent
+databases exist. `flock` is attached to the open file description (inode), not
+the path — renaming never blocks a newcomer.
+
+Instead, rotation **never renames or unlinks the main file and never releases
+its lock**:
+
+- Unix: `ftruncate(fd, 0)` on the open, locked fd is unaffected by the flock
+  and does not affect it. Subsequent writes use `lseek(fd, 0, SEEK_END)` +
+  `os.write` (`filelock.py:226`), which work identically after a truncate, so
+  append-mode operation continues unchanged.
+- Windows: this is also the *more* portable option — the DB is opened with
+  `FILE_SHARE_READ` only (`filelock.py:240`), so renaming the locked file would
+  fail outright on Windows. In-place rewrite only needs `SetFilePointer(0)` +
+  `SetEndOfFile` on a handle we own.
+- The main path therefore exists and remains locked throughout; a second
+  instance opening it at any moment gets either the old content or the new,
+  never a missing or half-created file, and never its own lock.
+
+The only new capability `LockedFile` needs is a `replace_content(data)` method
+(seek 0, truncate, write, fsync) implemented per platform.
+
+## When to rotate: at open time, not at runtime
+
+Rotation happens **inside `Kanta.open()`, after acquiring the lock, before
+replay**, gated by a retention option (see Configuration). Rationale:
+
+- The lock is already held and no background flush loop is running yet, so the
+  file is quiescent — no in-flight `pending_changes`, no concurrent snapshots.
+- Runtime rotation would have to fence the background writer, drain the queue,
+  and prove no record lands in the file after the cutoff was computed. That
+  is a second synchronization protocol for a rare operation; not worth it.
+- Open-time rotation also means rotation never races with `request_snapshot()`
+  or migration snapshot writes, which all happen under the same open() sequence.
+
+Consequence: a database that is never reopened never rotates. Document this;
+for long-running services, rotation takes effect on the next restart.
+
+## Rotation algorithm (under the exclusive lock)
+
+Let `cutoff = now - retention`. Steps 1–3 operate on the bytes already read
+into memory by `open_and_read`; no second disk read is needed.
+
+1. **Check eligibility.** Skip rotation when there is nothing to do:
+   - The file contains **no change records older than `cutoff`** — the
+     retention window already covers all history.
+   - The file contains **no change records at all** (snapshot-only file).
+     Opening a long-untouched database may legitimately rotate it down to a
+     single snapshot (that *is* the intended purge), but once a file has been
+     reduced to just a snapshot, rotating it again would be a pure no-op
+     rewrite. Treat "no change records" as "already fully rotated" and skip.
+
+2. **Find the replay base.** Replay normally starts at the most recent
+   snapshot, but that snapshot's `ts` is likely newer than `cutoff` — replaying
+   from it would silently drop history we intend to keep. Instead, scan
+   **backwards from the end of file**, collecting snapshots newest-first, and
+   pick the oldest snapshot `S` whose `ts <= cutoff` (i.e. walk back past
+   snapshots until one covers the required range, or until start of file). If
+   no such snapshot exists, `S` is "start of file" and the retained range is
+   replayed from the empty initial state.
+   - For `LineFramer` this is a reverse scan for `\nSNAPSHOT ` lines.
+   - For `BinFramer` frames are forward-scannable only; keep the forward scan
+     but record every snapshot position, then pick from the collected list.
+
+3. **Replay and validate.** Replay from `S` (or start of file) forward to end
+   of file, keeping every record with `ts >= cutoff`. At **every** snapshot
+   encountered after `S`, validate that the replayed state equals the snapshot
+   state; a mismatch means the history is corrupt or the chosen base is wrong —
+   abort rotation (leave the original file untouched) and surface the error.
+   The last snapshot in the file must always validate; if even that fails,
+   rotation must not proceed.
+   - Records with `ts < cutoff` are applied to the replay (they are needed to
+     reach the cutoff state) but not retained in the output.
+   - Remember `cutoff_end`: the byte offset in the original content just after
+     the last record with `ts < cutoff` (frame-boundary aligned). The rotated
+     file will be truncated to this length in step 6.
+
+4. **Copy the original aside.** `shutil.copy2(main_path, rotated_path)` —
+   no lock needed on the copy, and no temporary name: the content is written
+   directly to its final `{stem}@{ts}.kantadb` name. `copy2` preserves
+   metadata and, on filesystems with copy-on-write (btrfs, XFS with reflinks,
+   APFS, …), performs a cheap reflink copy instead of duplicating data; it is
+   also generally faster than re-writing the same bytes from memory. The
+   original bytes remain readable from the locked fd if the copy fails, so a
+   failure here simply aborts rotation.
+
+5. **Rewrite the main file in place.** On the locked fd: seek to 0, truncate
+   to 0, write the new content, `fdatasync`. The new content is, in order:
+   1. A **snapshot of the state at the cutoff** — the replayed state after
+      applying all records with `ts < cutoff`, stamped with the **schema
+      version in effect at the cutoff**. Its `ts` is the **ts of the last
+      pre-cutoff record** (not the rotation time), and this is exactly the
+      timestamp used in the rotated filename. This snapshot is the new replay
+      base and carries the version forward so migrations are not re-run; it is
+      always written.
+   2. The retained change records (`ts >= cutoff`), **recreated record by
+      record** — no internal snapshots are carried over, even if the original
+      file had many in the retained range.
+   3. A **final snapshot** of the state after the last retained record,
+      stamped with the version of the last retained record — written **only
+      if** there were
+      retained change records (and, in line with the existing snapshot policy
+      in `kanta/snapshot.py`, only when a meaningful number of changes
+      accumulated; a handful of trailing changes need not force one). If no
+      records survived the cutoff, the new file consists of the single leading
+      snapshot and nothing else — this is the steady state for databases whose
+      history has fully aged out, and the eligibility check in step 1 prevents
+      re-rotating such files.
+
+6. **Trim the rotated copy.** Truncate `{stem}@{ts}.kantadb` to `cutoff_end`
+   bytes, so it contains **only the dropped history** and does not duplicate
+   the records retained in the main file. The cut is at a frame boundary, so
+   the rotated file remains a valid, replayable database on its own (it is a
+   prefix of a valid log). This truncation happens only after step 5's fsync,
+   so until then the rotated file still holds the complete original content as
+   a crash-recovery anchor.
+
+7. **Continue normal open.** Replay/migrations proceed on the same locked fd.
+   Because the leading snapshot carries the current version, migrations run
+   exactly as they would have against the old content.
+
+Failure rule: any error before step 5 leaves the main file byte-identical
+(only an extra copy exists). A crash during step 5 may leave the main file
+torn, but the rotated copy still holds the complete original content
+(truncated only after the main file is durable) — recovery is copying it back.
+After step 6 the split is complete and both files are consistent.
+
+## Verbatim copy or rewrite?
+
+**Rewrite (re-frame), not verbatim copy**, for all records written to the main
+file:
+
+- `BinFramer` checksums include `record_offset`, so a verbatim byte copy to a
+  new offset is unreadable. Binary records must be re-framed at their new
+  offsets regardless.
+- Rewriting also normalizes encoding drift and lets us drop the redundant
+  intermediate snapshots the original file accumulated: none of them are
+  carried over — the new file contains only the leading cutoff snapshot, the
+  recreated change records, and (conditionally) the final snapshot.
+
+The rotated copy is the one place where verbatim bytes are used — a raw
+`copy2` plus a frame-aligned tail truncation — which is safe precisely because
+it preserves original offsets (the truncated prefix keeps every frame at its
+original `record_offset`, so binary checksums stay valid).
+
+## Configuration
+
+Add keyword options to `Kanta(...)` (`kanta/kanta.py`), surfaced through
+`open()`:
+
+- `retention: timedelta | int | None = None` — history window to keep; a plain
+  `int` is interpreted as a number of days. `None` (default) disables rotation
+  entirely; current behavior is unchanged.
+- `rotate_keep: int = 3` (optional, later) — how many rotated backups to
+  retain; older ones are pruned at rotation time.
+
+Rotation uses `impl.now()` so the `@Kanta.clock` test clock controls it, same
+as record timestamps.
+
+## Integrity checklist
+
+- Rotation runs under the exclusive lock, before the background writer starts.
+- The main path is never renamed, unlinked, or unlocked during rotation; no
+  bootstrap race with a second instance is possible.
+- Replay base is chosen by walking snapshots backwards until the retained range
+  is covered; replay is validated against every snapshot in range.
+- A leading cutoff snapshot (ts = last pre-cutoff record, current schema
+  version) is always written; a final snapshot is written only when warranted
+  by retained changes.
+- The full original content sits at `{stem}@{ts}.kantadb` before the main file
+  is touched, and is only trimmed to the dropped-history prefix after the
+  rewritten main file is `fdatasync`ed.
+- Rotated files are never deleted by the rotation itself.
+- Any validation failure aborts rotation with the original file intact.
+- Files with no change records (already reduced to a snapshot) are never
+  re-rotated.
+
+## Testing notes
+
+- Use the test clock (`tests/test_clock.py`) to age records past the cutoff.
+- Cover both framers: JSONL rotation and BinFramer rotation (assert the
+  rewritten binary file passes checksum validation and replays identically,
+  and that the truncated rotated prefix still passes checksum validation).
+- Assert state equality before/after rotation, version continuity (no
+  re-migration), correct behavior when no snapshot precedes the cutoff, when
+  the newest snapshot is already older than the cutoff, and when retention
+  covers everything (no-op).
+- Naming: databases named `x`, `x.kantadb`, and `x.db` all rotate to
+  `x@{ts}.kantadb`; the timestamp equals the last dropped record's ts and the
+  leading snapshot's ts.
+- Assert the rotated file ends exactly at the last dropped record's frame
+  boundary (no overlap with the retained history in the main file).
+- No-change files: a snapshot-only database opened with retention set is left
+  untouched (no copy, no rewrite).
+- Aged-out database: all history older than the cutoff → new file contains
+  exactly one snapshot; opening it again performs no rotation.
+- Concurrency: while one instance rotates, a second instance opening the main
+  path must fail with the normal "already locked" error at every stage.
@@ -1,63 +1,66 @@
-"""Diff computation and replay utilities."""
+"""Diff computation and replay utilities.

-import jsondiff
+Diff format: JSON-serializable dicts where ``$delete`` is the only command
+our producer emits; added keys and changed values (scalars, lists, type
+changes — lists always wholesale) are plain assignments. A dict value
+assigned over a non-dict needs no ``$replace``: the consumer can see from
+the old value whether to patch (old is a dict) or replace. User keys
+starting with ``$`` are escaped by prepending another ``$``
+(``$foo`` -> ``$$foo``); values are stored verbatim.
+
+The consumer additionally stays compatible with jsondiff's marshaled
+syntax, so it can replay diffs produced by jsondiff itself: ``$replace``,
+positional ``$insert``/``$delete`` and per-index nested diffs on lists,
+and jsondiff's escaping of ``$``-prefixed values.
+"""

 from kanta.structs import ChangeRecord
-from kanta.serialization.base import ReplayResult, replay
+from kanta.serialization.base import ReplayResult, apply_diff, replay
 from kanta.serialization.framing import LineFramer
 from kanta.serialization.json import JsonSerializer

+_UNCHANGED = object()
+
+
+def _escape_key(key: str) -> str:
+    """Escape a user key for use as a diff key (``$foo`` -> ``$$foo``)."""
+    if isinstance(key, str) and key.startswith("$"):
+        return "$" + key
+    return key
+
+
+def _diff(previous, current):
+    """Compute a raw diff, or _UNCHANGED if there is no difference."""
+    if isinstance(previous, dict) and isinstance(current, dict):
+        result = {}
+        deleted = [_escape_key(k) for k in previous if k not in current]
+        if deleted:
+            result["$delete"] = deleted
+        for key, new_value in current.items():
+            if key not in previous:
+                result[_escape_key(key)] = new_value
+            else:
+                sub = _diff(previous[key], new_value)
+                if sub is not _UNCHANGED:
+                    result[_escape_key(key)] = sub
+        return result if result else _UNCHANGED
+    if previous == current:
+        return _UNCHANGED
+    return current
+

 def compute_diff(previous: dict, current: dict) -> dict | None:
-    """Compute a jsondiff patch between two dicts.
+    """Compute a marshaled diff between two state 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
+    diff = _diff(previous, current)
+    return diff if diff is not _UNCHANGED else None


 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)
+    """Apply a marshaled diff to a state dict."""
+    return apply_diff(state, diff)


 # Backward-compatible JSONL replay using the default serializer.
@@ -83,6 +83,10 @@ if sys.platform == "win32":
    ]
    _kernel32.CloseHandle.restype = wintypes.BOOL
    _kernel32.CloseHandle.argtypes = [wintypes.HANDLE]
+    _kernel32.SetEndOfFile.restype = wintypes.BOOL
+    _kernel32.SetEndOfFile.argtypes = [wintypes.HANDLE]
+    _kernel32.FlushFileBuffers.restype = wintypes.BOOL
+    _kernel32.FlushFileBuffers.argtypes = [wintypes.HANDLE]

    def _is_invalid_handle(handle) -> bool:
        return ctypes.c_void_p(handle).value == ctypes.c_void_p(-1).value
@@ -177,6 +181,22 @@ class LockedFile:
        os.lseek(self._fd, current, os.SEEK_SET)
        return end

+    def replace_content(self, data: bytes) -> None:
+        """Atomically-ish rewrite the file's content in place, lock retained.
+
+        Seeks to the start, truncates, writes *data* and fsyncs, all on the
+        already-locked descriptor. The path is never unlinked or renamed, so
+        no other process can observe a missing file or acquire its own lock.
+        Used by database rotation.
+        """
+        if self._fd is None:
+            raise RuntimeError("LockedFile.replace_content() called on a closed file")
+
+        if sys.platform == "win32":
+            self._replace_content_win32(data)
+        else:
+            self._replace_content_unix(data)
+
    def close(self) -> None:
        """Release the lock and close the file."""
        if self._fd is None:
@@ -227,6 +247,15 @@ class LockedFile:
        os.lseek(self._fd, 0, os.SEEK_END)
        os.write(self._fd, data)

+    def _replace_content_unix(self, data: bytes) -> None:
+        os.lseek(self._fd, 0, os.SEEK_SET)
+        os.ftruncate(self._fd, 0)
+        view = memoryview(data)
+        while view:
+            written = os.write(self._fd, view)
+            view = view[written:]
+        os.fdatasync(self._fd)
+
    # -- Windows -------------------------------------------------------------

    def _open_win32(self, path: Path, create: bool, readonly: bool) -> None:
@@ -288,3 +317,24 @@ class LockedFile:
        )
        if not ok:
            raise OSError(f"WriteFile failed: Windows error {ctypes.get_last_error()}")
+
+    def _replace_content_win32(self, data: bytes) -> None:
+        _kernel32.SetFilePointer(self._fd, 0, None, _FILE_BEGIN)
+        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()}")
+        if not _kernel32.SetEndOfFile(self._fd):
+            raise OSError(
+                f"SetEndOfFile failed: Windows error {ctypes.get_last_error()}"
+            )
+        if not _kernel32.FlushFileBuffers(self._fd):
+            raise OSError(
+                f"FlushFileBuffers failed: Windows error {ctypes.get_last_error()}"
+            )
@@ -2,7 +2,7 @@

 from __future__ import annotations
 import logging
-from datetime import datetime
+from datetime import datetime, timedelta
 from pathlib import Path
 from types import ModuleType, SimpleNamespace
 from typing import Any, Generic, TypeVar
@@ -53,6 +53,7 @@ class Kanta(Generic[T]):
        migrations: ModuleType | str | None = None,
        serializer: Serializer | None = None,
        flush_interval: float = 0.1,
+        retention: timedelta | int | None = None,
    ):
        """Initialize a Kanta persistence instance.

@@ -63,6 +64,13 @@ class Kanta(Generic[T]):
            migrations: Optional migrations module object or import path.
            flush_interval: Background flush interval in seconds.
            serializer: Optional serializer implementation.
+            retention: Optional history retention window, either a
+                :class:`~datetime.timedelta` or a plain number of days. When
+                set, opening the database rotates it: history older than
+                ``now - retention`` is moved to a ``{stem}@{timestamp}.kantadb``
+                sibling file and the main file is rewritten with a fresh
+                snapshot plus the retained records (see ``docs/rotation.md``).
+                ``None`` (default) disables rotation.

        Raises:
            ImportError: If ``migrations`` is a string path that cannot be imported.
@@ -78,6 +86,7 @@ class Kanta(Generic[T]):
            type=data_type,
            migrations=migrations,
            flush_interval=flush_interval,
+            retention=retention,
            kanta=self,
        )

@@ -6,7 +6,7 @@ import asyncio
 import copy
 import importlib
 import logging
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
 from types import SimpleNamespace
 from typing import Any, Generic, TypeVar

@@ -21,6 +21,7 @@ from kanta.logging import (
 )
 from kanta.migrations import MigrationReport, Migrations
 from kanta.persistence import PersistenceMixin
+from kanta.rotation import execute_rotation, plan_rotation
 from kanta.serialization import restore_data_in_place, struct_to_dict
 from kanta.serialization.base import replay

@@ -42,6 +43,10 @@ class KantaImpl(PersistenceMixin, Generic[T]):
        self.data: T = kwargs.pop("data")
        self._kanta = kwargs.pop("kanta", None)
        migrations = kwargs.pop("migrations", None)
+        retention = kwargs.pop("retention", None)
+        if isinstance(retention, int) and not isinstance(retention, bool):
+            retention = timedelta(days=retention)
+        self.retention: timedelta | None = retention
        self.ctx = SimpleNamespace()
        super().__init__(**kwargs)
        self.migrations: Migrations | None = None
@@ -181,6 +186,9 @@ class KantaImpl(PersistenceMixin, Generic[T]):
        # From this point the file is open and must be closed via close().
        self.opened = True

+        if content and self.retention is not None and not readonly:
+            content = await self._maybe_rotate(content, log)
+
        if content:
            try:
                rr = replay(
@@ -375,6 +383,30 @@ class KantaImpl(PersistenceMixin, Generic[T]):
        if not self.readonly:
            self.background_task = asyncio.create_task(self._background_loop())

+    async def _maybe_rotate(self, content: bytes, log: bool | logging.Logger) -> bytes:
+        """Rotate history older than the retention window (see docs/rotation.md).
+
+        Runs while the file is locked and quiescent, before replay. Returns
+        the (possibly replaced) content to replay. Rotation failures abort the
+        open with the original file intact.
+        """
+        cutoff = self.now() - self.retention
+        plan = await asyncio.to_thread(
+            plan_rotation,
+            content,
+            framer=self.framer,
+            serializer=self.serializer,
+            cutoff=cutoff,
+            now=self.now(),
+            min_diffs=self.snapshot.min_diffs,
+        )
+        if plan is None:
+            return content
+        await asyncio.to_thread(
+            execute_rotation, self.filename, self.file, plan, log=log
+        )
+        return plan.new_content
+
    async def close(self) -> None:
        """Stop the background task, flush pending changes, and release the file lock."""
        if not self.opened:
@@ -15,6 +15,7 @@ from typing import Any

 import msgspec

+from kanta.serialization.base import _apply, unmarshal
 from kanta.tty import Line, displaywidth

 transaction_logger = logging.getLogger("kanta.transaction")
@@ -312,6 +313,13 @@ def _collect_changes(
        changes.append(("update" if existed else "add", path, diff))
        return

+    old_at_path = _get_nested(previous, path)
+    if isinstance(old_at_path, list):
+        # List edits ($insert/$delete/per-index) are shown as one whole-list
+        # update; the diff is already unmarshaled at this point.
+        changes.append(("update", path, _apply(old_at_path, diff)))
+        return
+
    for key, value in diff.items():
        if key == "$delete":
            if isinstance(value, list):
@@ -340,7 +348,14 @@ def _collect_changes(
                    ("update" if old_collection is not None else "add", path, value)
                )
        elif isinstance(key, str) and key.startswith("$"):
-            changes.append(("add", path, {key: value}))
+            # Unknown $-command or (post-unmarshal) a user key starting with
+            # "$": treat as a normal key.
+            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))
        else:
            new_path = path + [str(key)]
            existed = _get_nested(previous, new_path) is not None
@@ -418,7 +433,7 @@ def format_diff(
    Returns a list of formatted lines (without newlines).
    """
    changes: list[tuple[str, list[str], Any]] = []
-    _collect_changes(diff, [], changes, previous)
+    _collect_changes(unmarshal(diff), [], changes, previous)
    if not changes:
        return []
    lines = []
@@ -0,0 +1,225 @@
+"""Database rotation: bound on-disk history to a retention window.
+
+See docs/rotation.md for the design. All planning happens on the in-memory
+bytes of the database file; the caller (KantaImpl.open) performs the actual
+copy-aside, in-place rewrite and rotated-file trimming under the file lock.
+"""
+
+from __future__ import annotations
+
+import copy
+import logging
+import shutil
+from dataclasses import dataclass
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+from kanta.exceptions import DatabaseError
+from kanta.structs import ChangeRecord, Snapshot
+from kanta.serialization.base import Serializer, apply_diff
+from kanta.serialization.framing import Framer
+
+_logger = logging.getLogger(__name__)
+
+
+@dataclass
+class RotationPlan:
+    """Everything needed to execute a rotation on disk."""
+
+    new_content: bytes
+    cutoff_end: int  # byte length of the dropped-history prefix
+    rotated_ts: datetime  # ts of the last dropped change record
+    retained_changes: int
+
+
+def rotated_path_for(path: Path, ts: datetime) -> Path:
+    """Sibling path for the rotated history: ``{stem}@{ISO-basic-ts}.kantadb``.
+
+    The timestamp uses ISO 8601 basic format at second precision (e.g.
+    ``20260902T143000Z``); the exact microsecond timestamp remains available
+    inside the file if ever needed. On collision an incrementing suffix is
+    inserted before the extension.
+    """
+    stamp = ts.strftime("%Y%m%dT%H%M%SZ")
+    candidate = path.with_name(f"{path.stem}@{stamp}.kantadb")
+    n = 1
+    while candidate.exists():
+        candidate = path.with_name(f"{path.stem}@{stamp}.{n}.kantadb")
+        n += 1
+    return candidate
+
+
+class _Entry:
+    """One parsed record frame with its byte range."""
+
+    __slots__ = ("is_snapshot", "record", "byte_pos", "end_pos")
+
+    def __init__(
+        self,
+        is_snapshot: bool,
+        record: ChangeRecord | Snapshot,
+        byte_pos: int,
+        end_pos: int,
+    ) -> None:
+        self.is_snapshot = is_snapshot
+        self.record = record
+        self.byte_pos = byte_pos
+        self.end_pos = end_pos
+
+
+def _scan(content: bytes, *, framer: Framer, serializer: Serializer) -> list[_Entry]:
+    """Decode every record in *content* with byte ranges."""
+    raw = list(framer.iter_records(content, 0))
+    entries: list[_Entry] = []
+    for i, (is_snapshot, payload, _line, byte_pos) in enumerate(raw):
+        end_pos = raw[i + 1][3] if i + 1 < len(raw) else len(content)
+        record = serializer.decode(
+            payload, type=Snapshot if is_snapshot else ChangeRecord
+        )
+        entries.append(_Entry(is_snapshot, record, byte_pos, end_pos))
+    return entries
+
+
+def plan_rotation(
+    content: bytes,
+    *,
+    framer: Framer,
+    serializer: Serializer,
+    cutoff: datetime,
+    now: datetime,
+    min_diffs: int,
+) -> RotationPlan | None:
+    """Plan a rotation of *content*, or return None when there is nothing to do.
+
+    Raises:
+        DatabaseError: If replay from the chosen base snapshot does not match
+            a snapshot found inside the file (corrupt history). Rotation must
+            be aborted and the original file left untouched.
+    """
+    if not content:
+        return None
+    entries = _scan(content, framer=framer, serializer=serializer)
+    changes = [e for e in entries if not e.is_snapshot]
+    if not changes:
+        return None  # snapshot-only file: already fully rotated
+
+    dropped = [e for e in changes if e.record.ts < cutoff]
+    if not dropped:
+        return None  # retention window covers all history
+
+    retained = [e for e in changes if e.record.ts >= cutoff]
+    rotated_ts = dropped[-1].record.ts
+    cutoff_end = retained[0].byte_pos if retained else len(content)
+
+    # Replay base: walk snapshots newest-first and take the first (newest)
+    # one predating the cutoff; fall back to start of file.
+    base: _Entry | None = None
+    for e in reversed([e for e in entries if e.is_snapshot]):
+        if e.record.ts <= cutoff:
+            base = e
+            break
+
+    state: dict[str, Any] = {}
+    version = 0
+    m: datetime | None = None
+    if base is not None:
+        snap = base.record
+        assert isinstance(snap, Snapshot)
+        state = dict(snap.state)
+        version = snap.v
+        m = snap.m
+
+    # The cutoff state starts from the replay base: when the base snapshot
+    # already predates the cutoff, it may itself be the cutoff state.
+    state_at_cutoff: dict[str, Any] | None = (
+        copy.deepcopy(state) if base is not None else None
+    )
+    version_at_cutoff = version
+    m_at_cutoff = m
+    final_version = version
+    final_m = m
+
+    for e in entries:
+        if base is not None and e.byte_pos <= base.byte_pos:
+            continue
+        if e.is_snapshot:
+            snap = e.record
+            assert isinstance(snap, Snapshot)
+            if snap.state != state:
+                raise DatabaseError(
+                    "rotation aborted: replayed state does not match snapshot "
+                    f"at byte {e.byte_pos}",
+                    action="rotate",
+                )
+            if snap.m is not None:
+                m = snap.m
+            continue
+        change = e.record
+        assert isinstance(change, ChangeRecord)
+        state = apply_diff(state, change.diff)
+        version = change.v
+        if change.m is not None:
+            m = change.m
+        if change.ts < cutoff:
+            state_at_cutoff = copy.deepcopy(state)
+            version_at_cutoff = version
+            m_at_cutoff = m
+        final_version = version
+        final_m = m
+
+    # There is at least one dropped change, so the cutoff state is known.
+    assert state_at_cutoff is not None
+
+    # Build the new content: leading cutoff snapshot, retained changes
+    # re-framed at fresh offsets, and a final snapshot when enough changes
+    # survived to warrant one (mirrors the regular snapshot policy).
+    out = bytearray()
+    leading = serializer.encode(
+        Snapshot(
+            ts=rotated_ts, v=version_at_cutoff, state=state_at_cutoff, m=m_at_cutoff
+        )
+    )
+    out += framer.frame_snapshot(leading, record_offset=0)
+    for e in retained:
+        payload = serializer.encode(e.record)
+        out += framer.frame_change(payload, record_offset=len(out))
+    if len(retained) >= min_diffs:
+        closing = serializer.encode(
+            Snapshot(ts=now, v=final_version, state=state, m=final_m)
+        )
+        out += framer.frame_snapshot(closing, record_offset=len(out))
+
+    return RotationPlan(
+        new_content=bytes(out),
+        cutoff_end=cutoff_end,
+        rotated_ts=rotated_ts,
+        retained_changes=len(retained),
+    )
+
+
+def execute_rotation(
+    path: Path, file, plan: RotationPlan, *, log: bool | logging.Logger = True
+) -> Path:
+    """Execute a planned rotation on disk. Caller must hold the lock on *file*.
+
+    1. Copy the original content aside to ``{stem}@{ts}.kantadb``.
+    2. Rewrite the locked file in place with the new content and fsync.
+    3. Trim the rotated copy to the dropped-history prefix.
+
+    Returns the rotated file path.
+    """
+    rotated = rotated_path_for(path, plan.rotated_ts)
+    shutil.copy2(path, rotated)
+    file.replace_content(plan.new_content)
+    with open(rotated, "r+b") as f:
+        f.truncate(plan.cutoff_end)
+    if log:
+        _logger.info(
+            "rotated %s: kept %d change record(s), history before %s moved to %s",
+            path,
+            plan.retained_changes,
+            plan.rotated_ts.isoformat(),
+            rotated,
+        )
+    return rotated
@@ -121,33 +121,100 @@ def replay(


 def _patch_state(state: dict, diff: dict) -> dict:
-    return _apply_diff(state, diff)
+    return apply_diff(state, diff)


-def _apply_diff(state: dict, diff: dict) -> dict:
+def _unescape(value: str) -> str:
+    """Reverse jsondiff's ``$$`` escaping; command strings pass through.
+
+    Only a ``$$`` prefix is stripped: jsondiff escapes ``$x`` to ``$$x``,
+    while single ``$`` strings occur verbatim in our own diffs (we do not
+    escape values) and must be left alone.
+    """
+    if value.startswith("$$"):
+        return value[1:]
+    return value
+
+
+def unmarshal(diff: Any) -> Any:
+    """Unescape a marshaled diff (keys, values and ``$delete`` entries).
+
+    Needed for jsondiff-produced diffs, which escape ``$``-prefixed values
+    as well as keys; our own producer escapes keys only, so unescaping
+    values is a no-op for them.
+    """
+    if isinstance(diff, dict):
+        return {
+            _unescape(k) if isinstance(k, str) else k: unmarshal(v)
+            for k, v in diff.items()
+        }
+    if isinstance(diff, list):
+        return [unmarshal(v) for v in diff]
+    if isinstance(diff, str):
+        return _unescape(diff)
+    return diff
+
+
+def apply_diff(state: Any, diff: Any) -> Any:
+    """Apply a diff.
+
+    Understands our own format (plain assignment + ``$delete``) and
+    jsondiff's marshaled syntax: ``$replace``, positional
+    ``$delete``/``$insert`` and per-index nested diffs on lists. A bare
+    dict over a non-dict old value is a wholesale replacement.
+    """
+    return _apply(state, unmarshal(diff))
+
+
+def _is_list_patch(diff: dict) -> bool:
+    """Whether a dict diff against a list state is a jsondiff list edit."""
+    for key in diff:
+        if key in ("$delete", "$insert"):
+            continue
+        try:
+            int(key)
+        except (ValueError, TypeError):
+            return False
+    return True
+
+
+def _apply(state: Any, diff: Any) -> Any:
    if not isinstance(diff, dict):
        return diff
+    if not diff:
+        return state
+    if "$replace" in diff:
+        return diff["$replace"]

-    result = dict(state) if isinstance(state, dict) else state
-    if not isinstance(result, dict):
-        result = {}
+    if isinstance(state, list):
+        if not _is_list_patch(diff):
+            # Our own producer replaces a list with a dict (or any other
+            # type) by plain assignment — no $replace wrapper.
+            return diff
+        result = list(state)
+        deletes = diff.get("$delete")
+        if deletes:
+            for pos in deletes:
+                result.pop(pos)
+        for pos, value in diff.get("$insert", []):
+            result.insert(pos, value)
+        for key, value in diff.items():
+            if key in ("$delete", "$insert"):
+                continue
+            pos = int(key)
+            result[pos] = _apply(result[pos], value)
+        return result

+    result = dict(state) if isinstance(state, dict) else {}
    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)
+            keys = value if isinstance(value, list) else [value]
+            for k in keys:
+                result.pop(k, None)
+        elif key == "$insert":
            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
-
+        elif key in result:
+            result[key] = _apply(result[key], value)
+        else:
+            result[key] = value
    return result
@@ -35,6 +35,11 @@ class SnapshotState:
        """Force snapshot write on next check."""
        self._force_pending = True

+    @property
+    def min_diffs(self) -> int:
+        """Minimum accumulated changes before a snapshot may be written."""
+        return self._min_diffs
+
    def record_changes(self, count: int) -> None:
        self.changes += count

@@ -17,7 +17,6 @@ readme = "README.md"
 requires-python = ">=3.11"
 dependencies = [
    "blake3>=1.0.8",
-    "jsondiff>=2.2.1",
    "msgspec>=0.20.0",
 ]

@@ -30,10 +29,12 @@ bin = [
 ]

 [project.urls]
+Homepage = "https://vasanko.com/coders/kanta"
 Repository = "https://git.zi.fi/LeoVasanko/kanta"

 [dependency-groups]
 dev = [
+    "jsondiff>=2.2.1",
    "pytest>=9.0.2",
    "pytest-asyncio>=1.3.0",
 ]
@@ -1,8 +1,26 @@
-from kanta.diff import compute_diff
+"""Tests for our own diff producer/consumer and jsondiff compatibility.
+
+jsondiff is a dev dependency used only here, to verify that:
+
+- jsondiff.patch(..., marshal=True) can apply patches produced by
+  compute_diff (our format is a subset of jsondiff's marshaled syntax);
+- apply_diff can apply patches produced by jsondiff.diff(..., marshal=True),
+  including positional $insert/$delete list edits and per-index nested diffs.
+"""
+
+import jsondiff
+import pytest
+
+from kanta.diff import compute_diff, patch_state
+from kanta.logging import format_diff
+from kanta.serialization.base import apply_diff
+
+# --- Producer: compute_diff ------------------------------------------------


 def test_no_diff():
    assert compute_diff({"a": 1}, {"a": 1}) is None
+    assert compute_diff({}, {}) is None


 def test_simple_diff():
@@ -14,3 +32,227 @@ def test_simple_diff():
 def test_nested_diff():
    diff = compute_diff({"x": {"y": 1}}, {"x": {"y": 2}})
    assert diff == {"x": {"y": 2}}
+
+
+def test_key_added():
+    assert compute_diff({"a": 1}, {"a": 1, "b": 2}) == {"b": 2}
+
+
+def test_key_removed():
+    assert compute_diff({"a": 1, "b": 2}, {"a": 1}) == {"$delete": ["b"]}
+
+
+def test_last_key_removed_is_delete_not_replace():
+    # jsondiff's minimal-diff search emits {"$replace": {}} here; we emit
+    # what actually happened: the key was deleted.
+    assert compute_diff({"a": 1}, {}) == {"$delete": ["a"]}
+    assert compute_diff({"x": {"y": 1}}, {"x": {}}) == {"x": {"$delete": ["y"]}}
+
+
+def test_list_changes_are_full_assignment():
+    # No $insert/$delete positional edits: lists are replaced wholesale.
+    assert compute_diff({"l": [1, 2]}, {"l": [1, 2, 3]}) == {"l": [1, 2, 3]}
+    assert compute_diff({"l": [1, 2, 3]}, {"l": [1, 3]}) == {"l": [1, 3]}
+    assert compute_diff({"l": [1]}, {"l": []}) == {"l": []}
+
+
+def test_list_with_unchanged_prefix_is_full_assignment():
+    diff = compute_diff({"l": ["a", "b", "c"]}, {"l": ["a", "x", "b", "c"]})
+    assert diff == {"l": ["a", "x", "b", "c"]}
+
+
+def test_type_changes_are_full_assignment():
+    assert compute_diff({"a": {"x": 1}}, {"a": [1]}) == {"a": [1]}
+    # A dict replacing a non-dict is a plain assignment too: the consumer
+    # sees from the old value whether to patch (dict) or replace.
+    assert compute_diff({"a": [1]}, {"a": {"x": 1}}) == {"a": {"x": 1}}
+    assert compute_diff({"a": 1}, {"a": None}) == {"a": None}
+
+
+def test_new_dict_value_assigned_wholesale():
+    assert compute_diff({}, {"a": {"x": 1}}) == {"a": {"x": 1}}
+
+
+def test_dollar_keys_escaped():
+    assert compute_diff({}, {"$weird": 1}) == {"$$weird": 1}
+    assert compute_diff({"$weird": 1}, {"$weird": 2}) == {"$$weird": 2}
+    assert compute_diff({"$weird": 1}, {}) == {"$delete": ["$$weird"]}
+
+
+def test_dollar_values_not_escaped():
+    # Only keys are escaped; values are stored verbatim, even "$delete".
+    assert compute_diff({"s": 1}, {"s": "$y"}) == {"s": "$y"}
+    assert compute_diff({"s": 1}, {"s": "$delete"}) == {"s": "$delete"}
+    assert compute_diff({}, {"o": {"s": "$y", "l": ["$z"]}}) == {
+        "o": {"s": "$y", "l": ["$z"]}
+    }
+
+
+# --- Consumer: apply_diff / patch_state -------------------------------------
+
+
+def test_patch_state_delegates():
+    assert patch_state({"a": 1}, {"a": 2}) == {"a": 2}
+
+
+def test_apply_scalar_and_add():
+    assert apply_diff({"a": 1}, {"a": 2, "b": 3}) == {"a": 2, "b": 3}
+
+
+def test_apply_delete():
+    assert apply_diff({"a": 1, "b": 2}, {"$delete": ["b"]}) == {"a": 1}
+    assert apply_diff({"a": 1}, {"$delete": ["a"]}) == {}
+
+
+def test_apply_replace():
+    assert apply_diff({"a": 1, "b": 2}, {"$replace": {"c": 3}}) == {"c": 3}
+    assert apply_diff({"x": {"a": 1}}, {"x": {"$replace": [1]}}) == {"x": [1]}
+
+
+def test_apply_nested_delete():
+    diff = {"x": {"$delete": ["y"]}}
+    assert apply_diff({"x": {"y": 2, "z": 3}}, diff) == {"x": {"z": 3}}
+
+
+def test_apply_list_insert():
+    diff = {"l": {"$insert": [[1, "x"]]}}
+    assert apply_diff({"l": ["a", "b"]}, diff) == {"l": ["a", "x", "b"]}
+
+
+def test_apply_list_delete():
+    diff = {"l": {"$delete": [1]}}
+    assert apply_diff({"l": ["a", "b", "c"]}, diff) == {"l": ["a", "c"]}
+
+
+def test_apply_list_delete_multiple_positions():
+    # jsondiff emits positions in descending order for sequential pops.
+    diff = {"l": {"$delete": [4, 2, 0]}}
+    assert apply_diff({"l": [0, 1, 2, 3, 4]}, diff) == {"l": [1, 3]}
+
+
+def test_apply_list_insert_and_delete():
+    diff = {"l": {"$insert": [[0, 9], [2, 8], [4, 7]], "$delete": [2, 0]}}
+    assert apply_diff({"l": [0, 1, 2, 3]}, diff) == {"l": [9, 1, 8, 3, 7]}
+
+
+def test_apply_list_per_index_nested_diff():
+    diff = {"l": {"1": {"y": 3}}}
+    state = {"l": [{"x": 1}, {"y": 2}]}
+    assert apply_diff(state, diff) == {"l": [{"x": 1}, {"y": 3}]}
+
+
+def test_apply_escaped_keys_and_values():
+    assert apply_diff({}, {"$$weird": 1}) == {"$weird": 1}
+    assert apply_diff({"$weird": 1}, {"$delete": ["$$weird"]}) == {}
+    # jsondiff escapes $-values as "$$.."; those are unescaped on apply.
+    assert apply_diff({"s": 1}, {"s": "$$y"}) == {"s": "$y"}
+    assert apply_diff({"s": 1}, {"s": "$$delete"}) == {"s": "$delete"}
+    # Our own producer stores values verbatim; single-$ stays as-is.
+    assert apply_diff({"s": 1}, {"s": "$y"}) == {"s": "$y"}
+    assert apply_diff({"s": 1}, {"s": "$delete"}) == {"s": "$delete"}
+    assert apply_diff({}, {"o": {"s": "$$y", "l": ["$$z"]}}) == {
+        "o": {"s": "$y", "l": ["$z"]}
+    }
+
+
+def test_apply_empty_diff():
+    assert apply_diff({"a": 1}, {}) == {"a": 1}
+
+
+def test_apply_diff_on_missing_state():
+    assert apply_diff({}, {"a": {"b": 1}}) == {"a": {"b": 1}}
+
+
+def test_apply_bare_dict_replaces_non_dict():
+    # Our own producer emits no $replace; a dict over a non-dict old value
+    # is a wholesale replacement.
+    assert apply_diff({"a": [1, 2]}, {"a": {"x": 1}}) == {"a": {"x": 1}}
+    assert apply_diff({"a": 5}, {"a": {"x": 1}}) == {"a": {"x": 1}}
+    assert apply_diff({"a": None}, {"a": {"x": 1}}) == {"a": {"x": 1}}
+
+
+def test_apply_list_patch_still_works_on_lists():
+    # jsondiff-style per-index diff keeps list-op semantics on list state.
+    assert apply_diff({"l": [1, 2]}, {"l": {"1": 9}}) == {"l": [1, 9]}
+
+
+# --- jsondiff compatibility, both directions --------------------------------
+
+COMPAT_CASES = [
+    ("scalar change", {"a": 1}, {"a": 2}),
+    ("key add", {"a": 1}, {"a": 1, "b": 2}),
+    ("key remove", {"a": 1, "b": 2}, {"a": 1}),
+    ("last key removed", {"a": 1}, {}),
+    ("nested delete", {"a": {"x": 1, "y": 2}}, {"a": {"x": 1}}),
+    ("nested mixed", {"a": {"x": 1, "y": 2}}, {"a": {"x": 9, "z": 3}}),
+    ("list append", {"l": [1, 2]}, {"l": [1, 2, 3]}),
+    ("list insert mid", {"l": [1, 2, 3]}, {"l": [1, 9, 2, 3]}),
+    ("list remove mid", {"l": [1, 2, 3]}, {"l": [1, 3]}),
+    ("list remove many", {"l": [0, 1, 2, 3, 4]}, {"l": [1, 3]}),
+    ("list replace all", {"l": [1, 2]}, {"l": [3, 4]}),
+    ("list insert+delete", {"l": [0, 1, 2, 3]}, {"l": [9, 1, 8, 3, 7]}),
+    ("dict in list", {"l": [{"x": 1}, {"y": 2}]}, {"l": [{"x": 1}, {"y": 3}]}),
+    ("list to empty", {"l": [1]}, {"l": []}),
+    ("type change dict->list", {"a": {"x": 1}}, {"a": [1]}),
+    ("type change list->dict", {"a": [1]}, {"a": {"x": 1}}),
+    ("dollar key", {"$k": 1, "b": 1}, {"$k": 2}),
+    ("dollar value", {"s": "$x"}, {"s": "$y"}),
+    (
+        "deep nesting",
+        {"a": {"b": {"c": {"d": 1, "e": 2}}}},
+        {"a": {"b": {"c": {"d": 9}}}},
+    ),
+]
+
+
+# jsondiff.patch cannot apply our patches for "type change list->dict"
+# (we emit a bare dict where jsondiff needs $replace) and "dollar value"
+# (we do not escape "$"-prefixed values; jsondiff.patch would strip the
+# "$"), so those cases are excluded from this direction.
+JSONDIFF_APPLIES_CASES = [
+    c for c in COMPAT_CASES if c[0] not in {"type change list->dict", "dollar value"}
+]
+
+
+@pytest.mark.parametrize(
+    "name,old,new", JSONDIFF_APPLIES_CASES, ids=[c[0] for c in JSONDIFF_APPLIES_CASES]
+)
+def test_jsondiff_applies_our_patches(name, old, new):
+    diff = compute_diff(old, new)
+    assert diff is not None
+    assert jsondiff.patch(old, diff, marshal=True) == new
+
+
+@pytest.mark.parametrize("name,old,new", COMPAT_CASES, ids=[c[0] for c in COMPAT_CASES])
+def test_we_apply_jsondiff_patches(name, old, new):
+    diff = jsondiff.diff(old, new, marshal=True)
+    assert apply_diff(old, diff) == new
+
+
+@pytest.mark.parametrize("name,old,new", COMPAT_CASES, ids=[c[0] for c in COMPAT_CASES])
+def test_our_own_round_trip(name, old, new):
+    diff = compute_diff(old, new)
+    assert diff is not None
+    assert apply_diff(old, diff) == new
+
+
+def test_no_diff_means_equal_states():
+    for _name, old, new in COMPAT_CASES:
+        assert compute_diff(old, new) is not None  # cases really differ
+    assert compute_diff({"a": [1, {"b": "$x"}]}, {"a": [1, {"b": "$x"}]}) is None
+
+
+# --- Logging ----------------------------------------------------------------
+
+
+def test_format_diff_list_edit_shows_whole_list():
+    diff = jsondiff.diff({"l": [1, 2, 3]}, {"l": [1, 9, 3]}, marshal=True)
+    lines = format_diff(diff, previous={"l": [1, 2, 3]})
+    text = "\n".join(lines)
+    assert "$insert" not in text
+    assert "[1, 9, 3]" in text
+
+
+def test_format_diff_unescapes_dollar_keys():
+    lines = format_diff({"$$weird": 1}, previous={})
+    assert any("$weird" in line and "$$weird" not in line for line in lines)
@@ -0,0 +1,55 @@
+"""Tests for LockedFile low-level behaviors."""
+
+from kanta.exceptions import FileLockError
+from kanta.filelock import LockedFile
+
+
+def test_replace_content_rewrites_in_place(tmp_path):
+    path = tmp_path / "data.kantadb"
+    path.write_bytes(b"original content here")
+
+    f = LockedFile()
+    f.open(path)
+    try:
+        f.replace_content(b"new")
+        assert f.size() == 3
+        f.write(b"!")
+    finally:
+        f.close()
+
+    assert path.read_bytes() == b"new!"
+
+
+def test_replace_content_keeps_lock(tmp_path):
+    path = tmp_path / "data.kantadb"
+    path.write_bytes(b"abc")
+
+    f = LockedFile()
+    f.open(path)
+    try:
+        f.replace_content(b"xyz")
+        other = LockedFile()
+        try:
+            other.open(path)
+            raise AssertionError("second open should fail while lock is held")
+        except FileLockError:
+            pass
+    finally:
+        f.close()
+
+
+def test_replace_content_grow_and_shrink(tmp_path):
+    path = tmp_path / "data.kantadb"
+    path.write_bytes(b"x" * 100)
+
+    f = LockedFile()
+    f.open(path)
+    try:
+        f.replace_content(b"")
+        assert f.size() == 0
+        f.replace_content(b"y" * 200)
+        assert f.size() == 200
+    finally:
+        f.close()
+
+    assert path.read_bytes() == b"y" * 200
@@ -47,7 +47,7 @@ async def test_new_file_writes_bootstrap_record_without_handlers(
    records = read_changes(path, format_config)
    assert len(records) == 1
    assert records[0].a == "bootstrap"
-    assert records[0].diff == {"$replace": {"users": {}, "counter": 0}}
+    assert records[0].diff == {"users": {}, "counter": 0}


@pytest.mark.asyncio
@@ -63,7 +63,8 @@ async def test_new_file_persists_initial_state_for_roundtrip(tmp_path, format_co
    assert len(records) == 1
    assert records[0].a == "bootstrap"
    assert records[0].diff == {
-        "$replace": {"users": {"alice": {"name": "Alice", "age": 0}}, "counter": 5}
+        "users": {"alice": {"name": "Alice", "age": 0}},
+        "counter": 5,
    }

    kanta2 = make_kanta(path, Data, format_config)
@@ -183,7 +183,7 @@ def test_apply_returns_change_information():
    assert result.migrations[0].name == "migrate_v1"
    assert result.migrations[0].description == "Set x"
    assert result.migrations[0].changed is True
-    assert result.migrations[0].diff == {"$replace": {"x": 1}}
+    assert result.migrations[0].diff == {"x": 1}

    assert result.migrations[1].name == "migrate_v2"
    assert result.migrations[1].description == "No-op"
@@ -0,0 +1,214 @@
+"""Tests for retention-based database rotation (docs/rotation.md)."""
+
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+
+import pytest
+
+from kanta.structs import ChangeRecord, Snapshot
+from tests.support import Data, make_kanta, read_changes
+
+pytestmark = pytest.mark.asyncio
+
+DAY = timedelta(days=1)
+T0 = datetime(2026, 1, 1, tzinfo=UTC)
+
+
+def make_clock(cell: list[datetime]):
+    def clock() -> datetime:
+        return cell[0]
+
+    return clock
+
+
+async def write_history(path: Path, format_config, days: list[int]) -> None:
+    """Write one change per day offset (relative to T0) with a fake clock."""
+    cell = [T0 + (days[0] - 1) * DAY]  # bootstrap predates all history
+    kanta = make_kanta(path, Data, format_config)
+    kanta.clock(make_clock(cell))
+    await kanta.open(log=False)
+    for day in days:
+        cell[0] = T0 + day * DAY
+        with kanta.transaction(f"day{day}", log=False) as data:
+            data.counter += 1
+        await kanta.flush()
+    await kanta.close()
+
+
+def read_all(path: Path, format_config):
+    """All records (changes and snapshots) in file order."""
+    _, serializer_cls = format_config
+    serializer = serializer_cls()
+    framer = serializer.framer_cls()
+    out = []
+    for is_snapshot, payload, _, _ in framer.iter_records(path.read_bytes(), 0):
+        out.append(
+            serializer.decode(payload, type=Snapshot if is_snapshot else ChangeRecord)
+        )
+    return out
+
+
+def rotated_files(path: Path) -> list[Path]:
+    return sorted(path.parent.glob(f"{path.stem}@*.kantadb"))
+
+
+async def test_rotation_splits_history(tmp_path, format_config):
+    path = tmp_path / "data.kantadb"
+    await write_history(path, format_config, days=[-40, -20, -5])
+
+    cell = [T0]
+    kanta = make_kanta(path, Data, format_config, retention=30 * DAY)
+    kanta.clock(make_clock(cell))
+    await kanta.open(log=False)
+    assert kanta.data.counter == 3
+    await kanta.close()
+
+    rotated = rotated_files(path)
+    assert len(rotated) == 1
+
+    # Main file: leading snapshot (ts = last dropped record), the retained
+    # changes, and no final snapshot (too few retained changes).
+    records = read_all(path, format_config)
+    assert isinstance(records[0], Snapshot)
+    assert records[0].ts == T0 - 40 * DAY
+    assert records[0].state["counter"] == 1
+    changes = [r for r in records if isinstance(r, ChangeRecord)]
+    assert [c.a for c in changes] == ["day-20", "day-5"]
+
+    # Rotated file holds exactly the dropped history, ending at the last
+    # dropped record whose ts matches the filename.
+    stamp = (T0 - 40 * DAY).strftime("%Y%m%dT%H%M%SZ")
+    assert rotated[0].name == f"data@{stamp}.kantadb"
+    rrecords = read_all(rotated[0], format_config)
+    assert [r.a for r in rrecords] == ["bootstrap", "day-40"]
+
+
+async def test_rotation_reopens_cleanly_and_does_not_rerotate(tmp_path, format_config):
+    path = tmp_path / "data.kantadb"
+    await write_history(path, format_config, days=[-40, -5])
+
+    cell = [T0]
+    for expected_changes in (["day-5"], ["day-5"]):
+        kanta = make_kanta(path, Data, format_config, retention=30 * DAY)
+        kanta.clock(make_clock(cell))
+        async with kanta:
+            assert kanta.data.counter == 2
+        assert [c.a for c in read_changes(path, format_config)] == expected_changes
+
+    # Second open found a file whose history already fits the window.
+    assert len(rotated_files(path)) == 1
+
+
+async def test_rotation_noop_when_retention_covers_all(tmp_path, format_config):
+    path = tmp_path / "data.kantadb"
+    await write_history(path, format_config, days=[-5])
+    before = path.read_bytes()
+
+    cell = [T0]
+    kanta = make_kanta(path, Data, format_config, retention=30 * DAY)
+    kanta.clock(make_clock(cell))
+    async with kanta:
+        assert kanta.data.counter == 1
+
+    assert rotated_files(path) == []
+    assert path.read_bytes() == before
+
+
+async def test_rotation_aged_out_database_reduces_to_single_snapshot(
+    tmp_path, format_config
+):
+    path = tmp_path / "data.kantadb"
+    await write_history(path, format_config, days=[-40, -35])
+
+    cell = [T0]
+    kanta = make_kanta(path, Data, format_config, retention=30 * DAY)
+    kanta.clock(make_clock(cell))
+    async with kanta:
+        assert kanta.data.counter == 2
+
+    records = read_all(path, format_config)
+    assert len(records) == 1
+    assert isinstance(records[0], Snapshot)
+    assert records[0].state["counter"] == 2
+
+    # Opening again must not rotate the snapshot-only file.
+    before = path.read_bytes()
+    kanta = make_kanta(path, Data, format_config, retention=30 * DAY)
+    kanta.clock(make_clock(cell))
+    async with kanta:
+        assert kanta.data.counter == 2
+    assert path.read_bytes() == before
+    assert len(rotated_files(path)) == 1
+
+
+async def test_rotation_validates_against_internal_snapshots(tmp_path, format_config):
+    path = tmp_path / "data.kantadb"
+    cell = [T0 - 40 * DAY]
+    kanta = make_kanta(path, Data, format_config)
+    kanta.clock(make_clock(cell))
+    await kanta.open(log=False)
+    with kanta.transaction("old", log=False) as data:
+        data.counter = 1
+    await kanta.flush()
+    kanta.request_snapshot()
+    kanta._impl.maybe_snapshot()
+    cell[0] = T0 - 1 * DAY
+    with kanta.transaction("new", log=False) as data:
+        data.counter = 2
+    await kanta.flush()
+    await kanta.close()
+
+    cell[0] = T0
+    kanta = make_kanta(path, Data, format_config, retention=30 * DAY)
+    kanta.clock(make_clock(cell))
+    async with kanta:
+        assert kanta.data.counter == 2
+
+    records = read_all(path, format_config)
+    assert isinstance(records[0], Snapshot)
+    assert records[0].state["counter"] == 1
+    assert [r.a for r in records if isinstance(r, ChangeRecord)] == ["new"]
+
+
+@pytest.mark.parametrize("name", ["data", "data.db", "data.kantadb"])
+async def test_rotated_naming_normalizes_extension(tmp_path, format_config, name):
+    path = tmp_path / name
+    await write_history(path, format_config, days=[-40, -5])
+
+    cell = [T0]
+    kanta = make_kanta(path, Data, format_config, retention=30 * DAY)
+    kanta.clock(make_clock(cell))
+    async with kanta:
+        pass
+
+    stamp = (T0 - 40 * DAY).strftime("%Y%m%dT%H%M%SZ")
+    assert (tmp_path / f"data@{stamp}.kantadb").exists()
+
+
+async def test_retention_accepts_int_days(tmp_path, format_config):
+    path = tmp_path / "data.kantadb"
+    await write_history(path, format_config, days=[-40, -5])
+
+    cell = [T0]
+    kanta = make_kanta(path, Data, format_config, retention=30)
+    kanta.clock(make_clock(cell))
+    async with kanta:
+        assert kanta.data.counter == 2
+
+    assert len(rotated_files(path)) == 1
+    assert [c.a for c in read_changes(path, format_config)] == ["day-5"]
+
+
+async def test_rotation_disabled_by_default(tmp_path, format_config):
+    path = tmp_path / "data.kantadb"
+    await write_history(path, format_config, days=[-40, -5])
+    before = path.read_bytes()
+
+    cell = [T0]
+    kanta = make_kanta(path, Data, format_config)
+    kanta.clock(make_clock(cell))
+    async with kanta:
+        assert kanta.data.counter == 2
+
+    assert rotated_files(path) == []
+    assert path.read_bytes() == before