diff --git a/kanta/kantaimpl.py b/kanta/kantaimpl.py index 5f83a3c..eae4822 100644 --- a/kanta/kantaimpl.py +++ b/kanta/kantaimpl.py @@ -110,6 +110,9 @@ class KantaImpl(PersistenceMixin, Generic[T]): action="open", ) + # From this point the file is open and must be closed via close(). + self.opened = True + if content: try: rr = replay( @@ -140,12 +143,24 @@ class KantaImpl(PersistenceMixin, Generic[T]): ) from e migrations_ran = False + state_before_migrations = None if self.migrations is not None: previous_version = rr.version + state_before_migrations = copy.deepcopy(rr.state) rr.version = self.migrations.apply(rr.state, rr.version, self._kanta) migrations_ran = rr.version != previous_version - self.statedict = copy.deepcopy(rr.state) + self.snapshot.ts = ( + datetime.fromtimestamp(rr.last_snapshot_mtime, UTC) + if rr.last_snapshot_mtime is not None + else None + ) + + self.statedict = copy.deepcopy( + state_before_migrations + if state_before_migrations is not None + else rr.state + ) self.data = restore_data_in_place( self.data, rr.state, @@ -158,20 +173,24 @@ class KantaImpl(PersistenceMixin, Generic[T]): if self.readonly: self.statedict = copy.deepcopy(normalized) else: - if migrations_ran: - self.queue_change( + migration_record = None + if migrations_ran and self.statedict != rr.state: + migration_record = self.queue_change( f"migrate:v{self.version}", - self.statedict, + rr.state, mtime=False, - force=True, ) - self.queue_change("migrate:msgspec", normalized, mtime=False) - self.snapshot.ts = ( - datetime.fromtimestamp(rr.last_snapshot_mtime, UTC) - if rr.last_snapshot_mtime is not None - else None - ) + msgspec_record = self.queue_change( + "migrate:msgspec", normalized, mtime=False + ) + if migrations_ran or msgspec_record is not None: + self.snapshot.request_force() + await self.flush() + self.snapshot.maybe_write( + self.file, self.version, self.statedict, m=self.mtime + ) elif self.readonly: + self.opened = False self.file.close() raise DataIntegrityError( "Cannot open empty database in read-only mode", @@ -196,6 +215,7 @@ class KantaImpl(PersistenceMixin, Generic[T]): force=True, ) except Exception: + self.opened = False self.file.close() try: await asyncio.to_thread(self.filename.unlink, missing_ok=True) @@ -203,8 +223,6 @@ class KantaImpl(PersistenceMixin, Generic[T]): pass raise - self.opened = True - if not self.readonly: self.background_task = asyncio.create_task(self._background_loop()) diff --git a/kanta/snapshot.py b/kanta/snapshot.py index 7f51d92..8896f50 100644 --- a/kanta/snapshot.py +++ b/kanta/snapshot.py @@ -41,15 +41,16 @@ class SnapshotState: self, file, version: int, state: dict, m: datetime | None = None ) -> None: """Write snapshot when thresholds/time policy allows it.""" - if self.changes < self._min_diffs: - return 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 force: + if self.changes < self._min_diffs: + return + if now.weekday() != 6: # 6 = Sunday + return + sunday_midnight = now.replace(hour=0, minute=0, second=0, microsecond=0) + if self.ts is not None and self.ts >= sunday_midnight: + return if not file.is_open: return try: diff --git a/kanta/structs.py b/kanta/structs.py index 401069e..80afd61 100644 --- a/kanta/structs.py +++ b/kanta/structs.py @@ -23,7 +23,7 @@ class ChangeRecord(msgspec.Struct, omit_defaults=True, kw_only=True): v: int = 0 u: str | None = None m: datetime | None = None - diff: dict + diff: dict = {} class Snapshot(msgspec.Struct, omit_defaults=True): diff --git a/tests/support.py b/tests/support.py index e087008..4a60ba5 100644 --- a/tests/support.py +++ b/tests/support.py @@ -7,7 +7,7 @@ from uuid import UUID import msgspec from kanta.kanta import Kanta -from kanta.structs import ChangeRecord +from kanta.structs import ChangeRecord, Snapshot class User(msgspec.Struct): @@ -89,6 +89,17 @@ def make_migrations_module(name: str, fn_name: str, fn): return mod +def read_last_snapshot(path: Path, format_config) -> Snapshot | None: + _, serializer_cls = format_config + serializer = serializer_cls() + framer = serializer.framer_cls() + data = path.read_bytes() + payload, _, _ = framer.scan_last_snapshot(data) + if payload is None: + return None + return serializer.decode(payload, type=Snapshot) + + 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 diff --git a/tests/test_kanta_integration.py b/tests/test_kanta_integration.py index fa6e444..c4d48ef 100644 --- a/tests/test_kanta_integration.py +++ b/tests/test_kanta_integration.py @@ -19,6 +19,7 @@ from .support import ( make_kanta, make_migrations_module, read_changes, + read_last_snapshot, seed_single_change, ) @@ -70,6 +71,27 @@ async def test_new_file_persists_initial_state_for_roundtrip(tmp_path, format_co await kanta2.close() +@pytest.mark.asyncio +async def test_reopen_without_changes_does_not_force_snapshot( + tmp_path, format_config +): + path = tmp_path / "test.db" + kanta = make_kanta(path, Data(counter=5), format_config) + await kanta.open() + await kanta.close() + + # No snapshot should exist after the initial bootstrap and close. + assert read_last_snapshot(path, format_config) is None + + kanta2 = make_kanta(path, Data, format_config) + await kanta2.open() + assert kanta2.data.counter == 5 + await kanta2.close() + + # Re-opening without migrations or normalization changes must not force one. + assert read_last_snapshot(path, format_config) is None + + @pytest.mark.asyncio async def test_open_overwrites_caller_owned_root_data(tmp_path, format_config): path = tmp_path / "test.db" @@ -475,7 +497,9 @@ async def test_msgspec_normalization_logs_migration(tmp_path, format_config): @pytest.mark.asyncio -async def test_empty_migration_is_recorded_and_not_reapplied(tmp_path, format_config): +async def test_empty_migration_writes_snapshot_and_is_not_reapplied( + tmp_path, format_config +): path = tmp_path / "test.db" seed_single_change( path, fixed_change("init", {"counter": 0, "users": {}}), format_config @@ -491,26 +515,67 @@ async def test_empty_migration_is_recorded_and_not_reapplied(tmp_path, format_co kanta = make_kanta(path, Data, format_config, migrations=mod) await kanta.open() assert kanta.version == 1 - await kanta.flush() await kanta.close() + # Empty migrations must not produce empty change records. records = read_changes(path, format_config) migration_records = [r for r in records if r.a.startswith("migrate")] - assert len(migration_records) == 1 - assert migration_records[0].v == 1 - assert migration_records[0].diff == {} + assert not migration_records + + # The version bump is persisted via a snapshot instead. + snap = read_last_snapshot(path, format_config) + assert snap is not None + assert snap.v == 1 + assert snap.state == {"counter": 0, "users": {}} kanta2 = make_kanta(path, Data, format_config, migrations=mod) await kanta2.open() assert kanta2.version == 1 await kanta2.close() + # Re-opening must not create additional migration records or snapshots. records2 = read_changes(path, format_config) - assert len([r for r in records2 if r.a.startswith("migrate")]) == 1 + assert not [r for r in records2 if r.a.startswith("migrate")] finally: sys.modules.pop("empty_migration_mod", None) +@pytest.mark.asyncio +async def test_migration_with_changes_records_diff_and_snapshot( + tmp_path, format_config +): + path = tmp_path / "test.db" + seed_single_change(path, fixed_change("init", {"counter": 0}), format_config) + + mod = type(sys)("test_migrations_changes") + + def migrate_v1(d, kanta): + d["counter"] = 2 + + mod.__dict__["migrate_v1"] = migrate_v1 + + kanta = make_kanta(path, Data, format_config, migrations=mod) + await kanta.open() + assert kanta.version == 1 + assert kanta.data.counter == 2 + await kanta.close() + + records = read_changes(path, format_config) + migration_records = [r for r in records if r.a.startswith("migrate")] + assert len(migration_records) == 2 + assert migration_records[0].a == "migrate:v1" + assert migration_records[0].v == 1 + assert migration_records[0].diff == {"counter": 2} + assert migration_records[1].a == "migrate:msgspec" + assert migration_records[1].v == 1 + assert migration_records[1].diff == {"users": {}} + + snap = read_last_snapshot(path, format_config) + assert snap is not None + assert snap.v == 1 + assert snap.state == {"counter": 2, "users": {}} + + @pytest.mark.asyncio async def test_open_locked_file_raises_filelock_error(tmp_path, format_config): path = tmp_path / "test.db" diff --git a/tests/test_snapshot.py b/tests/test_snapshot.py index cd4a688..6a0bc2e 100644 --- a/tests/test_snapshot.py +++ b/tests/test_snapshot.py @@ -32,3 +32,20 @@ def test_force_writes(): f = FakeFile() ss.maybe_write(f, 1, {"x": 1}) assert len(f.written) == 1 + + +def test_force_bypasses_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=100) + ss.record_changes(5) + ss.request_force() + f = FakeFile() + ss.maybe_write(f, 1, {"x": 1}) + assert len(f.written) == 1