From b8285e91709d4ff461c13ac9500820945ec14c35 Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Wed, 2 Sep 2026 16:30:00 +0000 Subject: [PATCH] rotation: tests, replay-base cutoff state fix, formatting - tests/test_rotation.py covers both framers: history split, idempotent reopen, no-op cases, aged-out reduction to a single snapshot, internal snapshot validation, extension-agnostic rotated naming, and the disabled default. - Fix plan_rotation to seed the cutoff state from the replay base snapshot (previously asserted when the base already covered the cutoff). - docs: snapshots carry the schema version in effect at their position, keeping replay/migration behavior identical to the unrotated file. --- docs/rotation.md | 14 +-- kanta/kantaimpl.py | 4 +- kanta/rotation.py | 14 +-- tests/test_rotation.py | 200 +++++++++++++++++++++++++++++++++++++++++ 4 files changed, 220 insertions(+), 12 deletions(-) create mode 100644 tests/test_rotation.py diff --git a/docs/rotation.md b/docs/rotation.md index 7a0405f..8d9f6a6 100644 --- a/docs/rotation.md +++ b/docs/rotation.md @@ -155,16 +155,18 @@ into memory by `open_and_read`; no second disk read is needed. 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 **current - schema version**. 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. + 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 current schema version — written **only if** there were + 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 diff --git a/kanta/kantaimpl.py b/kanta/kantaimpl.py index 220ef3d..952c3c1 100644 --- a/kanta/kantaimpl.py +++ b/kanta/kantaimpl.py @@ -399,7 +399,9 @@ class KantaImpl(PersistenceMixin, Generic[T]): ) if plan is None: return content - await asyncio.to_thread(execute_rotation, self.filename, self.file, plan, log=log) + await asyncio.to_thread( + execute_rotation, self.filename, self.file, plan, log=log + ) return plan.new_content async def close(self) -> None: diff --git a/kanta/rotation.py b/kanta/rotation.py index 410e535..58f97c9 100644 --- a/kanta/rotation.py +++ b/kanta/rotation.py @@ -67,9 +67,7 @@ class _Entry: self.end_pos = end_pos -def _scan( - content: bytes, *, framer: Framer, serializer: Serializer -) -> list[_Entry]: +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] = [] @@ -131,7 +129,11 @@ def plan_rotation( version = snap.v m = snap.m - state_at_cutoff: dict[str, Any] | None = None + # 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 @@ -173,7 +175,9 @@ def plan_rotation( # 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) + 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: diff --git a/tests/test_rotation.py b/tests/test_rotation.py new file mode 100644 index 0000000..08d38a0 --- /dev/null +++ b/tests/test_rotation.py @@ -0,0 +1,200 @@ +"""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%S.%fZ") + 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%S.%fZ") + assert (tmp_path / f"data@{stamp}.kantadb").exists() + + +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