From 5d66d423dadee80ae6cf195de3b85057eb99dc00 Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Fri, 7 Aug 2026 00:01:16 +0000 Subject: [PATCH] Lazy clock reads; demo: auto-advancing +1h clock, module-level setup --- demo/main.py | 122 ++++++++++++++++++------------------------- docs/database.md | 2 + kanta/kanta.py | 8 +-- kanta/kantaimpl.py | 2 +- kanta/persistence.py | 15 +++--- kanta/snapshot.py | 16 +++--- tests/test_clock.py | 21 ++++++++ 7 files changed, 98 insertions(+), 88 deletions(-) diff --git a/demo/main.py b/demo/main.py index ba3efe6..e826eba 100644 --- a/demo/main.py +++ b/demo/main.py @@ -25,18 +25,15 @@ DB = Path(__file__).with_name("demo.kantadb") # Fake directory: user id -> display name, resolved by the logfmt callbacks. USERS = {"u1": "Alice", "u2": "Bob", "u3": "Carol"} - -class Clock: - """Deterministic clock: manually advanced, so every run is identical.""" - - def __init__(self) -> None: - self.current = datetime(2026, 8, 6, 12, 0, tzinfo=UTC) - - def advance(self, **kwargs) -> None: - self.current += timedelta(**kwargs) +_now = datetime(2026, 8, 6, tzinfo=UTC) -clock = Clock() +def fake_now() -> datetime: + """Deterministic clock: starts at midnight, +1h on every read.""" + global _now + ts = _now + _now += timedelta(hours=1) + return ts class DataV1(msgspec.Struct): @@ -63,37 +60,43 @@ migrations = ModuleType("demo_migrations") migrations.migrate_v1 = migrate_v1 -def add_clock(kanta: Kanta) -> None: - """Use the shared deterministic clock for all record timestamps.""" - - @kanta.clock - def fake_now() -> datetime: - return clock.current +def resolve_actor(value: str) -> str | None: + """Resolve the transaction user id to a display name.""" + return USERS.get(value) -def add_logfmts(kanta: Kanta) -> None: - """Resolve user ids to display names in headers and diff paths.""" - - @kanta.logfmt(path="$user") - def resolve_actor(value: str) -> str | None: +def resolve_user_key(value: str, path: str) -> str | None: + """Resolve user ids in diff paths to display names.""" + if path.startswith("users."): return USERS.get(value) - - @kanta.logfmt - def resolve_user_key(value: str, path: str) -> str | None: - if path.startswith("users."): - return USERS.get(value) - return None + return None -def add_header(kanta: Kanta) -> None: +def header(action: str, user: str | None, extra: dict | None) -> str: """Aligned rich header: actor, session id, action, target.""" + actor = f"{_ACTOR}{user or '-':<8}{_RESET}" + session = f"{_SESSION}{extra.get('session_id', '-'):>2}{_RESET}" + target = f"{_TARGET}{extra['target']}{_RESET}" + return f"{actor} {session} {_ACTION}{action}{_RESET} {target}" - @kanta.logheader - def header(action: str, user: str | None, extra: dict | None) -> str: - actor = f"{_ACTOR}{user or '-':<8}{_RESET}" - session = f"{_SESSION}{extra.get('session_id', '-'):>2}{_RESET}" - target = f"{_TARGET}{extra['target']}{_RESET}" - return f"{actor} {session} {_ACTION}{action}{_RESET} {target}" + +def seed(data: DataV1) -> None: + """Bootstrap: create the initial admin user.""" + data.users["u1"] = {"name": "Alice", "role": "admin"} + + +# Phase 1 instance: default logging, original schema. +kanta_v0 = Kanta(DB, DataV1()) +# Phase 2 instance: migrations and a custom log header. +kanta_v1 = Kanta(DB, Data(), migrations=migrations) + +for k in (kanta_v0, kanta_v1): + k.clock(fake_now) + k.logfmt(resolve_user_key) + k.logfmt(resolve_actor, path="$user") + +kanta_v0.bootstrap(seed) +kanta_v1.logheader(header) def section(title: str) -> None: @@ -104,83 +107,62 @@ async def main() -> None: DB.unlink(missing_ok=True) section("Standard logging: bootstrap, diffs, toggles, rollback") + await kanta_v0.open() - kanta = Kanta(DB, DataV1()) - add_clock(kanta) - add_logfmts(kanta) - - @kanta.bootstrap - def seed(data: DataV1) -> None: - data.users["u1"] = {"name": "Alice", "role": "admin"} - - await kanta.open() - - clock.advance(minutes=2) - with kanta.transaction(action="create", user="u2") as data: + with kanta_v0.transaction(action="create", user="u2") as data: data.users["u2"] = {"name": "Bob", "role": "user"} - clock.advance(minutes=5) - with kanta.transaction(action="update", user="u1") as data: + with kanta_v0.transaction(action="update", user="u1") as data: data.users["u2"]["role"] = "editor" data.counter = 1 - clock.advance(seconds=30) - with kanta.transaction(action="delete", user="u1") as data: + with kanta_v0.transaction(action="delete", user="u1") as data: del data.users["u2"] # Display-only extra string, appended after the action. - clock.advance(hours=1) - with kanta.transaction(action="export", user="u1", extra=DB.name) as data: + with kanta_v0.transaction(action="export", user="u1", extra=DB.name) as data: data.counter = 2 # Compact logging, diff only: a system fix stamped by the clock, but the # modification time (m) is not updated. - clock.advance(minutes=10) - with kanta.transaction( + with kanta_v0.transaction( action="repair", mtime=False, log={"header": False, "diff": True} ) as data: data.users["u3"] = {"name": "Carol", "role": "user"} # A failing transaction rolls back and logs a warning. try: - with kanta.transaction(action="reset", user="u1") as data: + with kanta_v0.transaction(action="reset", user="u1") as data: data.counter = 99 raise ValueError("simulated failure") except ValueError: pass # Compact logging, header only. - clock.advance(minutes=5) - with kanta.transaction( + with kanta_v0.transaction( action="import", user="u1", log={"header": True, "diff": False} ) as data: data.counter = 3 - await kanta.close() + await kanta_v0.close() section("Reopen with migrations and a custom log header") - - clock.advance(days=1) - kanta = Kanta(DB, Data(), migrations=migrations) - add_clock(kanta) - add_logfmts(kanta) - add_header(kanta) - await kanta.open() + await kanta_v1.open() # No target given: defaults to the database filename. - clock.advance(minutes=3) - with kanta.transaction(action="update", user="u1", extra={"session_id": 3}) as data: + with kanta_v1.transaction( + action="update", user="u1", extra={"session_id": 3} + ) as data: data.settings["theme"] = "light" - clock.advance(minutes=1) - with kanta.transaction( + with kanta_v1.transaction( action="update", user="u2", extra={"session_id": 7, "target": "settings (demo)"}, ) as data: data.settings["lang"] = "en" - await kanta.close() + await kanta_v1.close() # The pretty names only exist in the logs; the database stores raw ids. section("Raw database records (user ids and timestamps, not pretty names)") diff --git a/docs/database.md b/docs/database.md index 475650e..e4d2aaa 100644 --- a/docs/database.md +++ b/docs/database.md @@ -168,6 +168,8 @@ when they have a default value. - `@kanta.clock` registers a callback `() -> datetime` that replaces the default UTC clock. Its value is used for all record timestamps (`ts`, and `m` when `mtime` is `True`) and for snapshot timestamps. +- The clock is only read when a timestamp is actually produced; no-op + transactions and skipped snapshot checks do not read it. - Register before `open()` so that bootstrap and migration records use the custom clock as well. This is mainly useful for tests and reproducible demos. diff --git a/kanta/kanta.py b/kanta/kanta.py index ea76bdf..73d152e 100644 --- a/kanta/kanta.py +++ b/kanta/kanta.py @@ -259,9 +259,11 @@ class Kanta(Generic[T]): Can be used as ``@kanta.clock``. The callback takes no arguments and must return a :class:`~datetime.datetime`; its value is used for all record timestamps (``ts``, and ``m`` when ``mtime`` is ``True``) and - snapshot timestamps. Register before :meth:`open` so that bootstrap - and migration records use the custom clock as well. This is mainly - useful for tests and reproducible demos. + snapshot timestamps. The clock is only read when a timestamp is + actually produced, so read-count-dependent clocks (e.g. advancing on + every read) stay deterministic. Register before :meth:`open` so that + bootstrap and migration records use the custom clock as well. This is + mainly useful for tests and reproducible demos. """ def _register(callback): diff --git a/kanta/kantaimpl.py b/kanta/kantaimpl.py index aedc4bb..93e5b75 100644 --- a/kanta/kantaimpl.py +++ b/kanta/kantaimpl.py @@ -300,7 +300,7 @@ class KantaImpl(PersistenceMixin, Generic[T]): self.version, self.statedict, m=self.mtime, - now=self.now(), + now=self.now, ) if migrations_ran and migration_result is not None: diff --git a/kanta/persistence.py b/kanta/persistence.py index 76a5fe8..81b0436 100644 --- a/kanta/persistence.py +++ b/kanta/persistence.py @@ -129,7 +129,7 @@ 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, m=self.mtime, now=self.now() + self.file, self.version, self.statedict, m=self.mtime, now=self.now ) def queue_change( @@ -158,6 +158,13 @@ class PersistenceMixin: The queued :class:`ChangeRecord`, or ``None`` if the diff was empty and *force* is ``False``. """ + diff = compute_diff(self.statedict, current) + if not diff: + if not force: + return None + diff = {} + + # The clock is only read when a record is actually queued. now = self.now() if mtime is True: @@ -169,12 +176,6 @@ class PersistenceMixin: else: raise TypeError("mtime must be True, False, or a datetime") - diff = compute_diff(self.statedict, current) - if not diff: - if not force: - return None - diff = {} - record = ChangeRecord( ts=now, a=action, diff --git a/kanta/snapshot.py b/kanta/snapshot.py index c6389fe..e97357b 100644 --- a/kanta/snapshot.py +++ b/kanta/snapshot.py @@ -3,6 +3,7 @@ from __future__ import annotations import logging +from collections.abc import Callable from datetime import UTC, datetime from kanta.structs import Snapshot @@ -43,23 +44,24 @@ class SnapshotState: version: int, state: dict, m: datetime | None = None, - now: datetime | None = None, + now: Callable[[], datetime] | None = None, ) -> None: """Write snapshot when thresholds/time policy allows it.""" force = self._force_pending - now = now if now is not None else datetime.now(UTC) + if not force and self.changes < self._min_diffs: + return + # The clock is only read when a snapshot may actually be written. + ts = now() if now is not None else datetime.now(UTC) if not force: - if self.changes < self._min_diffs: + if ts.weekday() != 6: # 6 = Sunday return - if now.weekday() != 6: # 6 = Sunday - return - sunday_midnight = now.replace(hour=0, minute=0, second=0, microsecond=0) + sunday_midnight = ts.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: - self._write(file, version, state, now, m=m) + self._write(file, version, state, ts, m=m) self._force_pending = False except Exception as exc: _logger.error("snapshot: failed to write snapshot: %r", exc) diff --git a/tests/test_clock.py b/tests/test_clock.py index 10cb4dc..4845996 100644 --- a/tests/test_clock.py +++ b/tests/test_clock.py @@ -72,6 +72,27 @@ async def test_clock_controls_record_timestamps(tmp_path, format_config): assert kanta.mtime == T0 + timedelta(hours=1) +@pytest.mark.asyncio +async def test_clock_not_read_without_changes(tmp_path, format_config): + kanta = make_kanta(tmp_path / "test.db", Data, format_config) + reads = 0 + + @kanta.clock + def fake_now() -> datetime: + nonlocal reads + reads += 1 + return T0 + + await kanta.open(log=False) # bootstrap record: one read + reads = 0 + + with kanta.transaction(action="noop"): + pass # no changes, no record, no clock read + await kanta.close() # no snapshot written, no clock read + + assert reads == 0 + + @pytest.mark.asyncio async def test_clock_controls_migration_and_snapshot_timestamps( tmp_path, format_config