Proper ts/mtime setting and tracking. Added Kanta.mtime property for reading current change time.

This commit is contained in:
Leo Vasanko
2026-06-13 00:49:35 +00:00
parent 97209b8556
commit 74150b783c
8 changed files with 247 additions and 30 deletions
+16 -1
View File
@@ -82,15 +82,30 @@ history.
## Transaction Semantics ## Transaction Semantics
- `kanta.transaction(action=...)` captures a pre-transaction snapshot dict. - `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: - On success:
- compute diff between previous builtins and current builtins, - 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: - On exception:
- restore in-memory data from snapshot, - restore in-memory data from snapshot,
- re-raise the exception. - re-raise the exception.
Nested transactions are rejected. 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 ## Flush and Lifecycle
- Writes are queued in memory. - Writes are queued in memory.
+25 -1
View File
@@ -3,6 +3,7 @@
from __future__ import annotations from __future__ import annotations
from collections.abc import Callable from collections.abc import Callable
from datetime import datetime
from pathlib import Path from pathlib import Path
from types import ModuleType from types import ModuleType
from typing import Any, Generic, TypeVar from typing import Any, Generic, TypeVar
@@ -132,6 +133,17 @@ class Kanta(Generic[T]):
""" """
return self._impl.filename 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: async def open(self) -> None:
"""Open the database file and start background persistence. """Open the database file and start background persistence.
@@ -184,6 +196,7 @@ class Kanta(Generic[T]):
user: str | None = None, user: str | None = None,
user_display: str | None = None, user_display: str | None = None,
resolver: Any = None, resolver: Any = None,
mtime: bool | datetime = True,
): ):
"""Create a transactional mutation context manager. """Create a transactional mutation context manager.
@@ -192,6 +205,12 @@ class Kanta(Generic[T]):
user: Optional user identifier stored in metadata. user: Optional user identifier stored in metadata.
user_display: Optional display name used for logging/resolution. user_display: Optional display name used for logging/resolution.
resolver: Optional callable for resolving identifiers in logs. 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: Returns:
A context manager yielding the live state object for mutation. A context manager yielding the live state object for mutation.
@@ -202,5 +221,10 @@ class Kanta(Generic[T]):
rolled back. rolled back.
""" """
return _transaction( 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,
) )
+2 -1
View File
@@ -104,8 +104,9 @@ class KantaImpl(PersistenceMixin, Generic[T]):
serializer=self.serializer, serializer=self.serializer,
) )
self.version = rr.version self.version = rr.version
self.mtime = rr.m
normalized = struct_to_dict(self.data, serializer=self.serializer) 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 = ( self.snapshot.ts = (
datetime.fromtimestamp(rr.last_snapshot_mtime, UTC) datetime.fromtimestamp(rr.last_snapshot_mtime, UTC)
if rr.last_snapshot_mtime is not None if rr.last_snapshot_mtime is not None
+47 -15
View File
@@ -8,7 +8,7 @@ import logging
import threading import threading
from collections import deque from collections import deque
from collections.abc import Callable from collections.abc import Callable
from datetime import datetime from datetime import UTC, datetime
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
@@ -41,6 +41,7 @@ class PersistenceMixin:
flush_interval: float flush_interval: float
version: int version: int
opened: bool opened: bool
mtime: datetime | None
def __init__(self, **kwargs: Any) -> None: def __init__(self, **kwargs: Any) -> None:
"""Initialize persistence-owned state used by mixin methods.""" """Initialize persistence-owned state used by mixin methods."""
@@ -63,6 +64,7 @@ class PersistenceMixin:
self.background_error = None self.background_error = None
self.flush_interval = flush_interval self.flush_interval = flush_interval
self.version = 0 self.version = 0
self.mtime: datetime | None = None
async def _background_loop(self) -> None: async def _background_loop(self) -> None:
"""Background task that periodically flushes changes to disk.""" """Background task that periodically flushes changes to disk."""
@@ -89,30 +91,60 @@ class PersistenceMixin:
def maybe_snapshot(self) -> None: def maybe_snapshot(self) -> None:
"""Evaluate and possibly write a snapshot from current state.""" """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( def queue_change(
self, self,
action: str, action: str,
current: dict, current: dict,
*,
user: str | None = None, user: str | None = None,
m: datetime | None = None, mtime: bool | datetime = True,
) -> None: ) -> ChangeRecord | None:
"""Queue a change record internally (thread-safe).""" """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) diff = compute_diff(self.statedict, current)
if not diff: 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: with self.pending_lock:
self.pending_changes.append( self.pending_changes.append(record)
ChangeRecord(
a=action,
v=self.version,
u=user,
m=m,
diff=diff,
)
)
self.statedict = copy.deepcopy(current) self.statedict = copy.deepcopy(current)
if m is not None:
self.mtime = m
return record
def flush_sync(self) -> None: def flush_sync(self) -> None:
"""Synchronously flush all pending changes to disk.""" """Synchronously flush all pending changes to disk."""
-5
View File
@@ -23,14 +23,12 @@ class ReplayResult:
state: dict[str, Any], state: dict[str, Any],
version: int = 0, version: int = 0,
has_migration: bool = False, has_migration: bool = False,
last_patch_mtime: float | None = None,
last_snapshot_mtime: float | None = None, last_snapshot_mtime: float | None = None,
m: datetime | None = None, m: datetime | None = None,
): ):
self.state = state self.state = state
self.version = version self.version = version
self.has_migration = has_migration self.has_migration = has_migration
self.last_patch_mtime = last_patch_mtime
self.last_snapshot_mtime = last_snapshot_mtime self.last_snapshot_mtime = last_snapshot_mtime
self.m = m self.m = m
@@ -63,7 +61,6 @@ def replay(
last_snapshot_mtime: float | None = None last_snapshot_mtime: float | None = None
m: datetime | None = None m: datetime | None = None
has_migration = False has_migration = False
last_patch_mtime: float | None = None
if snap_payload is not None: if snap_payload is not None:
try: try:
@@ -112,14 +109,12 @@ def replay(
has_migration = True has_migration = True
if change.m is not None: if change.m is not None:
m = change.m m = change.m
last_patch_mtime = change.ts.timestamp()
version = change.v version = change.v
state = _patch_state(state, change.diff) state = _patch_state(state, change.diff)
return ReplayResult( return ReplayResult(
state=state, state=state,
version=version, version=version,
has_migration=has_migration, has_migration=has_migration,
last_patch_mtime=last_patch_mtime,
last_snapshot_mtime=last_snapshot_mtime, last_snapshot_mtime=last_snapshot_mtime,
m=m, m=m,
) )
+8 -4
View File
@@ -37,7 +37,9 @@ class SnapshotState:
def record_changes(self, count: int) -> None: def record_changes(self, count: int) -> None:
self.changes += count 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.""" """Write snapshot when thresholds/time policy allows it."""
if self.changes < self._min_diffs: if self.changes < self._min_diffs:
return return
@@ -51,14 +53,16 @@ class SnapshotState:
if not file.is_open: if not file.is_open:
return return
try: try:
self._write(file, version, state, now) self._write(file, version, state, now, m=m)
self._force_pending = False self._force_pending = False
except Exception as exc: except Exception as exc:
_logger.error("snapshot: failed to write snapshot: %r", 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.""" """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 record_offset = file.size() if hasattr(file, "size") else 0
file.write(self._framer.frame_snapshot(payload, record_offset=record_offset)) file.write(self._framer.frame_snapshot(payload, record_offset=record_offset))
self.changes = 0 self.changes = 0
+6 -3
View File
@@ -4,6 +4,7 @@ from __future__ import annotations
import logging import logging
from contextlib import contextmanager from contextlib import contextmanager
from datetime import datetime
from typing import Any from typing import Any
from kanta.diff import compute_diff from kanta.diff import compute_diff
@@ -22,6 +23,7 @@ def transaction(
user: str | None = None, user: str | None = None,
user_display: str | None = None, user_display: str | None = None,
resolver: Any = None, resolver: Any = None,
mtime: bool | datetime = True,
): ):
"""Wrap writes in a transaction and yield the live db object.""" """Wrap writes in a transaction and yield the live db object."""
if impl.in_transaction: if impl.in_transaction:
@@ -58,9 +60,10 @@ def transaction(
new_dict = struct_to_dict(impl.data, serializer=impl.serializer) new_dict = struct_to_dict(impl.data, serializer=impl.serializer)
diff = compute_diff(impl.statedict, new_dict) diff = compute_diff(impl.statedict, new_dict)
if diff: if diff:
impl.queue_change(action, new_dict, user=user) previous = impl.statedict
log_change(action, diff, user_display, impl.statedict, resolver) record = impl.queue_change(action, new_dict, user=user, mtime=mtime)
impl.statedict = new_dict if record is not None:
log_change(action, record.diff, user_display, previous, resolver)
except Exception: except Exception:
_logger.warning("Transaction '%s' failed, rolling back changes", action) _logger.warning("Transaction '%s' failed, rolling back changes", action)
if impl.transaction_snapshot is not None: if impl.transaction_snapshot is not None:
+143
View File
@@ -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()