From 74150b783cb59d11a1af2b5f76e7147b4bac6f0a Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Sat, 13 Jun 2026 00:49:35 +0000 Subject: [PATCH] Proper ts/mtime setting and tracking. Added Kanta.mtime property for reading current change time. --- docs/database.md | 17 ++++- kanta/kanta.py | 26 ++++++- kanta/kantaimpl.py | 3 +- kanta/persistence.py | 62 ++++++++++++---- kanta/serialization/base.py | 5 -- kanta/snapshot.py | 12 ++- kanta/transaction.py | 9 ++- tests/test_mtime.py | 143 ++++++++++++++++++++++++++++++++++++ 8 files changed, 247 insertions(+), 30 deletions(-) create mode 100644 tests/test_mtime.py diff --git a/docs/database.md b/docs/database.md index 317c7b9..e920603 100644 --- a/docs/database.md +++ b/docs/database.md @@ -82,15 +82,30 @@ history. ## Transaction Semantics - `kanta.transaction(action=...)` captures a pre-transaction snapshot dict. +- By default a transaction updates the modification time `m` to the current UTC + time. +- `mtime=True|False|datetime` controls the modification time `m`: + - `True` (default) sets `m` to the current UTC time. + - `False` omits `m`, leaving the previous modification time in effect. + - A `datetime` sets `m` to that explicit value. +- System operations such as `migrate:msgspec` use `mtime=False` so they are not + considered modifications and do not advance `m`. - On success: - compute diff between previous builtins and current builtins, - - queue a `ChangeRecord` if non-empty. + - queue a `ChangeRecord` if non-empty, + - update `kanta.mtime` when the change carries an `m` value. - On exception: - restore in-memory data from snapshot, - re-raise the exception. Nested transactions are rejected. +## Modification Time + +`kanta.mtime` exposes the last modification time carried forward from change +records. It is updated by normal transactions and preserved across snapshots and +reloads, while system operations such as migrations leave it unchanged. + ## Flush and Lifecycle - Writes are queued in memory. diff --git a/kanta/kanta.py b/kanta/kanta.py index 6bf596e..17ff4d4 100644 --- a/kanta/kanta.py +++ b/kanta/kanta.py @@ -3,6 +3,7 @@ from __future__ import annotations from collections.abc import Callable +from datetime import datetime from pathlib import Path from types import ModuleType from typing import Any, Generic, TypeVar @@ -132,6 +133,17 @@ class Kanta(Generic[T]): """ return self._impl.filename + @property + def mtime(self) -> datetime | None: + """Last modification time carried forward from change records. + + Returns: + The latest ``m`` value, or ``None`` if no modification time has + been set yet. System operations such as migrations do not update + this value. + """ + return self._impl.mtime + async def open(self) -> None: """Open the database file and start background persistence. @@ -184,6 +196,7 @@ class Kanta(Generic[T]): user: str | None = None, user_display: str | None = None, resolver: Any = None, + mtime: bool | datetime = True, ): """Create a transactional mutation context manager. @@ -192,6 +205,12 @@ class Kanta(Generic[T]): user: Optional user identifier stored in metadata. user_display: Optional display name used for logging/resolution. resolver: Optional callable for resolving identifiers in logs. + mtime: Controls the modification time ``m``. ``True`` (default) + sets ``m`` to the current UTC time. ``False`` omits ``m`` so the + previous modification time remains in effect; this is used for + system operations that are not considered modifications. A + :class:`~datetime.datetime` value sets ``m`` to that explicit + time. Returns: A context manager yielding the live state object for mutation. @@ -202,5 +221,10 @@ class Kanta(Generic[T]): rolled back. """ return _transaction( - self._impl, action, user=user, user_display=user_display, resolver=resolver + self._impl, + action, + user=user, + user_display=user_display, + resolver=resolver, + mtime=mtime, ) diff --git a/kanta/kantaimpl.py b/kanta/kantaimpl.py index f695feb..e97fa74 100644 --- a/kanta/kantaimpl.py +++ b/kanta/kantaimpl.py @@ -104,8 +104,9 @@ class KantaImpl(PersistenceMixin, Generic[T]): serializer=self.serializer, ) self.version = rr.version + self.mtime = rr.m normalized = struct_to_dict(self.data, serializer=self.serializer) - self.queue_change("migrate:msgspec", normalized) + 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 diff --git a/kanta/persistence.py b/kanta/persistence.py index a66c440..ccc85ef 100644 --- a/kanta/persistence.py +++ b/kanta/persistence.py @@ -8,7 +8,7 @@ import logging import threading from collections import deque from collections.abc import Callable -from datetime import datetime +from datetime import UTC, datetime from pathlib import Path from typing import Any @@ -41,6 +41,7 @@ class PersistenceMixin: flush_interval: float version: int opened: bool + mtime: datetime | None def __init__(self, **kwargs: Any) -> None: """Initialize persistence-owned state used by mixin methods.""" @@ -63,6 +64,7 @@ class PersistenceMixin: self.background_error = None self.flush_interval = flush_interval self.version = 0 + self.mtime: datetime | None = None async def _background_loop(self) -> None: """Background task that periodically flushes changes to disk.""" @@ -89,30 +91,60 @@ class PersistenceMixin: def maybe_snapshot(self) -> None: """Evaluate and possibly write a snapshot from current state.""" - self.snapshot.maybe_write(self.file, self.version, self.statedict) + self.snapshot.maybe_write(self.file, self.version, self.statedict, m=self.mtime) def queue_change( self, action: str, current: dict, + *, user: str | None = None, - m: datetime | None = None, - ) -> None: - """Queue a change record internally (thread-safe).""" + mtime: bool | datetime = True, + ) -> ChangeRecord | None: + """Queue a change record internally (thread-safe). + + Args: + action: Action label stored in the change record. + current: New serialized state after the change. + user: Optional actor identifier. + mtime: Controls the modification timestamp. ``True`` (default) + sets ``m`` to the current UTC time. ``False`` omits ``m`` so the + previous modification time remains in effect; this is used for + system operations that are not considered modifications. A + :class:`~datetime.datetime` value sets ``m`` to that explicit time. + + Returns: + The queued :class:`ChangeRecord`, or ``None`` if the diff was empty. + """ + now = datetime.now(UTC) + + if mtime is True: + m = now + elif mtime is False: + m = None + elif isinstance(mtime, datetime): + m = mtime + else: + raise TypeError("mtime must be True, False, or a datetime") + diff = compute_diff(self.statedict, current) if not diff: - return + return None + + record = ChangeRecord( + ts=now, + a=action, + v=self.version, + u=user, + m=m, + diff=diff, + ) with self.pending_lock: - self.pending_changes.append( - ChangeRecord( - a=action, - v=self.version, - u=user, - m=m, - diff=diff, - ) - ) + self.pending_changes.append(record) self.statedict = copy.deepcopy(current) + if m is not None: + self.mtime = m + return record def flush_sync(self) -> None: """Synchronously flush all pending changes to disk.""" diff --git a/kanta/serialization/base.py b/kanta/serialization/base.py index f8b405a..9008c3e 100644 --- a/kanta/serialization/base.py +++ b/kanta/serialization/base.py @@ -23,14 +23,12 @@ class ReplayResult: 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 @@ -63,7 +61,6 @@ def replay( 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: @@ -112,14 +109,12 @@ def replay( 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, ) diff --git a/kanta/snapshot.py b/kanta/snapshot.py index c859724..7f51d92 100644 --- a/kanta/snapshot.py +++ b/kanta/snapshot.py @@ -37,7 +37,9 @@ class SnapshotState: def record_changes(self, count: int) -> None: self.changes += count - def maybe_write(self, file, version: int, state: dict) -> None: + def maybe_write( + 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 @@ -51,14 +53,16 @@ class SnapshotState: if not file.is_open: return try: - self._write(file, version, state, now) + self._write(file, version, state, now, m=m) 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: + def _write( + self, file, version: int, state: dict, now: datetime, m: datetime | None = None + ) -> None: """Write a snapshot and update internal state.""" - payload = self._serializer.encode(Snapshot(ts=now, v=version, state=state)) + payload = self._serializer.encode(Snapshot(ts=now, v=version, state=state, m=m)) record_offset = file.size() if hasattr(file, "size") else 0 file.write(self._framer.frame_snapshot(payload, record_offset=record_offset)) self.changes = 0 diff --git a/kanta/transaction.py b/kanta/transaction.py index 77263bb..8e37afb 100644 --- a/kanta/transaction.py +++ b/kanta/transaction.py @@ -4,6 +4,7 @@ from __future__ import annotations import logging from contextlib import contextmanager +from datetime import datetime from typing import Any from kanta.diff import compute_diff @@ -22,6 +23,7 @@ def transaction( user: str | None = None, user_display: str | None = None, resolver: Any = None, + mtime: bool | datetime = True, ): """Wrap writes in a transaction and yield the live db object.""" if impl.in_transaction: @@ -58,9 +60,10 @@ def transaction( 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 + previous = impl.statedict + record = impl.queue_change(action, new_dict, user=user, mtime=mtime) + if record is not None: + log_change(action, record.diff, user_display, previous, resolver) except Exception: _logger.warning("Transaction '%s' failed, rolling back changes", action) if impl.transaction_snapshot is not None: diff --git a/tests/test_mtime.py b/tests/test_mtime.py new file mode 100644 index 0000000..9fc759a --- /dev/null +++ b/tests/test_mtime.py @@ -0,0 +1,143 @@ +"""Tests for mtime handling and the public ``kanta.mtime`` property.""" + +from datetime import UTC, datetime + +import pytest + +from kanta import ChangeRecord + +from .support import Data, make_kanta, seed_single_change + + +def _read_last_change(path, format_config): + name, serializer_cls = format_config + serializer = serializer_cls() + framer = serializer.framer_cls() + last = None + for is_snapshot, payload, _, _ in framer.iter_records(path.read_bytes(), 0): + if is_snapshot: + continue + last = serializer.decode(payload, type=ChangeRecord) + assert last is not None + return last + + +@pytest.mark.asyncio +async def test_default_transaction_updates_mtime(tmp_path, format_config): + path = tmp_path / "test.db" + kanta = make_kanta(path, Data, format_config) + await kanta.open() + + before = datetime.now(UTC) + with kanta.transaction(action="inc") as data: + data.counter = 1 + await kanta.flush() + await kanta.close() + + rec = _read_last_change(path, format_config) + assert rec.ts == rec.m + assert before <= rec.m <= datetime.now(UTC) + assert kanta.mtime == rec.m + + +@pytest.mark.asyncio +async def test_transaction_custom_mtime(tmp_path, format_config): + path = tmp_path / "test.db" + kanta = make_kanta(path, Data, format_config) + await kanta.open() + + custom_m = datetime(2026, 1, 1, 8, 0, tzinfo=UTC) + with kanta.transaction(action="inc", mtime=custom_m) as data: + data.counter = 1 + await kanta.flush() + await kanta.close() + + rec = _read_last_change(path, format_config) + assert rec.m == custom_m + assert kanta.mtime == custom_m + + +@pytest.mark.asyncio +async def test_transaction_mtime_false_preserves_mtime(tmp_path, format_config): + path = tmp_path / "test.db" + kanta = make_kanta(path, Data, format_config) + await kanta.open() + + first_m = datetime(2026, 1, 1, 10, 0, tzinfo=UTC) + with kanta.transaction(action="first", mtime=first_m) as data: + data.counter = 1 + + with kanta.transaction(action="second", mtime=False) as data: + data.counter = 2 + + await kanta.flush() + await kanta.close() + + records = [] + name, serializer_cls = format_config + serializer = serializer_cls() + framer = serializer.framer_cls() + for is_snapshot, payload, _, _ in framer.iter_records(path.read_bytes(), 0): + if is_snapshot: + continue + records.append(serializer.decode(payload, type=ChangeRecord)) + + assert records[0].m == first_m + assert records[1].m is None + assert kanta.mtime == first_m + + +@pytest.mark.asyncio +async def test_migration_does_not_update_mtime(tmp_path, format_config): + path = tmp_path / "test.db" + seed_m = datetime(2025, 12, 31, 23, 0, tzinfo=UTC) + seed_single_change( + path, + ChangeRecord( + ts=seed_m, + m=seed_m, + a="seed", + v=0, + diff={"counter": 0}, + ), + format_config, + ) + + kanta = make_kanta(path, Data, format_config) + await kanta.open() + + assert kanta.mtime == seed_m + + new_m = datetime(2026, 1, 5, 10, 0, tzinfo=UTC) + with kanta.transaction(action="inc", mtime=new_m) as data: + data.counter = 5 + await kanta.flush() + + assert kanta.mtime == new_m + await kanta.close() + + +@pytest.mark.asyncio +async def test_rollback_does_not_update_mtime(tmp_path, format_config): + path = tmp_path / "test.db" + kanta = make_kanta(path, Data, format_config) + await kanta.open() + + seed_m = datetime(2026, 1, 1, 10, 0, tzinfo=UTC) + with kanta.transaction(action="seed", mtime=seed_m) as data: + data.counter = 1 + + before = kanta.mtime + + try: + with kanta.transaction( + action="boom", mtime=datetime(2099, 1, 1, tzinfo=UTC) + ) as data: + data.counter = 99 + raise RuntimeError("fail") + except RuntimeError: + pass + + assert kanta.data.counter == 1 + assert kanta.mtime == before + await kanta.close()