Add revision numbers: snapshot r field, kanta.rev, CLI rN selectors, k1 meta migration

Revisions are absolute state positions between diffs: the empty initial
state is revision 0 and each change record advances it by one.  Snapshots
persist the revision of the state they contain in a new r field, stamped
with a meta format version k (kanta.structs.META_FORMAT, currently 1);
change records carry no revision of their own.

- kanta.rev exposes the current revision (raises before open)
- replay restores the revision, trusting r on k-marked snapshots and
  counting change records from the start of the file for legacy (k0) ones
- rotation preserves absolute revisions, counted from record positions
- meta migration k1 rewrites a legacy file in full on writable open:
  leading snapshot preserved and re-stamped, all change records kept,
  closing snapshot with the total revision; read-only opens compute the
  same revision without touching the file
- CLI range selectors accept rN bounds: rN is the record that produced
  state N with its state (the change, or the snapshot materializing it);
  single indexes must exist, ranges truncate Python-style
- CLI snapshot indicator lines show the snapshot revision
This commit is contained in:
LeoVasanko committed 2026-09-28 00:25:52 +00:00
1 parent ec8695ba0c
commit de2476c59c
14 files changed
+793 -28

No files matched your search

+29 -4
View File
@@ -51,7 +51,7 @@ Fields:
Snapshot lines are prefixed with `SNAPSHOT `, followed by JSON:
```text
SNAPSHOT {"ts":"2026-06-10T00:00:00Z","v":5,"state":{"users":{}},"m":"2026-06-10T00:00:00Z"}
SNAPSHOT {"ts":"2026-06-10T00:00:00Z","v":5,"state":{"users":{}},"m":"2026-06-10T00:00:00Z","r":12,"k":"0.9.2"}
```
Fields:
@@ -59,13 +59,38 @@ Fields:
- `v`: schema version represented by the snapshot.
- `state`: full state dictionary.
- `m`: optional domain modification timestamp.
- `r`: revision of the snapshotted state — the total number of change records that occurred before this snapshot.
- `k`: meta format version (integer), the module-level constant `kanta.structs.META_FORMAT` at write time. It is bumped manually when the meta format changes. `0` (a missing field in old files) marks a legacy snapshot whose `r` cannot be trusted; readers then count change records from the start of the file instead. k1 introduced revision numbers.
## Revision Numbers
A revision number identifies a *state position between diffs*, not a diff:
- The empty initial state is revision 0.
- Each change record — including the bootstrap record — advances the position by one, so after the first record the state is at revision 1.
- Change records carry no revision of their own; the revision of a state is counted from the preceding snapshot (`snapshot.r + n` records after it), or from the start of the file.
- Revisions are absolute and survive rotation: the leading snapshot of a rotated file carries the revision of its cutoff state, so `r` values keep referring to the same state positions across history rewrites.
`kanta.rev` exposes the current revision at runtime: the number of change records applied since revision 0, advanced in memory as transactions queue records. Reading it before `open()` raises `DataIntegrityError`. The CLI accepts `rN` range bounds to select records by revision (e.g. `-r r10:r20`): a bound `rN` addresses state N together with the record that produced it — the change record resulting in state N, or, when that change is no longer in the file, the snapshot materializing state N. `-r r5` and `-r r5:r6` therefore show the same record and output state 5, while `-r r5..r6` also includes the next change and outputs state 6. Ranges truncate Python-style instead of erroring: on a rotated file whose leading snapshot holds revision 3, `-r r1:r4` starts at the beginning of the file and shows that snapshot. Snapshot indicator lines show the snapshot's revision.
## Meta Format Migrations
The snapshot `k` field carries an integer meta format version (`kanta.structs.META_FORMAT`, currently 1), bumped manually whenever the record-level format around the payloads changes. k1 introduced revision numbers.
When a writable `open()` finds a legacy (k0) last snapshot, the file is rewritten in full under the lock, before replay results are used further (meta migration k1, `kanta/metamigr.py`):
1. A leading snapshot, when the file begins with one, is preserved with its data and timestamps, re-stamped k1 with `r = 0` (the only count knowable from the file).
2. All change records follow in order, re-framed; redundant intermediate snapshots are dropped.
3. A closing k1 snapshot captures the final state with `r` equal to the total number of change records.
After the rewrite every snapshot in the file is k1 with a trustworthy `r`, so readers never deal with mixed r/k meanings in one file. Rotation composes with this: it runs first and counts revisions from record positions regardless of `k`, so rotating a legacy file already produces k1 snapshots and no separate migration is needed. A read-only open computes the same revision the migration would have produced but leaves the file untouched.
## Replay Model
1. Find the last snapshot in the file, if present.
2. Initialize replay state from snapshot state (or `{}` if none).
3. Replay subsequent change records in order using patch application.
4. The final replay state becomes in-memory `kanta.data`.
2. Initialize replay state from snapshot state (or `{}` if none) and the base revision from the snapshot's `r` (or a count of the preceding change records when the snapshot is a legacy one without `k`).
3. Replay subsequent change records in order using patch application, counting them toward the revision.
4. The final replay state becomes in-memory `kanta.data`, and the final count becomes `kanta.rev`.
This model provides fast startup for large logs while retaining append-only history.
+3 -3
View File
@@ -43,13 +43,13 @@ Let `cutoff = now - retention`. All planning happens on the in-memory bytes of t
4. **Copy the original aside.** The main file is copied with `shutil.copy2` directly to its final `{stem}@{ts}.kantadb` name. On filesystems with copy-on-write this performs a cheap reflink copy. A failure here aborts rotation with the original file intact.
5. **Rewrite the main file in place.** On the locked file descriptor the content is replaced (seek 0, truncate, write, fsync) with, in order:
1. A **snapshot of the state at the cutoff**, stamped with the schema version and modification time in effect at the cutoff. Its `ts` is the `ts` of the last pre-cutoff record — the same 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.
1. A **snapshot of the state at the cutoff**, stamped with the schema version and modification time in effect at the cutoff. Its `ts` is the `ts` of the last pre-cutoff record — the same timestamp used in the rotated filename. Its `r` is the absolute revision of the cutoff state (the number of change records before the cutoff position in the original file), so revision numbers keep referring to the same state positions after the rotation. 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.
3. A **final snapshot** of the state after the last retained record, written only when enough changes were retained to warrant one (the same policy as regular snapshot writes). If no records survived the cutoff, the new file consists of the single leading snapshot and nothing else — the steady state for databases whose history has fully aged out.
3. A **final snapshot** of the state after the last retained record, written only when enough changes were retained to warrant one (the same policy as regular snapshot writes); its `r` is the total number of change records in the original file. If no records survived the cutoff, the new file consists of the single leading snapshot and nothing else — the steady state for databases whose history has fully aged out.
6. **Trim the rotated copy.** The rotated file is truncated to `cutoff_end` bytes, so it contains only the dropped history and does not duplicate the records retained in the main file. The cut is at a frame boundary, so the rotated file remains a valid, replayable database on its own. This happens only after step 5's fsync, so until then the rotated file still holds the complete original content as a crash-recovery anchor.
7. **Continue normal open.** Replay and migrations proceed on the same locked file. Because the leading snapshot carries the current version, migrations run exactly as they would have against the old content.
7. **Continue normal open.** Replay and migrations proceed on the same locked file. Because the leading snapshot carries the current version, migrations run exactly as they would have against the old content. The rotation's snapshots are stamped with the current meta format version (`k`), so rotating a legacy (pre-revision) file also performs the k1 meta migration — revision numbers are counted from record positions before the rotation (see [Database format](database.md#meta-format-migrations)).
Failure rule: any error before step 5 leaves the main file byte-identical (only an extra copy exists). A crash during step 5 may leave the main file torn, but the rotated copy still holds the complete original content — recovery is copying it back. After step 6 the split is complete and both files are consistent.
+13 -5
View File
@@ -237,10 +237,13 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
"--range",
help=(
"Python-style range to process. Units: plain number = change index,"
" lN = line number, sN = snapshot, vN = version. Negative snapshot"
" values count from the end (s-1 is the last snapshot). Use ':' for"
" lN = line number, sN = snapshot, vN = version, rN = revision"
" (rN is the record that produced state N, with its state)."
" Negative snapshot values count from the end (s-1 is the last"
" snapshot). Use ':' for"
" half-open ranges and '..' for inclusive end ranges. Examples:"
" '2:5', '2..5', 'l10:l20', 's1:s3', 'v0:v2', 's-1:', ':-1', '-1'."
" '2:5', '2..5', 'l10:l20', 's1:s3', 'v0:v2', 'r10:r20', 's-1:',"
" ':-1', '-1'."
),
)
parser.add_argument(
@@ -373,16 +376,19 @@ def _print_snapshot_indicator(
label: str,
snap: Snapshot,
index: int,
revision: int,
serializer: Serializer,
) -> None:
"""Print a snapshot indicator line to stderr.
``snapshot s<N>`` is rendered in bright white; the version, optional mtime
and data size are printed in normal and dark colors respectively.
``snapshot s<N>`` is rendered in bright white; the version, revision,
optional mtime and data size are printed in normal and dark colors
respectively.
"""
ts = _format_ts(snap.ts)
line = Line().snapshot("snapshot").snapshot(f" s{index}")
line.target(f" v{snap.v}")
line.target(f" r{revision}")
if snap.m is not None:
line.target(f" {_format_ts(snap.m)}")
size = len(serializer.encode(snap.state))
@@ -523,6 +529,7 @@ async def _run(args: argparse.Namespace) -> int:
record_label(snap_event.line_number, snap_event.record_index),
snap_event.snap,
snapshot_line_to_index[snap_event.line_number],
snap_event.revision,
kanta._impl.serializer,
)
_print()
@@ -552,6 +559,7 @@ async def _run(args: argparse.Namespace) -> int:
label,
event.snap,
snapshot_line_to_index[event.line_number],
event.revision,
kanta._impl.serializer,
)
else:
+25
View File
@@ -7,6 +7,7 @@ from pathlib import Path
from types import ModuleType, SimpleNamespace
from typing import Any, Generic, TypeVar
from kanta.exceptions import DataIntegrityError
from kanta.kantaimpl import KantaImpl
from kanta.serialization import JsonSerializer, Serializer
from kanta.transaction import transaction as _transaction
@@ -146,6 +147,30 @@ class Kanta(Generic[T]):
"""
return self._impl.ctx
@property
def rev(self) -> int:
"""Current revision: the state position as a count of change records.
Revision numbers identify state positions *between* diffs, not the
diffs themselves: the empty initial state is revision 0, and each
change record (including the bootstrap record) advances the position
by one. Snapshots persist the revision of the state they contain in
their ``r`` field; change records carry no revision of their own.
Returns:
The number of change records applied since revision 0.
Raises:
kanta.exceptions.DataIntegrityError: If the instance is not open.
"""
if not self._impl.opened:
raise DataIntegrityError(
"Kanta instance must be opened before reading rev",
db_path=self._impl.filename,
action="rev",
)
return self._impl.rev
@property
def mtime(self) -> datetime | None:
"""Last modification time carried forward from change records.
+30 -3
View File
@@ -19,6 +19,7 @@ from kanta.logging import (
emit_event,
migration_logger,
)
from kanta.metamigr import migrate_k1
from kanta.migrations import MigrationReport, Migrations
from kanta.persistence import PersistenceMixin
from kanta.rotation import execute_rotation, plan_rotation
@@ -217,6 +218,26 @@ class KantaImpl(PersistenceMixin, Generic[T]):
cause_type=type(e).__name__,
) from e
# Meta migration k0 → k1: a file whose last snapshot is legacy
# is rewritten in full so every snapshot gains a trustworthy r
# (see kanta/metamigr.py). Rotation, when it ran, already
# produced k1 snapshots with positionally counted revisions, so
# this only fires when no rotation happened. Read-only opens
# skip the rewrite; replay counted the same revision either way.
meta_migrated_at: datetime | None = None
if rr.legacy_snapshot and not self.readonly:
meta_migrated_at = self.now()
new_content = migrate_k1(
content,
framer=self.framer,
serializer=self.serializer,
state=rr.state,
version=rr.version,
m=rr.m,
now=meta_migrated_at,
)
await asyncio.to_thread(self.file.replace_content, new_content)
migration_report = None
state_before_migrations = None
previous_version = rr.version
@@ -230,9 +251,13 @@ class KantaImpl(PersistenceMixin, Generic[T]):
migrations_ran = rr.version != previous_version
self.snapshot.ts = (
datetime.fromtimestamp(rr.last_snapshot_mtime, UTC)
if rr.last_snapshot_mtime is not None
else None
meta_migrated_at
if meta_migrated_at is not None
else (
datetime.fromtimestamp(rr.last_snapshot_mtime, UTC)
if rr.last_snapshot_mtime is not None
else None
)
)
self.statedict = copy.deepcopy(
@@ -258,6 +283,7 @@ class KantaImpl(PersistenceMixin, Generic[T]):
raise
self.version = rr.version
self.mtime = rr.m
self.rev = rr.rev
if log is not False and not migrations_ran:
logger = log if isinstance(log, logging.Logger) else bootstrap_logger
emit_event(
@@ -314,6 +340,7 @@ class KantaImpl(PersistenceMixin, Generic[T]):
self.statedict,
m=self.mtime,
now=self.now,
r=self.rev,
)
elif self.readonly:
self.opened = False
+75
View File
@@ -0,0 +1,75 @@
"""Meta format migrations: rewriting files between snapshot ``k`` versions.
The k0 → k1 migration introduces revision numbers. Like a rotation, the
whole file is rewritten under the lock so that every snapshot in it is a
k1 snapshot with a trustworthy ``r`` — mixed r/k meanings never persist in
one file. Unlike a rotation, all change records are kept; redundant
intermediate snapshots are dropped.
"""
from __future__ import annotations
from datetime import datetime
from typing import Any
from kanta.structs import META_FORMAT, ChangeRecord, Snapshot
from kanta.serialization.base import Serializer
from kanta.serialization.framing import Framer
def migrate_k1(
content: bytes,
*,
framer: Framer,
serializer: Serializer,
state: dict[str, Any],
version: int,
m: datetime | None,
now: datetime,
) -> bytes:
"""Rewrite *content* wholly into k1 form.
The leading snapshot, when the file begins with one (the file start is
not the database start), is preserved with its data and timestamps,
re-stamped k1; its ``r`` is 0, the only count knowable from the file.
All change records follow, re-framed in order, and a closing snapshot
captures the replayed final state with ``r`` equal to the total number
of change records. *state*, *version* and *m* are the replayed final
values (pre-schema-migration, so migrations still run afterwards).
The caller gates on a legacy last snapshot and performs the in-place
rewrite under the file lock.
"""
entries = []
for is_snapshot, payload, _line, _pos in framer.iter_records(content, 0):
entries.append(
(
is_snapshot,
serializer.decode(
payload, type=Snapshot if is_snapshot else ChangeRecord
),
)
)
out = bytearray()
if entries and entries[0][0]:
snap = entries[0][1]
assert isinstance(snap, Snapshot)
leading = Snapshot(
ts=snap.ts, v=snap.v, state=snap.state, m=snap.m, r=0, k=META_FORMAT
)
out += framer.frame_snapshot(serializer.encode(leading), record_offset=0)
n_changes = 0
for is_snapshot, record in entries:
if is_snapshot:
continue
out += framer.frame_change(serializer.encode(record), record_offset=len(out))
n_changes += 1
if n_changes:
closing = Snapshot(
ts=now, v=version, state=state, m=m, r=n_changes, k=META_FORMAT
)
out += framer.frame_snapshot(serializer.encode(closing), record_offset=len(out))
return bytes(out)
+9 -1
View File
@@ -40,6 +40,7 @@ class PersistenceMixin:
background_error: DatabaseError | None
flush_interval: float
version: int
rev: int
opened: bool
readonly: bool
mtime: datetime | None
@@ -64,6 +65,7 @@ class PersistenceMixin:
self.background_error = None
self.flush_interval = flush_interval
self.version = 0
self.rev = 0
self.mtime: datetime | None = None
self.clock: Callable[[], datetime] | None = None
@@ -123,7 +125,12 @@ 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,
r=self.rev,
)
def queue_change(
@@ -180,6 +187,7 @@ class PersistenceMixin:
)
self.pending_changes.append(record)
self.statedict = copy.deepcopy(current)
self.rev += 1
if m is not None:
self.mtime = m
return record
+58 -3
View File
@@ -33,6 +33,7 @@ class SnapshotEvent:
byte_pos: int
record_index: int
snap: Snapshot
revision: int = 0
@property
def version(self) -> int:
@@ -47,6 +48,7 @@ class ChangeEvent:
byte_pos: int
record_index: int
record: ChangeRecord
revision: int = 0
@property
def version(self) -> int:
@@ -96,6 +98,7 @@ def scan_events(content: bytes, kanta: Kanta[Any]) -> tuple[list[Event], int]:
state: dict[str, Any] = {}
events: list[Event] = []
change_count = 0
revision = 0
for is_snapshot, payload, line_number, byte_pos in impl.framer.iter_records(
content, 0
):
@@ -113,12 +116,22 @@ def scan_events(content: bytes, kanta: Kanta[Any]) -> tuple[list[Event], int]:
record_type="snapshot",
)
state = snap.state
events.append(SnapshotEvent(line_number, byte_pos, record_index, snap))
# A k-marked snapshot carries an absolute revision, which may
# start past zero (rotated files). Legacy snapshots leave the
# file-relative count in effect, which is exact for them.
if snap.k:
revision = snap.r
events.append(
SnapshotEvent(line_number, byte_pos, record_index, snap, revision)
)
else:
record = impl.serializer.decode(payload, type=ChangeRecord)
state = patch(state, record.diff)
events.append(ChangeEvent(line_number, byte_pos, record_index, record))
change_count += 1
revision += 1
events.append(
ChangeEvent(line_number, byte_pos, record_index, record, revision)
)
except msgspec.DecodeError as exc:
raise ReplayError(
f"Parse error at {label}: {exc}",
@@ -224,6 +237,21 @@ def _version_lines(events: list[Event]) -> dict[int, int]:
return lines
def _revision_lines(events: list[Event]) -> dict[int, int]:
"""Map each selectable revision to the line of the record that produced it.
A revision bound ``rN`` addresses state N together with the record that
produced it: the change record whose resulting revision is N, or — when
that change is no longer in the file — the snapshot that materializes
state N. A change takes precedence over a redundant snapshot of the
same revision (file order).
"""
lines: dict[int, int] = {}
for event in events:
lines.setdefault(event.revision, event.line_number)
return lines
def _event_at_line(events: list[Event], line: int) -> Event | None:
"""Return the event whose file line number exactly matches ``line``."""
for event in events:
@@ -236,7 +264,7 @@ def _parse_bound(bound_str: str) -> tuple[str, int | None]:
"""Parse a range bound with optional unit prefix (l, s, v) or change index."""
if not bound_str:
return "change", None
unit_map = {"l": "line", "s": "snapshot", "v": "version"}
unit_map = {"l": "line", "s": "snapshot", "v": "version", "r": "revision"}
if bound_str[0] in unit_map:
unit = unit_map[bound_str[0]]
rest = bound_str[1:]
@@ -287,6 +315,16 @@ def _bound_to_line(
raise ValueError("version numbers do not support negative indexing")
return _version_lines(events).get(value, eof)
if unit == "revision":
if value < 0:
raise ValueError("revision numbers do not support negative indexing")
lines = _revision_lines(events)
if value in lines:
return lines[value]
if not lines or value < min(lines):
return 0
return eof
raise ValueError(f"unknown range unit: {unit}")
@@ -368,6 +406,23 @@ def select(spec: str, events: list[Event], total: int) -> Selection:
later = [line for line in lines.values() if line > start_line]
return Selection(start_line, min(later) if later else end_of_file(events))
if unit == "revision":
_negative_check(unit, value)
lines = _revision_lines(events)
if value not in lines:
raise RangeNotFoundError(
f"Revision {spec!r} not found in file"
f" ({_plural(len(lines), 'revision')})"
)
start_line = lines[value]
event = _event_at_line(events, start_line)
if isinstance(event, SnapshotEvent):
# The producing change is no longer in the file; the snapshot is
# the record materializing this state.
return Selection(start_line, start_line + 1, event)
later = [line for line in lines.values() if line > start_line]
return Selection(start_line, min(later) if later else end_of_file(events))
lines = _change_lines(events)
idx = total + value if value < 0 else value
if not 0 <= idx < total:
+18 -3
View File
@@ -16,7 +16,7 @@ from pathlib import Path
from typing import Any
from kanta.exceptions import DatabaseError
from kanta.structs import ChangeRecord, Snapshot
from kanta.structs import META_FORMAT, ChangeRecord, Snapshot
from kanta.serialization.base import Serializer, apply_diff
from kanta.serialization.framing import Framer
@@ -174,10 +174,18 @@ def plan_rotation(
# Build the new content: leading cutoff snapshot, retained changes
# re-framed at fresh offsets, and a final snapshot when enough changes
# survived to warrant one (mirrors the regular snapshot policy).
# Revisions are absolute positions counted from the start of the
# original file, so they survive the rotation unchanged.
r_at_cutoff = sum(1 for e in changes if e.byte_pos < cutoff_end)
out = bytearray()
leading = serializer.encode(
Snapshot(
ts=rotated_ts, v=version_at_cutoff, state=state_at_cutoff, m=m_at_cutoff
ts=rotated_ts,
v=version_at_cutoff,
state=state_at_cutoff,
m=m_at_cutoff,
r=r_at_cutoff,
k=META_FORMAT,
)
)
out += framer.frame_snapshot(leading, record_offset=0)
@@ -186,7 +194,14 @@ def plan_rotation(
out += framer.frame_change(payload, record_offset=len(out))
if len(retained) >= min_diffs:
closing = serializer.encode(
Snapshot(ts=now, v=final_version, state=state, m=final_m)
Snapshot(
ts=now,
v=final_version,
state=state,
m=final_m,
r=len(changes),
k=META_FORMAT,
)
)
out += framer.frame_snapshot(closing, record_offset=len(out))
+23
View File
@@ -25,12 +25,16 @@ class ReplayResult:
has_migration: bool = False,
last_snapshot_mtime: float | None = None,
m: datetime | None = None,
rev: int = 0,
legacy_snapshot: bool = False,
):
self.state = state
self.version = version
self.has_migration = has_migration
self.last_snapshot_mtime = last_snapshot_mtime
self.m = m
self.rev = rev
self.legacy_snapshot = legacy_snapshot
class Serializer(Protocol):
@@ -61,6 +65,8 @@ def replay(
last_snapshot_mtime: float | None = None
m: datetime | None = None
has_migration = False
r_base = 0
legacy_snapshot = False
if snap_payload is not None:
try:
@@ -76,7 +82,21 @@ def replay(
version = snap.v
last_snapshot_mtime = snap.ts.timestamp()
m = snap.m
if snap.k == 0:
# Legacy snapshot: r cannot be trusted, so count change records
# from the start of the file (framing only, no payload decode).
legacy_snapshot = True
r_base = sum(
1
for is_snapshot, _payload, _line, byte_pos in framer.iter_records(
data, 0
)
if not is_snapshot and byte_pos < snap_byte_pos
)
else:
r_base = snap.r
changes = 0
for is_snapshot, payload, line_number, byte_pos in framer.iter_records(
data, offset
):
@@ -111,12 +131,15 @@ def replay(
m = change.m
version = change.v
state = _patch_state(state, change.diff)
changes += 1
return ReplayResult(
state=state,
version=version,
has_migration=has_migration,
last_snapshot_mtime=last_snapshot_mtime,
m=m,
rev=r_base + changes,
legacy_snapshot=legacy_snapshot,
)
+13 -4
View File
@@ -6,7 +6,7 @@ import logging
from collections.abc import Callable
from datetime import UTC, datetime
from kanta.structs import Snapshot
from kanta.structs import META_FORMAT, Snapshot
from kanta.serialization import JsonSerializer, Serializer
from kanta.serialization.framing import Framer, LineFramer
@@ -50,6 +50,7 @@ class SnapshotState:
state: dict,
m: datetime | None = None,
now: Callable[[], datetime] | None = None,
r: int = 0,
) -> None:
"""Write snapshot when thresholds/time policy allows it."""
force = self._force_pending
@@ -66,16 +67,24 @@ class SnapshotState:
if not file.is_open:
return
try:
self._write(file, version, state, ts, m=m)
self._write(file, version, state, ts, m=m, r=r)
self._force_pending = False
except Exception as exc:
_logger.error("Kanta snapshot failed: %r", exc)
def _write(
self, file, version: int, state: dict, now: datetime, m: datetime | None = None
self,
file,
version: int,
state: dict,
now: datetime,
m: datetime | None = None,
r: int = 0,
) -> None:
"""Write a snapshot and update internal state."""
payload = self._serializer.encode(Snapshot(ts=now, v=version, state=state, m=m))
payload = self._serializer.encode(
Snapshot(ts=now, v=version, state=state, m=m, r=r, k=META_FORMAT)
)
record_offset = file.size() if hasattr(file, "size") else 0
file.write(self._framer.frame_snapshot(payload, record_offset=record_offset))
self.changes = 0
+20 -1
View File
@@ -5,6 +5,11 @@ from typing import Any
import msgspec
# On-disk meta format version, stored in the snapshot ``k`` field and bumped
# manually when the meta format changes. k0 (missing field) predates
# revision numbers; k1 introduces them (see docs/database.md).
META_FORMAT = 1
class ChangeRecord(msgspec.Struct, omit_defaults=True, kw_only=True):
"""A single change record in the JSONL log.
@@ -27,9 +32,23 @@ class ChangeRecord(msgspec.Struct, omit_defaults=True, kw_only=True):
class Snapshot(msgspec.Struct, omit_defaults=True):
"""Full state snapshot embedded in the JSONL log."""
"""Full state snapshot embedded in the JSONL log.
Attributes:
ts: Snapshot creation time.
v: Schema version represented by the snapshot.
state: Full state dictionary.
m: Optional domain modification timestamp.
r: Revision of the snapshotted state: the total number of change
records that occurred before this snapshot.
k: Meta format version (``META_FORMAT`` when written). 0 marks a
legacy snapshot whose ``r`` cannot be trusted; readers count
change records from the start of the file instead.
"""
ts: datetime
v: int
state: dict[str, Any]
m: datetime | None = None
r: int = 0
k: int = 0
+1 -1
View File
@@ -105,7 +105,7 @@ def test_cli_snapshot_line_format(tmp_path, capsys, monkeypatch):
# Snapshot line: bright white snapshot/sN, white version/mtime, dark size.
assert "\x1b[97msnapshot s0" in err
assert "\x1b[38;5;250m v1 2026-08-12 09:00:00" in err
assert "\x1b[38;5;250m v1 r0 2026-08-12 09:00:00" in err
assert "\x1b[38;5;242m 13 B" in err
+476
View File
@@ -0,0 +1,476 @@
"""Tests for revision numbers: runtime counter, snapshot ``r`` field, CLI rN."""
import asyncio
import json
from datetime import UTC, datetime, timedelta
import pytest
from kanta.exceptions import DataIntegrityError
from kanta.serialization import JsonSerializer
from kanta.structs import META_FORMAT, ChangeRecord, Snapshot
from tests.support import Data, make_kanta, read_last_snapshot
from tests.test_rotation import make_clock, read_all, write_history
T0 = datetime(2026, 1, 1, tzinfo=UTC)
# The CLI reads files with its own JSON Kanta instance unless -k is given.
JSON = ("json", JsonSerializer)
@pytest.mark.asyncio
async def test_revision_requires_open(tmp_path, format_config):
kanta = make_kanta(tmp_path / "data.kantadb", Data, format_config)
with pytest.raises(DataIntegrityError):
kanta.rev
@pytest.mark.asyncio
async def test_revision_counts_change_records(tmp_path, format_config):
path = tmp_path / "data.kantadb"
kanta = make_kanta(path, Data, format_config)
await kanta.open(log=False)
# The bootstrap record is the first diff: state is at revision 1.
assert kanta.rev == 1
with kanta.transaction("inc", log=False) as data:
data.counter += 1
assert kanta.rev == 2
# A no-op transaction queues no record and does not advance the revision.
with kanta.transaction("noop", log=False):
pass
assert kanta.rev == 2
await kanta.close()
@pytest.mark.asyncio
async def test_revision_restored_on_reopen(tmp_path, format_config):
path = tmp_path / "data.kantadb"
kanta = make_kanta(path, Data, format_config)
await kanta.open(log=False)
for _ in range(3):
with kanta.transaction("inc", log=False) as data:
data.counter += 1
await kanta.close()
kanta2 = make_kanta(path, Data, format_config)
await kanta2.open(log=False)
assert kanta2.rev == 4
await kanta2.close()
@pytest.mark.asyncio
async def test_snapshot_carries_revision(tmp_path, format_config):
path = tmp_path / "data.kantadb"
kanta = make_kanta(path, Data, format_config)
await kanta.open(log=False)
for _ in range(2):
with kanta.transaction("inc", log=False) as data:
data.counter += 1
await kanta.flush()
kanta.request_snapshot()
kanta._impl.maybe_snapshot()
await kanta.close()
snap = read_last_snapshot(path, format_config)
assert snap is not None
assert snap.r == 3
assert snap.k == META_FORMAT
# Reopening from the snapshot keeps the absolute revision.
kanta2 = make_kanta(path, Data, format_config)
await kanta2.open(log=False)
assert kanta2.rev == 3
with kanta2.transaction("inc", log=False) as data:
data.counter += 1
assert kanta2.rev == 4
await kanta2.close()
@pytest.mark.asyncio
async def test_legacy_snapshot_recounted_and_rewritten(tmp_path, format_config):
"""Meta migration k1 rewrites a k0 file in full (see kanta/metamigr.py)."""
path = tmp_path / "data.kantadb"
kanta = make_kanta(path, Data, format_config)
await kanta.open(log=False)
for _ in range(2):
with kanta.transaction("inc", log=False) as data:
data.counter += 1
await kanta.flush()
kanta.request_snapshot()
kanta._impl.maybe_snapshot()
await kanta.close()
# Simulate a pre-revision database: rewrite the last snapshot without k,
# with a wrong r, then append one more change record.
_, serializer_cls = format_config
serializer = serializer_cls()
framer = serializer.framer_cls()
records = read_all(path, format_config)
out = bytearray()
for rec in records:
if isinstance(rec, Snapshot):
rec = Snapshot(ts=rec.ts, v=rec.v, state=rec.state, m=rec.m, r=0)
out += framer.frame_snapshot(
serializer.encode(rec), record_offset=len(out)
)
else:
out += framer.frame_change(
serializer.encode(rec), record_offset=len(out)
)
extra = ChangeRecord(ts=T0, a="inc", v=0, diff={"counter": 3})
out += framer.frame_change(serializer.encode(extra), record_offset=len(out))
path.write_bytes(bytes(out))
kanta2 = make_kanta(path, Data, format_config)
await kanta2.open(log=False)
# bootstrap + 2 changes + trailing change = revision 4, despite r=0 above.
assert kanta2.rev == 4
assert kanta2.data.counter == 3
await kanta2.close()
# The file was rewritten in full: all change records preserved in order,
# the legacy intermediate snapshot dropped, and a k1 closing snapshot
# carrying the total revision appended.
records = read_all(path, format_config)
changes = [r for r in records if isinstance(r, ChangeRecord)]
assert [c.a for c in changes] == ["bootstrap", "inc", "inc", "inc"]
snapshots = [r for r in records if isinstance(r, Snapshot)]
assert len(snapshots) == 1
assert isinstance(records[-1], Snapshot)
closing = snapshots[0]
assert closing.k == META_FORMAT
assert closing.r == 4
assert closing.state["counter"] == 3
# The next open trusts the stored r (file unchanged this time).
content = path.read_bytes()
kanta3 = make_kanta(path, Data, format_config)
await kanta3.open(log=False)
assert kanta3.rev == 4
await kanta3.close()
assert path.read_bytes() == content
@pytest.mark.asyncio
async def test_legacy_migration_preserves_leading_snapshot(tmp_path, format_config):
"""A legacy file starting with a snapshot keeps it, re-stamped k1."""
path = tmp_path / "data.kantadb"
_, serializer_cls = format_config
serializer = serializer_cls()
framer = serializer.framer_cls()
snap_ts = T0 - timedelta(days=40)
snap_m = T0 - timedelta(days=41)
leading = Snapshot(ts=snap_ts, v=3, state={"users": {}, "counter": 7}, m=snap_m)
out = bytearray()
out += framer.frame_snapshot(serializer.encode(leading), record_offset=0)
for i in range(2):
rec = ChangeRecord(ts=T0, a=f"inc{i}", v=3, diff={"counter": 8 + i})
out += framer.frame_change(serializer.encode(rec), record_offset=len(out))
path.write_bytes(bytes(out))
kanta = make_kanta(path, Data, format_config)
await kanta.open(log=False)
# Only the in-file history is countable: 2 change records.
assert kanta.rev == 2
assert kanta.data.counter == 9
await kanta.close()
records = read_all(path, format_config)
assert isinstance(records[0], Snapshot)
assert records[0].k == META_FORMAT
assert records[0].r == 0
assert records[0].ts == snap_ts
assert records[0].m == snap_m
assert records[0].v == 3
assert records[0].state == {"users": {}, "counter": 7}
changes = [r for r in records if isinstance(r, ChangeRecord)]
assert [c.a for c in changes] == ["inc0", "inc1"]
assert isinstance(records[-1], Snapshot)
assert records[-1].k == META_FORMAT
assert records[-1].r == 2
assert records[-1].state["counter"] == 9
@pytest.mark.asyncio
async def test_rotation_of_legacy_file_migrates(tmp_path, format_config):
"""Rotation and meta migration coincide: rotation already produces k1
snapshots with revisions counted from pre-rotation record positions."""
path = tmp_path / "data.kantadb"
_, serializer_cls = format_config
serializer = serializer_cls()
framer = serializer.framer_cls()
def change(action, day, counter):
return ChangeRecord(
ts=T0 + day * timedelta(days=1), a=action, v=0, diff={"counter": counter}
)
legacy_snap = Snapshot(
ts=T0 - 40 * timedelta(days=1), v=0, state={"users": {}, "counter": 1}, r=0
)
out = bytearray()
out += framer.frame_change(
serializer.encode(change("bootstrap", -41, 0)), record_offset=len(out)
)
out += framer.frame_change(
serializer.encode(change("day-40", -40, 1)), record_offset=len(out)
)
out += framer.frame_snapshot(
serializer.encode(legacy_snap), record_offset=len(out)
)
out += framer.frame_change(
serializer.encode(change("day-20", -20, 2)), record_offset=len(out)
)
out += framer.frame_change(
serializer.encode(change("day-5", -5, 3)), record_offset=len(out)
)
path.write_bytes(bytes(out))
cell = [T0]
kanta = make_kanta(path, Data, format_config, retention=30 * timedelta(days=1))
kanta.clock(make_clock(cell))
await kanta.open(log=False)
assert kanta.rev == 4
assert kanta.data.counter == 3
await kanta.close()
# The rotation's leading snapshot is k1 with the pre-rotation revision.
records = read_all(path, format_config)
assert isinstance(records[0], Snapshot)
assert records[0].k == META_FORMAT
assert records[0].r == 2
assert [r.a for r in records if isinstance(r, ChangeRecord)] == [
"day-20",
"day-5",
]
# No further rewriting on the next open.
content = path.read_bytes()
kanta2 = make_kanta(path, Data, format_config)
await kanta2.open(log=False)
assert kanta2.rev == 4
await kanta2.close()
assert path.read_bytes() == content
@pytest.mark.asyncio
async def test_legacy_snapshot_readonly_recounts_without_writing(
tmp_path, format_config
):
path = tmp_path / "data.kantadb"
kanta = make_kanta(path, Data, format_config)
await kanta.open(log=False)
with kanta.transaction("inc", log=False) as data:
data.counter += 1
await kanta.close()
# Inject a legacy snapshot (no k marker) at the start; the whole file is
# re-framed with a single framer so binary checksums stay valid.
_, serializer_cls = format_config
serializer = serializer_cls()
framer = serializer.framer_cls()
legacy = Snapshot(ts=T0, v=0, state={"users": {}, "counter": 0}, r=0)
out = bytearray()
out += framer.frame_snapshot(serializer.encode(legacy), record_offset=0)
for rec in read_all(path, format_config):
out += framer.frame_change(serializer.encode(rec), record_offset=len(out))
path.write_bytes(bytes(out))
content = path.read_bytes()
kanta2 = make_kanta(path, Data, format_config)
await kanta2.open(readonly=True, log=False)
assert kanta2.rev == 2
await kanta2.close()
assert path.read_bytes() == content
@pytest.mark.asyncio
async def test_rotation_preserves_revisions(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 * timedelta(days=1))
kanta.clock(make_clock(cell))
await kanta.open(log=False)
# bootstrap + 3 changes = revision 4 before and after the rotation.
assert kanta.rev == 4
await kanta.close()
records = read_all(path, format_config)
leading = records[0]
assert isinstance(leading, Snapshot)
# One change was dropped (bootstrap + day-40 are before the cutoff
# position... bootstrap is a change record too: 2 records before).
assert leading.r == 2
assert leading.k == META_FORMAT
kanta2 = make_kanta(path, Data, format_config)
await kanta2.open(log=False)
assert kanta2.rev == 4
with kanta2.transaction("inc", log=False) as data:
data.counter += 1
assert kanta2.rev == 5
await kanta2.close()
def test_cli_revision_snapshot_base_file(tmp_path, capsys):
"""File = snapshot {r: 2} + one change; the two producing changes are gone."""
path = tmp_path / "data.kantadb"
serializer = JsonSerializer()
framer = serializer.framer_cls()
snap = Snapshot(
ts=T0, v=0, state={"foo": 1, "bar": 1}, r=2, k=META_FORMAT
)
out = bytearray()
out += framer.frame_snapshot(serializer.encode(snap), record_offset=0)
change = ChangeRecord(ts=T0, a="add_baz", v=0, diff={"baz": 1})
out += framer.frame_change(serializer.encode(change), record_offset=len(out))
path.write_bytes(bytes(out))
capsys.readouterr()
# r1's producing change is gone and no snapshot holds state 1: error.
assert _run_cli(path, "-r", "r1") == 2
assert "r1" in capsys.readouterr().err
# r2 prints the snapshot line (it literally carries r: 2) and its state.
assert _run_cli(path, "-r", "r2", "-o", "-") == 0
out = capsys.readouterr()
assert "snapshot s0" in out.err
assert "r2" in out.err
assert json.loads(out.out) == {"foo": 1, "bar": 1}
# r3 is the change, replayed on the snapshot base; state includes baz.
assert _run_cli(path, "-r", "r3", "-o", "-") == 0
out = capsys.readouterr()
assert "add_baz" in out.err
assert json.loads(out.out) == {"foo": 1, "bar": 1, "baz": 1}
# Single index and half-open range agree.
assert _run_cli(path, "-r", "r2") == 0
single = capsys.readouterr().err
assert _run_cli(path, "-r", "r2:r3") == 0
assert capsys.readouterr().err == single
def _run_cli(path, *args):
from kanta.__main__ import main
return main([str(path), *args])
def test_cli_revision_ranges(tmp_path, capsys):
path = tmp_path / "data.kantadb"
async def build():
kanta = make_kanta(path, Data, JSON)
await kanta.open(log=False)
for i in range(3):
with kanta.transaction(f"inc{i}", log=False) as data:
data.counter += 1
await kanta.close()
asyncio.run(build())
capsys.readouterr()
# A revision selects the record that produced that state, with the
# state: r1 and r1:r2 both show only the bootstrap and output state 1.
assert _run_cli(path, "-r", "r1") == 0
single = capsys.readouterr().err
assert "bootstrap" in single
assert "inc0" not in single
assert _run_cli(path, "-r", "r1:r2") == 0
assert capsys.readouterr().err == single
assert _run_cli(path, "-r", "r1", "-o", "-") == 0
assert json.loads(capsys.readouterr().out) == {"users": {}, "counter": 0}
# r1..r2 (inclusive end) adds the next change and outputs state 2.
assert _run_cli(path, "-r", "r1..r2", "-o", "-") == 0
out = capsys.readouterr()
assert "inc0" in out.err
assert json.loads(out.out) == {"users": {}, "counter": 1}
# r0 is not meaningful: no record produces the empty initial state.
assert _run_cli(path, "-r", "r0") == 2
assert "r0" in capsys.readouterr().err
# Range by revision.
assert _run_cli(path, "-r", "r2:r4") == 0
err = capsys.readouterr().err
assert "inc0" in err
assert "inc1" in err
assert "inc2" not in err
# Negative revisions are rejected.
assert _run_cli(path, "-r", "r-1") == 1
assert "negative" in capsys.readouterr().err
# Out-of-range single revision is a range error (exit 2).
assert _run_cli(path, "-r", "r99") == 2
assert "r99" in capsys.readouterr().err
# The same bounds as a range truncate Python-style instead: an empty
# display and the closest known state, not an error.
assert _run_cli(path, "-r", "r99:r100", "-o", "-") == 0
out = capsys.readouterr()
assert "bootstrap" not in out.err
assert "inc" not in out.err
assert json.loads(out.out) == {"users": {}, "counter": 3}
# Only the end beyond the last revision: truncated to end of file.
assert _run_cli(path, "-r", "r3:r99", "-o", "-") == 0
out = capsys.readouterr()
assert "inc1" in out.err
assert "inc2" in out.err
assert json.loads(out.out) == {"users": {}, "counter": 3}
def test_cli_revision_after_rotation(tmp_path, capsys):
path = tmp_path / "data.kantadb"
async def build():
await write_history(path, JSON, days=[-40, -20, -5])
cell = [T0]
kanta = make_kanta(
path, Data, JSON, retention=30 * timedelta(days=1)
)
kanta.clock(make_clock(cell))
await kanta.open(log=False)
await kanta.close()
asyncio.run(build())
capsys.readouterr()
# The leading snapshot holds revision 2 (bootstrap + day-40). Since the
# change that produced state 2 is gone, r2 shows the snapshot itself and
# outputs the snapshot state.
assert _run_cli(path, "-r", "r2", "-o", "-") == 0
out = capsys.readouterr()
assert "snapshot s0" in out.err
assert "r2" in out.err
assert "day-20" not in out.err
assert json.loads(out.out) == {"users": {}, "counter": 1}
# r1 has no record in this file: a single index does not truncate.
assert _run_cli(path, "-r", "r1") == 2
assert "r1" in capsys.readouterr().err
# A range truncates like Python slicing instead: r1:r3 starts at the
# beginning of the file and stops before the r3 record, showing just the
# leading snapshot.
assert _run_cli(path, "-r", "r1:r3") == 0
err = capsys.readouterr().err
assert "snapshot s0" in err
assert "r2" in err
assert "day-20" not in err
# Absolute revisions reach the retained change records.
assert _run_cli(path, "-r", "r3:r4") == 0
err = capsys.readouterr().err
assert "day-20" in err
assert "snapshot" not in err
assert "day-5" not in err