From f954fdd595245909671dff30c43ea67523743a8f Mon Sep 17 00:00:00 2001 From: Libby Date: Fri, 13 Nov 2026 19:52:51 +0000 Subject: [PATCH 1/6] cultcache-py: conditional update, content version, store-minted stored_at (update-with-retry-py) Replaces the blind update and update_global with update(document, key, apply, max_attempts) over store read and commit_if; push and push_all restamp a replacing write later and report the stamp. Co-Authored-By: Claude Sonnet 5.5 --- .../cultcache-py/src/cultcache_py/__init__.py | 4 +- .../src/cultcache_py/backing_store.py | 11 +- .../cultcache-py/src/cultcache_py/cache.py | 193 +++++++++++++++--- .../cultcache-py/src/cultcache_py/stores.py | 105 ++++++++-- .../tests/test_content_version_vector.py | 38 ++++ packages/cultcache-py/tests/test_cultcache.py | 2 +- .../cultcache-py/tests/test_unique_index.py | 1 - 7 files changed, 305 insertions(+), 49 deletions(-) create mode 100644 packages/cultcache-py/tests/test_content_version_vector.py diff --git a/packages/cultcache-py/src/cultcache_py/__init__.py b/packages/cultcache-py/src/cultcache_py/__init__.py index a2752511..0c26cb25 100644 --- a/packages/cultcache-py/src/cultcache_py/__init__.py +++ b/packages/cultcache-py/src/cultcache_py/__init__.py @@ -1,7 +1,7 @@ from .backing_store import BackingStore, CultCacheEnvelope from .backing_store import CultCacheSchemaCatalogEntry, CultCacheSchemaCatalogMember from . import leaves -from .cache import CultCache, CultCacheBuilder, ForeignRecord +from .cache import CultCache, CultCacheBuilder, ForeignRecord, UpdateOutcome, UpdateResult from .documents import ( MAX_PAYLOAD_DEPTH, MAX_SLOT_KEY, @@ -42,6 +42,8 @@ __all__ = [ "StoreUnreadableError", "UnionArm", "UnreadSlots", + "UpdateOutcome", + "UpdateResult", "read_payload_tree", "leaves", "database_entry_field", diff --git a/packages/cultcache-py/src/cultcache_py/backing_store.py b/packages/cultcache-py/src/cultcache_py/backing_store.py index bded4b11..c20ac661 100644 --- a/packages/cultcache-py/src/cultcache_py/backing_store.py +++ b/packages/cultcache-py/src/cultcache_py/backing_store.py @@ -58,10 +58,17 @@ class CultCacheEnvelope: class BackingStore(Protocol): + """The members every store has. A store may also offer `read(key) -> tuple[CultCacheEnvelope, str] | None` (the record at the + key, of any type, and its content version, from one read) and `commit_if(envelope, expected: str | None) -> str | None` (write + only if the record at the key is still the version `expected`, None meaning no record; returns the stored_at written, or None + with nothing written). They are looked up with getattr, not Protocol members, and `CultCache.update` needs both. `push` and + `push_all` may report the stored_at they wrote (`str` and `dict[str, str]` by key); None keeps the cache's own stamp. A store + stamps a record that replaces another strictly later than the one it replaces.""" + def pull_all(self) -> list[CultCacheEnvelope]: ... - def push(self, envelope: CultCacheEnvelope) -> None: + def push(self, envelope: CultCacheEnvelope) -> str | None: ... def delete(self, type: str, key: str) -> None: @@ -74,7 +81,7 @@ class BackingStore(Protocol): that can tell nothing-at-the-path overrides this, and the default push_all then writes where nothing is on a soft push.""" return self.pull_all() - def push_all(self, envelopes: list[CultCacheEnvelope], soft: bool = False) -> None: + def push_all(self, envelopes: list[CultCacheEnvelope], soft: bool = False) -> dict[str, str] | None: # Soft: write only where pull_or_absent says nothing is; skip over a store it read, empty included; its raise refuses. pulled = self.pull_or_absent() if soft and pulled is not None: diff --git a/packages/cultcache-py/src/cultcache_py/cache.py b/packages/cultcache-py/src/cultcache_py/cache.py index 80a12706..ebf9b2d2 100644 --- a/packages/cultcache-py/src/cultcache_py/cache.py +++ b/packages/cultcache-py/src/cultcache_py/cache.py @@ -1,5 +1,6 @@ from __future__ import annotations +import enum import functools import math import threading @@ -9,13 +10,39 @@ from typing import Any, Callable, Generic, Iterable, TypeVar from .backing_store import BackingStore, CultCacheEnvelope from .documents import DocumentDefinition, UnreadSlots, extract_value -from .stores import SchemaConflictError, retyped, written_owns +from .stores import SchemaConflictError, _parse_stored_at, retyped, stored_at_over, written_owns T = TypeVar("T") F = TypeVar("F", bound=Callable[..., Any]) GLOBAL_KEY = "__global__" +class UpdateOutcome(enum.Enum): + UPDATED = "updated" + DECLINED = "declined" + LOST_RACE = "lost_race" + + +@dataclass(frozen=True) +class UpdateResult(Generic[T]): + """What `CultCache.update` did and the cache's record at the key after it (None when it holds nothing). The document decides + nothing: it is a read after the outcome.""" + + outcome: UpdateOutcome + document: T | None + + +@dataclass(frozen=True) +class _Expected: + """What an update's commit requires to still hold. A cache with no store requires the held envelope object at the key (None: + nothing held); a store-backed one requires the type's home to be `store` and the record there to have content version + `version` (None: no record).""" + + held: CultCacheEnvelope | None = None + store: BackingStore | None = None + version: str | None = None + + def _locked(method: F) -> F: """Runs a read holding the cache's lock. Reads are reentrant and may run inside a mutation.""" @@ -482,7 +509,7 @@ class CultCache: def put(self, document: DocumentDefinition[T], key: str, value: T) -> None: self._put(document, key, value) - def _put(self, document: DocumentDefinition[T], key: str, value: T) -> None: + def _put(self, document: DocumentDefinition[T], key: str, value: T, expected: _Expected | None = None) -> str: self._assert_registered(document) if document.global_document and key != GLOBAL_KEY: raise CultCacheError(f"Global document {document.type} must use key {GLOBAL_KEY}") @@ -494,7 +521,7 @@ class CultCache: schema_id=catalog_entry.schema_id, catalog_entry=catalog_entry, ) - self._write(document, [(envelope, value)], batch=False) + return self._write(document, [(envelope, value)], batch=False, expected=expected) @_mutating def put_envelope(self, document: DocumentDefinition[T], envelope: CultCacheEnvelope) -> T: @@ -562,9 +589,12 @@ class CultCache: *, batch: bool, unread: dict[str, UnreadSlots | None] | None = None, - ) -> None: + expected: _Expected | None = None, + ) -> str: """Validates every record (home and name and index extractors included) before the store is - touched; once the store accepts, applying to the cache has no fallible step.""" + touched; once the store accepts, applying to the cache has no fallible step. With `expected` (an update's commit) the write + also requires the record to be unchanged since it was read and returns 'mismatch', having written and installed nothing, + when it is not; otherwise 'committed'. A record is held under the stored_at its store reports writing.""" type = document.type # A batch naming a key twice lands the last record for it, as C# CultCacheBatch stages one operation per key. The # judge, the store and the apply all read this one list. @@ -584,13 +614,34 @@ class CultCache: ) store = self._store_for_type(type) legacy_key = self._state.legacy_global_keys.get(type) - if store is not None: - if batch: - store.push_all([envelope for envelope, _ in records]) + stamps: dict[str, Any] = {} + if expected is not None: + if legacy_key is not None: + raise self._legacy_global_refusal(document, legacy_key) + if expected.store is None: + held = self._state.envelopes.get(type, {}).get(records[0][0].key) + if store is not None or held is not expected.held: + return "mismatch" + elif store is not expected.store: + return "mismatch" else: - store.push(records[0][0]) + stamp = store.commit_if(records[0][0], expected.version) # type: ignore[attr-defined] + if stamp is None: + return "mismatch" + stamps[records[0][0].key] = stamp + elif store is not None: + if batch: + reported = store.push_all([envelope for envelope, _ in records]) + stamps = reported if isinstance(reported, dict) else {} + else: + stamps[records[0][0].key] = store.push(records[0][0]) if legacy_key is not None: self._retire_legacy_global(store, type, legacy_key) + else: + for envelope, _ in records: + held_at = next((held for held in (by_type.get(envelope.key) for by_type in self._state.envelopes.values()) if held is not None), None) + if held_at is not None: + stamps[envelope.key] = stored_at_over(envelope.stored_at, held_at.stored_at) self._state.legacy_global_keys.pop(type, None) values = self._state.values.setdefault(type, {}) envelopes = self._state.envelopes.setdefault(type, {}) @@ -598,6 +649,9 @@ class CultCache: for other, key in displaced: self._drop(other, key) for envelope, value, keys in admitted: + stamp = stamps.get(envelope.key) + if isinstance(stamp, str) and _parse_stored_at(stamp) is not None: + envelope = replace(envelope, stored_at=stamp) self._remove_lookups(type, envelope.key) values[envelope.key] = value envelopes[envelope.key] = envelope @@ -663,18 +717,34 @@ class CultCache: if held is None: continue type, slots = held - raise SchemaConflictError( - self._state.envelopes[type][key].schema_id or type, - [self._state.documents[type].catalog_entry().schema_name], - key, - message=( - f"{self._state.documents[type].catalog_entry().schema_name} record {key} is read-only: {slots.type_name} was read with " - f"{slots.slot_count} slots where this build declares {slots.declared_count} ({slots.occurrences} object(s) in the " - "record). Writing it would drop data a newer writer stored; read it with types that declare those slots, or " - "write it from that build." - ), - unread_slots=slots, - ) + raise self._read_only_refusal(self._state.documents[type], key, self._state.envelopes[type][key].schema_id, slots) + + @staticmethod + def _read_only_refusal( + document: DocumentDefinition[Any], key: str, schema_id: str | None, slots: UnreadSlots + ) -> SchemaConflictError: + """The one refusal of a write to a record that carries slots this build would drop: a record the cache holds read-only, + or one an update reads from the store.""" + name = document.catalog_entry().schema_name + return SchemaConflictError( + schema_id or document.type, + [name], + key, + message=( + f"{name} record {key} is read-only: {slots.type_name} was read with " + f"{slots.slot_count} slots where this build declares {slots.declared_count} ({slots.occurrences} object(s) in the " + "record). Writing it would drop data a newer writer stored; read it with types that declare those slots, or " + "write it from that build." + ), + unread_slots=slots, + ) + + @staticmethod + def _legacy_global_refusal(document: DocumentDefinition[Any], legacy_key: str) -> CultCacheError: + return CultCacheError( + f"Global {document.type} is held under the legacy key {legacy_key!r}; update cannot copy it. " + "put_global writes the global and retires the legacy key." + ) @staticmethod def _foreign_refusal(record: ForeignRecord) -> SchemaConflictError: @@ -707,20 +777,81 @@ class CultCache: self._assert_global(document) self._put(document, GLOBAL_KEY, value) - @_mutating - def update(self, document: DocumentDefinition[T], key: str, updater: Any) -> T: - return self._update(document, key, updater) + def update( + self, document: DocumentDefinition[T], key: str, apply: Callable[[T | None], T | None], *, max_attempts: int = 5 + ) -> UpdateResult[T]: + """Reads the record at the key as its home store holds it, applies `apply` to a fresh copy of it, and commits the result + only if the record is still the one that was read; when another writer got there first it reads again and runs `apply` + again, up to `max_attempts` times. - def _update(self, document: DocumentDefinition[T], key: str, updater: Any) -> T: - current = self.get_required(document, key) - updated = updater(current) - self._put(document, key, updated) - return updated + `apply` may run up to `max_attempts` times, on a fresh copy of the record as the store holds it now (None when the key + holds nothing), which it may change and return. It must not write the same key through this cache or have side effects. + Returning None writes nothing (DECLINED). A lost race that outlasts `max_attempts` returns LOST_RACE and does not raise; + a refusal, or a raise from `apply`, propagates and is not retried. The write is durable when update returns. The + result's document is the cache's record after the outcome. `put` is the blind write. A cache with no store copies its + held instance; a store without `read` and `commit_if` (the JSON lines store among them) refuses before `apply` runs.""" + if not callable(apply): + raise TypeError("CultCache update needs a function to apply.") + if not isinstance(max_attempts, int) or isinstance(max_attempts, bool): + raise TypeError(f"CultCache update max_attempts must be an int, not {type(max_attempts).__name__}.") + if max_attempts < 1: + raise ValueError(f"CultCache update max_attempts must be at least 1, not {max_attempts}.") + for attempt in range(1, max_attempts + 1): + copy, expected = self._read_for_update(document, key) + # Outside the cache's lock and the store's: apply may take its time and may write other keys through this cache. + next_value = apply(copy) + if next_value is None: + return UpdateResult(UpdateOutcome.DECLINED, self.get(document, key)) + if self._commit_update(document, key, next_value, expected) == "committed": + return UpdateResult(UpdateOutcome.UPDATED, self.get(document, key)) + # Display only: the cache shows what the winner left. + if self._state.stores_by_type or self._state.generic_store is not None: + self.pull_all_backing_stores() + return UpdateResult(UpdateOutcome.LOST_RACE, self.get(document, key)) @_mutating - def update_global(self, document: DocumentDefinition[T], updater: Any) -> T: + def _read_for_update(self, document: DocumentDefinition[T], key: str) -> tuple[T | None, _Expected]: + """One attempt's read: the copy to hand `apply` and the condition its commit holds. Every refusal that does not depend on + what `apply` returns is raised here, so `apply` never runs for a write that cannot land. Installs nothing.""" + self._assert_registered(document) + self._refuse_at(document, key) + legacy_key = self._state.legacy_global_keys.get(document.type) + if legacy_key is not None: + raise self._legacy_global_refusal(document, legacy_key) + store = self._store_for_type(document.type) + if store is None: + held = self._state.envelopes.get(document.type, {}).get(key) + return (None if held is None else document.decode_payload(held.payload)), _Expected(held=held) + read, commit_if = getattr(store, "read", None), getattr(store, "commit_if", None) + if read is None or commit_if is None: + raise CultCacheError( + f"Backing store {type(store).__name__} for {document.type} cannot write conditionally; " + "update needs a store with read and commit_if." + ) + if document.global_document: + for envelope in store.pull_all(): + if envelope.key != GLOBAL_KEY and self._resolve_document_for_envelope(envelope) is document: + raise self._legacy_global_refusal(document, envelope.key) + found = read(key) + if found is None: + return None, _Expected(store=store) + envelope, version = found + if self._resolve_document_for_envelope(envelope) is not document: + raise CultCacheError(f"Record {key} is not a {document.type}; update cannot copy it.") + value, unread = self._decode_marked(document, envelope) + if unread is not None: + raise self._read_only_refusal(document, key, envelope.schema_id, unread) + return value, _Expected(store=store, version=version) + + @_mutating + def _commit_update(self, document: DocumentDefinition[T], key: str, value: T, expected: _Expected) -> str: + return self._put(document, key, value, expected) + + def update_global( + self, document: DocumentDefinition[T], apply: Callable[[T | None], T | None], *, max_attempts: int = 5 + ) -> UpdateResult[T]: self._assert_global(document) - return self._update(document, GLOBAL_KEY, updater) + return self.update(document, GLOBAL_KEY, apply, max_attempts=max_attempts) @_mutating def delete(self, document: DocumentDefinition[Any], key: str) -> None: diff --git a/packages/cultcache-py/src/cultcache_py/stores.py b/packages/cultcache-py/src/cultcache_py/stores.py index 894252c8..c7211640 100644 --- a/packages/cultcache-py/src/cultcache_py/stores.py +++ b/packages/cultcache-py/src/cultcache_py/stores.py @@ -1,13 +1,17 @@ from __future__ import annotations import base64 +import hashlib import json import os +import re import sys import threading import time import uuid from contextlib import contextmanager +from dataclasses import replace +from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any, Callable, Iterator, TypeVar @@ -107,14 +111,23 @@ class JsonLinesBackingStore: raise ValueError(f"Malformed CultCache JSONL envelope at {self.path}:{line_number}: missing {exc}") from exc return envelopes - def push(self, envelope: CultCacheEnvelope) -> None: - self.push_all([envelope]) + def push(self, envelope: CultCacheEnvelope) -> str | None: + return self.push_all([envelope])[envelope.key] + + def push_all(self, envelopes: list[CultCacheEnvelope], soft: bool = False) -> dict[str, str]: + """Merges the envelopes, each stored later than any record it replaces, and returns the stamps written by key + ({} when a soft push skipped).""" + stamps: dict[str, str] = {} - def push_all(self, envelopes: list[CultCacheEnvelope], soft: bool = False) -> None: def merge(existing: dict[tuple[str, str], CultCacheEnvelope]) -> None: for envelope in envelopes: + replaced = existing.get((envelope.type, envelope.key)) + if replaced is not None: + envelope = replace(envelope, stored_at=stored_at_over(envelope.stored_at, replaced.stored_at)) existing[(envelope.type, envelope.key)] = envelope + stamps[envelope.key] = envelope.stored_at self._rewrite(merge, soft) + return stamps def delete(self, type: str, key: str) -> None: self._rewrite(lambda existing: existing.pop((type, key), None)) @@ -177,23 +190,40 @@ class SingleFileMessagePackBackingStore: except Exception as exc: raise StoreUnreadableError(self.path, exc) from exc - def push(self, envelope: CultCacheEnvelope) -> None: + def push(self, envelope: CultCacheEnvelope) -> str | None: with self._write_transaction(): - self._replace_all([envelope], self.pull_all()) + return self._replace_all([envelope], self.pull_all())[envelope.key] - def push_all(self, envelopes: list[CultCacheEnvelope], soft: bool = False) -> None: + def push_all(self, envelopes: list[CultCacheEnvelope], soft: bool = False) -> dict[str, str]: """A soft push writes only where nothing is at the path: a store the reader reads is left byte for byte, and a file it - cannot read is refused by the same read, as in TS and Rust.""" + cannot read is refused by the same read, as in TS and Rust. Returns the stamps written by key ({} when skipped).""" with self._write_transaction(): arrived = self._pull() if soft and arrived is not None: - return - self._replace_all(list(envelopes), arrived or []) + return {} + return self._replace_all(list(envelopes), arrived or []) def delete(self, type: str, key: str) -> None: with self._write_transaction(): self._replace_all([], [item for item in self.pull_all() if not (item.type == type and item.key == key)]) + def read(self, key: str) -> tuple[CultCacheEnvelope, str] | None: + """The record at the key, of any type, and its content version, from one read; None when the key holds nothing.""" + for envelope in self._pull() or []: + if envelope.key == key: + return envelope, record_version(envelope) + return None + + def commit_if(self, envelope: CultCacheEnvelope, expected: str | None) -> str | None: + """Writes the envelope only if the record at its key (any type) is still the one whose version is `expected` + (None: no record), judged under the write lock; returns the stored_at written, or None with nothing written.""" + with self._write_transaction(): + arrived = self._pull() or [] + current = next((item for item in arrived if item.key == envelope.key), None) + if (None if current is None else record_version(current)) != expected: + return None + return self._replace_all([envelope], arrived)[envelope.key] + @contextmanager def _write_transaction(self) -> Iterator[None]: """The thread lock, then the cross-process lock on `.lock`, held across the read and the replace so two @@ -203,8 +233,10 @@ class SingleFileMessagePackBackingStore: with store_write_lock(self.path, self.lock_timeout): yield - def _replace_all(self, supplied: list[CultCacheEnvelope], arrived: list[CultCacheEnvelope]) -> None: - commit_store_file(self.path, msgpack.packb(_encode_snapshot(supplied, arrived, self._format), use_bin_type=True)) + def _replace_all(self, supplied: list[CultCacheEnvelope], arrived: list[CultCacheEnvelope]) -> dict[str, str]: + snapshot, stamps = _encode_snapshot(supplied, arrived, self._format) + commit_store_file(self.path, msgpack.packb(snapshot, use_bin_type=True)) + return stamps @@ -305,7 +337,7 @@ def written_owns(held: CultCacheEnvelope, written: CultCacheEnvelope) -> bool: def _encode_snapshot( supplied: list[CultCacheEnvelope], arrived: list[CultCacheEnvelope], format_version: str -) -> list[Any]: +) -> tuple[list[Any], dict[str, str]]: # A key names one record whatever its type, so a supplied record replaces the arrived one by key alone, and only of the # type that owns it: retyping a key is a removal and an add, so a record another type owns under the key refuses the write. held_by_key = {envelope.key: envelope for envelope in arrived} @@ -314,8 +346,12 @@ def _encode_snapshot( for held in (landing.get(envelope.key), held_by_key.get(envelope.key)): if held is not None and not written_owns(held, envelope): raise retyped(held, envelope) + held = held_by_key.get(envelope.key) + if held is not None: + envelope = replace(envelope, stored_at=stored_at_over(envelope.stored_at, held.stored_at)) landing[envelope.key] = envelope supplied = list(landing.values()) + stamps = {envelope.key: envelope.stored_at for envelope in supplied} replaced = set(landing) kept = [envelope for envelope in arrived if envelope.key not in replaced] envelopes = [*kept, *supplied] @@ -330,7 +366,7 @@ def _encode_snapshot( [envelope.key, _schema_id_for(envelope), envelope.stored_at, envelope.payload] for envelope in sorted(envelopes, key=lambda item: item.key) ] - return [format_version, catalog, records] + return [format_version, catalog, records], stamps def commit_store_file(path: Path, data: bytes) -> None: """Writes `data` to a unique staged file beside the store, fsyncs it, replaces the store with it and, off Windows, @@ -739,6 +775,49 @@ def _schema_id_for(envelope: CultCacheEnvelope) -> str: return envelope.schema_id or envelope.type +def _parse_stored_at(text: Any) -> datetime | None: + """The instant a stored_at text names, or None: a trailing Z is UTC, fraction digits past the sixth are cut (C# writes + seven), a text with no offset is read as UTC. Written to read on Python 3.10, whose fromisoformat takes neither.""" + if not isinstance(text, str): + return None + body = text.strip() + if body.endswith(("Z", "z")): + body = body[:-1] + "+00:00" + body = re.sub(r"(\.\d{6})\d+", r"\1", body) + try: + parsed = datetime.fromisoformat(body) + except ValueError: + return None + return parsed.replace(tzinfo=timezone.utc) if parsed.tzinfo is None else parsed.astimezone(timezone.utc) + + +def mint_stored_at(previous: str | None, now: datetime) -> str: + """A stored_at later than `previous` (C# MintStoredAt): `now`, or the microsecond after `previous` when `now` is not later. + A previous that does not parse leaves `now`.""" + now = now.astimezone(timezone.utc) + last = _parse_stored_at(previous) + if last is not None and now <= last: + now = last + timedelta(microseconds=1) + return now.isoformat(timespec="microseconds") + + +def stored_at_over(staged: str, replaced: str) -> str: + """The stored_at a record replacing `replaced` is written with (C# StoredAtOver): `staged` when both parse and it is strictly + later, else one minted over `replaced`. No content is compared: every replacing write is stamped later.""" + later, earlier = _parse_stored_at(staged), _parse_stored_at(replaced) + if later is not None and earlier is not None and later > earlier: + return staged + return mint_stored_at(replaced, datetime.now(timezone.utc)) + + +def record_version(envelope: CultCacheEnvelope) -> str: + """The content version of a stored record: the lowercase hex SHA-256 of its persisted encoding, the array + [key, schema id, stored_at, payload] exactly as the single-file store writes it + (variants:ruling:write-set-content-version-declared-ownership).""" + record = [envelope.key, _schema_id_for(envelope), envelope.stored_at, envelope.payload] + return hashlib.sha256(msgpack.packb(record, use_bin_type=True)).hexdigest() + + def _default_catalog_entry(envelope: CultCacheEnvelope) -> CultCacheSchemaCatalogEntry: schema_id = _schema_id_for(envelope) return CultCacheSchemaCatalogEntry( diff --git a/packages/cultcache-py/tests/test_content_version_vector.py b/packages/cultcache-py/tests/test_content_version_vector.py new file mode 100644 index 00000000..393784e0 --- /dev/null +++ b/packages/cultcache-py/tests/test_content_version_vector.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +import shutil +import tempfile +import unittest +from pathlib import Path + +import msgpack + +from cultcache_py import SingleFileMessagePackBackingStore +from cultcache_py.stores import record_version + +# variants:ruling:write-set-content-version-declared-ownership. The vector in tests/vectors/content-version is written only by +# the C# reference; every row of it is computed here from the store file, by the store's read and by record_version. +VECTORS = Path(__file__).resolve().parents[3] / "tests" / "vectors" / "content-version" + + +class ContentVersionVectorTests(unittest.TestCase): + def test_every_row_of_the_vector_equals_this_runtimes_version(self) -> None: + rows = msgpack.unpackb((VECTORS / "versions.msgpack").read_bytes(), raw=False) + self.assertEqual([key for key, _ in rows], ["alpha", "beta"]) + directory = Path(tempfile.mkdtemp()) + self.addCleanup(shutil.rmtree, directory, ignore_errors=True) + copy = directory / "store.cc" + shutil.copyfile(VECTORS / "store.cc", copy) + store = SingleFileMessagePackBackingStore(copy) + by_key = {envelope.key: envelope for envelope in store.pull_all()} + for key, version in rows: + with self.subTest(key=key): + found = store.read(key) + self.assertIsNotNone(found) + self.assertEqual(found[1], version) + self.assertEqual(record_version(by_key[key]), version) + self.assertIsNone(store.read("gamma")) + + +if __name__ == "__main__": + unittest.main() diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index 6b95f638..7ff4e297 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -86,7 +86,7 @@ class CultCacheTests(unittest.TestCase): ) cache.pull_all_backing_stores() cache.put(document, "item:potion", Item(name="Potion", category="Consumable", value=50)) - cache.update(document, "item:potion", lambda item: Item(item.name, item.category, item.value + 5)) + cache.put(document, "item:potion", Item(name="Potion", category="Consumable", value=55)) self.assertEqual(cache.get_required(document, "item:potion").value, 55) cache.delete(document, "item:potion") self.assertIsNone(cache.get(document, "item:potion")) diff --git a/packages/cultcache-py/tests/test_unique_index.py b/packages/cultcache-py/tests/test_unique_index.py index accc27dd..6e831b9a 100644 --- a/packages/cultcache-py/tests/test_unique_index.py +++ b/packages/cultcache-py/tests/test_unique_index.py @@ -148,7 +148,6 @@ class UniqueIndexTests(unittest.TestCase): "put": lambda cache, d, k, i: cache.put(d, k, i), "put_envelope": lambda cache, d, k, i: cache.put_envelope(d, envelope(d, k, i)), "put_envelopes": lambda cache, d, k, i: cache.put_envelopes(d, [envelope(d, k, i)]), - "update": lambda cache, d, k, i: cache.update(d, k, lambda _held: i), } for name, write in paths.items(): with self.subTest(alias=alias, path=name): -- 2.47.3 From 1b4f8ffd77db5de8804dd92871d46a357252880c Mon Sep 17 00:00:00 2001 From: Libby Date: Fri, 13 Nov 2026 20:09:50 +0000 Subject: [PATCH 2/6] cultcache-py: update tests, first-call snippet, docs, changelog (update-with-retry-py) Co-Authored-By: Claude Sonnet 5.5 --- docs/start/README.md | 25 +- packages/cultcache-py/CHANGELOG.md | 4 + packages/cultcache-py/README.md | 5 +- .../cultcache-py/src/cultcache_py/cache.py | 1 + .../cultcache-py/tests/test_first_call.py | 37 +- packages/cultcache-py/tests/test_update.py | 665 ++++++++++++++++++ 6 files changed, 731 insertions(+), 6 deletions(-) create mode 100644 packages/cultcache-py/tests/test_update.py diff --git a/docs/start/README.md b/docs/start/README.md index a86f4732..c478467c 100644 --- a/docs/start/README.md +++ b/docs/start/README.md @@ -19,7 +19,7 @@ this page. | **C#** | [first call, tested](#c-first-call); [update, tested](#c-update-with-retry) | [package](../../packages/cultmath/README.md); snippet not yet | [package](../../src/GameCult.Geometry/README.md); snippet not yet | [first call, tested](#c-cultnet-first-call); [package](../../src/GameCult.Networking) | [quickstart](../../src/GameCult.Mesh/docs/durable-node-quickstart.md); run by `scripts/verify-eve-getting-started.ps1` | | **TypeScript** | [first call, tested](#typescript-first-call); [update, tested](#typescript-update-with-retry) | not yet | not yet | [first call, tested](#typescript-cultnet-first-call); [package](../../packages/cultnet-ts/README.md) | [package](../../packages/cultmesh-ts/README.md); snippet not yet; [recipe: two-client .io game](io-game.md), tested with two live clients, browser page not yet run | | **Rust** | [first call, tested](#rust-first-call); [update, tested](#rust-update-with-retry) | [native core](../../packages/cultmath/native/cultmath-core); snippet not yet | not yet | [first call, tested](#rust-cultnet-first-call); [package](../../packages/cultnet-rs/README.md), [examples](../../packages/cultnet-rs/examples) | [package](../../packages/cultmesh-rs); snippet not yet | -| **Python** | [first call, tested](#python-first-call) | not yet | not yet | [first call, tested](#python-cultnet-first-call); [package](../../packages/cultnet-py/README.md) | [package](../../packages/cultmesh-py/README.md); snippet not yet | +| **Python** | [first call, tested](#python-first-call); [update, tested](#python-update-with-retry) | not yet | not yet | [first call, tested](#python-cultnet-first-call); [package](../../packages/cultnet-py/README.md) | [package](../../packages/cultmesh-py/README.md); snippet not yet | For the whole picture of what each piece is for, see the @@ -305,6 +305,29 @@ def first_call() -> Item: return reopened.get_required(item_doc, "item:potion") ``` +## Python update with retry + +`update` changes a record from what the store holds now and commits only if it is still that record. The source is the +`update` region of +[`packages/cultcache-py/tests/test_first_call.py`](../../packages/cultcache-py/tests/test_first_call.py), which uses the +definitions of the first call above; a test in that file fails if the block stops matching the file. + +```python +def restock(path: Path): + cache = ( + CultCache.builder() + .register_document_type(item_doc) + .add_generic_store(SingleFileMessagePackBackingStore(path)) + .build() + ) + cache.pull_all_backing_stores() + return cache.update(item_doc, "item:potion", lambda item: Item("Potion", "Consumable", (item.value if item else 0) + 10)) +``` + +The write is durable when `update` returns. A lost race re-runs the function, so keep it free of side effects; when the +attempts run out the result's `outcome` is `UpdateOutcome.LOST_RACE` instead of a raise. `put` is the blind opt-out: last +writer wins. + ## CultNet first calls CultNet is schema-v0 messages in length-prefixed frames. The first call in each runtime is the same diff --git a/packages/cultcache-py/CHANGELOG.md b/packages/cultcache-py/CHANGELOG.md index 77324c91..a0d5cbe8 100644 --- a/packages/cultcache-py/CHANGELOG.md +++ b/packages/cultcache-py/CHANGELOG.md @@ -4,10 +4,12 @@ ### Added +- `CultCache.update` and `update_global` return `UpdateResult(outcome, document)`; `UpdateOutcome` is `UPDATED`, `DECLINED` or `LOST_RACE`. The single-file store gains `read(key)` (the record and its content version, the SHA-256 C# computes) and `commit_if(envelope, expected)`, both looked up by the cache and not `BackingStore` protocol members (variants:cut_spec:cut-update-with-retry-py). - `read_payload_tree(type_name, payload, decoder=None)`: the exported owner of the one-value, depth-first payload read that `decode_payload` uses, for payloads read outside a definition (`cultnet-py`'s witness bundle decode needs it, so `cultnet-py` now requires `cultcache-py>=0.4.0`). The manifest version is 0.4.0, the minor these Unreleased breaking changes imply, set in this change so the workspace install resolves the new bound; the `## Unreleased` heading becomes `## [0.4.0]` when it is tagged. ### Changed +- A write that replaces a record is stored at a later `stored_at` than the record it replaces, in the single-file store, the JSONL store and a store-less cache, with no exception for identical content (`variants:ruling:py-rs-update-follow-csharp-restamp`); the cache holds the stamp the store wrote. - `store_write_lock` looks for a link at the store path again once the lock is held, so a link put at the path while a writer waited on the lock is refused with `StoreLinkedError` and the lock released, instead of being replaced. `JsonLinesBackingStore` takes no cross-process lock (only its thread lock, under which it already refuses a link), so it has no lock wait to re-check across. Pinned by a test that puts the link while a foreign process holds the lock. - `CultCache.is_writable` does not look at store paths: a link at a store path is not predicted, and the write refuses it with `StoreLinkedError`. Pinned by a test through a cache attached to a linked store. - `JsonLinesBackingStore` writes LF line endings on Windows. It wrote CRLF there, through `write_text` newline translation, before the commit_store_file change (#154, adbbb780). @@ -17,6 +19,8 @@ ### Breaking +- `update(document, key, apply, *, max_attempts=5)` and `update_global(document, apply, *, max_attempts=5)` replace the blind pair. Each attempt reads the record from the home store, runs `apply` on a copy (None when absent) outside every lock, and commits only if the record is unchanged, running `apply` again on a lost race; it returns `UpdateResult`, not the value. `apply` returning None writes nothing; a lost race that outlasts `max_attempts` returns `LOST_RACE`; a store without `read` and `commit_if` (JSON lines among them) refuses before `apply`. `put` is the blind write. +- `push` returns the `stored_at` it wrote and `push_all` a dict of them by key (`{}` when a soft push skipped); a store returning None keeps the cache's own stamp. - `push_all(envelopes, soft=False)` on the single-file and JSONL stores and on the `BackingStore` protocol takes the soft flag, as TypeScript, Rust and C# do: a soft push writes only where nothing is at the path; over a store the reader reads, empty included, it returns without writing, and over a file the reader cannot read it refuses (`StoreUnreadableError` for the single-file store). The `BackingStore` protocol has `pull_or_absent()`: the envelopes the reader reads, `None` when nothing is at the path, a raise when the path cannot be read; its default is `pull_all()`, which cannot say absent, so a store that inherits it is always read. The default `push_all` decides soft by it (writes where it says absent, skips over a store it read, refuses where it raises), and the single-file and JSONL stores expose their pull through it; an inheriting store that can tell nothing-at-the-path overrides `pull_or_absent`, not `push_all`. Before, a soft push through the default always skipped. A store kind that overrides `push_all(envelopes)` without the keyword fails when a caller passes `soft`. - A write at a store path that is a link, live or dangling, is refused with the new `StoreLinkedError` (an `OSError`, exported from `cultcache_py`; `.path` names the path) before the lock and the replace, for the single-file store and for `JsonLinesBackingStore` (`push`, `push_all`, `delete`), and nothing is written. Before, a write at a live link replaced the link with a regular file, and at a dangling link both stores raised `FileNotFoundError`. A caller catching `FileNotFoundError` there now catches `StoreLinkedError` (`OSError`); configure the link's target as the store path. On Windows a junction or any reparse point counts as a link. Reads through a live link work. - A record whose payload carries a slot past its reader's declared slots, at any depth (the document, a nested record, a list element, a union arm, a unit arm), is read, held read-only, and every write or removal of its key (`put`, `put_global`, `update`, `put_envelope`, `put_envelopes`, `delete`, `delete_global`) is refused with `SchemaConflictError` before any store is touched (`cultlib-gaps:ruling:extra-trailing-slots`). Reading dropped those slots silently and the next write shed them. `SchemaConflictError.unread_slots` is an `UnreadSlots(type_name, slot_count, declared_count, occurrences, depth0_only)` and the message names the type and both counts, never a slot value. A depth-0 mark is consented, and the record writable, only when the record is stored under an id the document lists in `compatible_schema_ids` and that id's catalog entry has slots past the document's, covering the mark's slot count (`cultlib-gaps:ruling:slot-mark-consent`); a nested mark, or a mark under the document's own id, keeps the record read-only whatever is declared. Missing trailing slots keep their declared defaults. An untouched marked record keeps its stored bytes through every other write. Added `CultCache.is_writable(document, key)` (advisory; the write still decides: it also answers False for a foreign-carried key, for a key held as a type the document does not own, and for a global document at a non-global key; `delete` asks the same judges, so removing a key held as another type is refused with `SchemaConflictError` unless the document owns that record's schema id (its own or one it lists compatible), in which case the removal removes that record, its file row and its lookups, as a write of the key would replace it; it returned silently and removed nothing), `DocumentDefinition.decode_payload_marked`, and `UnreadSlots`. A document with a custom `decode` (no slot codec) never marks. A store holding a variant record is still unreadable here. Implies a minor bump while the version is 0.x (0.3.0 to 0.4.0): code that wrote back a record from a newer writer now gets `SchemaConflictError`. diff --git a/packages/cultcache-py/README.md b/packages/cultcache-py/README.md index 4a7236fd..8733682f 100644 --- a/packages/cultcache-py/README.md +++ b/packages/cultcache-py/README.md @@ -94,8 +94,7 @@ store. - `put(...)` - `put_envelopes(...)` - `put_global(...)` -- `update(...)` -- `update_global(...)` +- `update(document, key, apply, *, max_attempts=5)` and `update_global(document, apply, *, max_attempts=5)`: the conditional golden path; reads the record as its home store holds it now, runs `apply` on a fresh copy, writes only if the record is unchanged and runs `apply` again on a lost race; returns `UpdateResult(outcome, document)` with `UpdateOutcome.UPDATED`, `DECLINED` (`apply` returned `None`) or `LOST_RACE` (see below) - `delete(...)` - `delete_global(...)` - `is_writable(document, key)` (advisory: would a write or removal of the key be refused; see below) @@ -143,6 +142,8 @@ used by the TypeScript, Rust, and C# runtimes: canonicalSchemaJson, compatibleSchemaIds, members]` - records: `[key, schemaId, storedAt, payload]` +`update` needs a home store with the optional `read(key)` (the record at the key, of any type, and its content version, from one read) and `commit_if(envelope, expected)` (write only if the record at the key is still that version, under the store's write lock). The content version is the lowercase hex SHA-256 of the record as the store writes it, `[key, schemaId, storedAt, payload]`, the same value C# computes (`tests/vectors/content-version`). The single-file store has both; `JsonLinesBackingStore` and a store without them refuse `update` before `apply` runs. A cache with no store copies its held instance. Every write that replaces a record is stored at a later `stored_at` than the one it replaces, in each store kind and in a store-less cache; `push` and `push_all` return the stamp written, and the cache holds that stamp. + Large corpora should use a sharded store or a real database. A single snapshot file is a scalpel, not a forklift. ## Store format and rollback diff --git a/packages/cultcache-py/src/cultcache_py/cache.py b/packages/cultcache-py/src/cultcache_py/cache.py index ebf9b2d2..0a396f61 100644 --- a/packages/cultcache-py/src/cultcache_py/cache.py +++ b/packages/cultcache-py/src/cultcache_py/cache.py @@ -663,6 +663,7 @@ class CultCache: self._state.names[(type, keys.name)] = envelope.key for index, index_value in keys.indexes.items(): self._state.indexes[(type, index, index_value)] = envelope.key + return "committed" def _refuse_retype(self, document: DocumentDefinition[Any], written: list[CultCacheEnvelope]) -> list[tuple[str, str]]: """The one key-identity judge (variants:ruling:cache-refuses-retype): a key holds one record whatever its type, so a write diff --git a/packages/cultcache-py/tests/test_first_call.py b/packages/cultcache-py/tests/test_first_call.py index 10d6eb27..232f6c7d 100644 --- a/packages/cultcache-py/tests/test_first_call.py +++ b/packages/cultcache-py/tests/test_first_call.py @@ -4,6 +4,8 @@ import unittest from pathlib import Path +from cultcache_py import UpdateOutcome + # snippet:begin import tempfile from dataclasses import asdict, dataclass @@ -51,15 +53,44 @@ def first_call() -> Item: # snippet:end +# snippet:update:begin +def restock(path: Path): + cache = ( + CultCache.builder() + .register_document_type(item_doc) + .add_generic_store(SingleFileMessagePackBackingStore(path)) + .build() + ) + cache.pull_all_backing_stores() + return cache.update(item_doc, "item:potion", lambda item: Item("Potion", "Consumable", (item.value if item else 0) + 10)) +# snippet:update:end + + class FirstCallTests(unittest.TestCase): def test_first_call_round_trips_through_a_second_cache(self) -> None: self.assertEqual(first_call(), Item("Potion", "Consumable", 50)) - def test_docs_show_the_tested_snippet(self) -> None: + def test_update_with_retry_through_a_second_cache(self) -> None: + path = Path(tempfile.mkdtemp()) / "restock.cc" + restock(path) + result = restock(path) + self.assertIs(result.outcome, UpdateOutcome.UPDATED) + reopened = ( + CultCache.builder() + .register_document_type(item_doc) + .add_generic_store(SingleFileMessagePackBackingStore(path)) + .build() + ) + reopened.pull_all_backing_stores() + self.assertEqual(reopened.get_required(item_doc, "item:potion"), Item("Potion", "Consumable", 20)) + + def test_docs_show_the_tested_snippets(self) -> None: source = Path(__file__).read_text(encoding="utf-8") - snippet = source.split("# snippet:begin\n", 1)[1].split("# snippet:end\n", 1)[0] docs = (Path(__file__).parents[3] / "docs" / "start" / "README.md").read_text(encoding="utf-8") - self.assertIn("```python\n" + snippet + "```", docs) + for region in ("snippet", "snippet:update"): + with self.subTest(region=region): + snippet = source.split(f"# {region}:begin\n", 1)[1].split(f"# {region}:end\n", 1)[0] + self.assertIn("```python\n" + snippet + "```", docs) if __name__ == "__main__": diff --git a/packages/cultcache-py/tests/test_update.py b/packages/cultcache-py/tests/test_update.py new file mode 100644 index 00000000..7ccd5b52 --- /dev/null +++ b/packages/cultcache-py/tests/test_update.py @@ -0,0 +1,665 @@ +"""CultCache.update (variants:cut_spec:cut-update-with-retry-py): read the record as its home store holds it, apply a function, +commit only if the record is unchanged, run the function again on a lost race. Mirrors C# UpdateTests and the TS update tests.""" + +from __future__ import annotations + +import re +import shutil +import tempfile +import threading +import time +import unittest +from dataclasses import asdict, dataclass +from datetime import datetime, timezone +from pathlib import Path +from unittest import mock + +from cultcache_py import ( + CultCache, + CultCacheEnvelope, + JsonLinesBackingStore, + SchemaConflictError, + SingleFileMessagePackBackingStore, + UpdateOutcome, + UpdateResult, + database_entry_field, + define_database_entry_type, + define_document_type, + leaves as t, +) +from cultcache_py.cache import GLOBAL_KEY, CultCacheError +from cultcache_py.stores import _parse_stored_at, mint_stored_at, record_version, stored_at_over + + +@dataclass +class Item: + name: str = "n" + value: int = 0 + sku: str | None = None + + +def _doc(type: str, **extra): + return define_document_type(type, encode=asdict, decode=lambda raw: Item(**raw), **extra) + + +item_doc = _doc("item") +other_doc = _doc("other") +sku_doc = _doc("skuitem", indexes={"sku": "sku"}) +global_doc = _doc("gitem", global_document=True) + +@dataclass +class Wide: + name: str = "" + extra: str = "" + + +@dataclass +class Narrow: + name: str = "" + + +ALPHA = "2099-01-01T00:00:00.0000000Z" + + +class Fixture(unittest.TestCase): + def setUp(self) -> None: + self.dir = Path(tempfile.mkdtemp()) + self.addCleanup(shutil.rmtree, self.dir, ignore_errors=True) + self.path = self.dir / "state.cc" + + def cache(self, *documents, path: Path | None = None, store=True) -> CultCache: + builder = CultCache.builder() + for document in documents or (item_doc,): + builder.register_document_type(document) + if store: + builder.add_generic_store(SingleFileMessagePackBackingStore(path or self.path)) + cache = builder.build() + if store: + cache.pull_all_backing_stores() + return cache + + def stored(self, key: str = "k", document=item_doc): + """What a fresh cache reads at the key.""" + return self.cache(document).get(document, key) + + def bytes(self) -> bytes | None: + return self.path.read_bytes() if self.path.exists() else None + + +class Counting: + """An apply that records what it was given.""" + + def __init__(self, fn=None) -> None: + self.seen: list = [] + self.fn = fn or (lambda item, calls: Item("n", (item.value if item else 0) + 1)) + + def __call__(self, item): + self.seen.append(None if item is None else item.value) + # An always-losing apply must fail by assertion when the attempt bound is gone, not hang. + assert len(self.seen) <= 10, "apply ran past the attempt bound" + return self.fn(item, len(self.seen)) + + @property + def calls(self) -> int: + return len(self.seen) + + +class UpdateTests(Fixture): + def test_two_writers_updating_lose_no_increment(self) -> None: + results: list[UpdateResult] = [] + calls = [0] + lock = threading.Lock() + a, b = self.cache(), self.cache() + a.put(item_doc, "k", Item("n", 0)) + + def bump(item): + with lock: + calls[0] += 1 + time.sleep(0) + return Item("n", item.value + 1) + + def run(cache) -> None: + for _ in range(50): + results.append(cache.update(item_doc, "k", bump, max_attempts=200)) + + threads = [threading.Thread(target=run, args=(cache,)) for cache in (a, b)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + self.assertTrue(all(result.outcome is UpdateOutcome.UPDATED for result in results)) + self.assertEqual(self.stored().value, 100) + self.assertGreater(calls[0], 100) + + def test_a_lost_attempt_reruns_apply_on_the_fresh_value(self) -> None: + cache, rival = self.cache(), self.cache() + cache.put(item_doc, "k", Item("n", 1)) + rival.pull_all_backing_stores() + apply = Counting(lambda item, calls: (rival.put(item_doc, "k", Item("n", 100)) if calls == 1 else None) or Item("n", item.value + 1)) + result = cache.update(item_doc, "k", apply) + self.assertEqual(apply.seen, [1, 100]) + self.assertIs(result.outcome, UpdateOutcome.UPDATED) + self.assertEqual(result.document.value, 101) + self.assertEqual(self.stored().value, 101) + + def test_exhausted_attempts_return_lost_race(self) -> None: + cache, rival = self.cache(), self.cache() + cache.put(item_doc, "k", Item("n", 1)) + rival.pull_all_backing_stores() + apply = Counting(lambda item, calls: rival.put(item_doc, "k", Item("n", 100 * calls)) or Item("n", -1)) + result = cache.update(item_doc, "k", apply, max_attempts=3) + self.assertEqual(apply.calls, 3) + self.assertIs(result.outcome, UpdateOutcome.LOST_RACE) + self.assertEqual(result.document.value, 300) # the pull before LOST_RACE shows the winner + self.assertEqual(self.stored().value, 300) + + def test_max_attempts_bounds(self) -> None: + cache, rival = self.cache(), self.cache() + cache.put(item_doc, "k", Item("n", 1)) + rival.pull_all_backing_stores() + + def always_losing(item, calls): + rival.put(item_doc, "k", Item("n", 100 + calls)) + return Item("n", -1) + + apply = Counting(always_losing) + result = cache.update(item_doc, "k", apply, max_attempts=1) + self.assertEqual((apply.calls, result.outcome), (1, UpdateOutcome.LOST_RACE)) + for bad, error in ((0, ValueError), (-1, ValueError), (True, TypeError), (1.5, TypeError), (None, TypeError)): + with self.subTest(max_attempts=bad): + untouched = Counting() + with self.assertRaises(error): + cache.update(item_doc, "k", untouched, max_attempts=bad) + self.assertEqual(untouched.calls, 0) + with self.assertRaises(TypeError): + cache.update(item_doc, "k", "not callable") + with self.assertRaises(TypeError): + cache.update(item_doc, "k", None) + + def test_apply_gets_a_fresh_copy(self) -> None: + cache, rival = self.cache(), self.cache() + cache.put(item_doc, "k", Item("n", 1)) + rival.pull_all_backing_stores() + arguments: list[Item] = [] + + def apply(item): + arguments.append(item) + self.assertIsNot(item, cache.get(item_doc, "k")) + if len(arguments) == 1: + item.value = 777 # changing the copy in place changes nothing the cache or the next attempt sees + rival.put(item_doc, "k", Item("n", 5)) + self.assertEqual(cache.get(item_doc, "k").value, 1) + return Item("n", item.value + 1) + + result = cache.update(item_doc, "k", apply) + self.assertEqual([item.value for item in arguments], [777, 5]) + self.assertIsNot(arguments[0], arguments[1]) + self.assertEqual(result.document.value, 6) + + def test_an_in_place_change_to_the_held_value_is_neither_applied_nor_persisted(self) -> None: + cache = self.cache() + cache.put(item_doc, "k", Item("n", 1)) + cache.get(item_doc, "k").value = 999 + apply = Counting(lambda item, calls: Item("n", item.value + 1)) + result = cache.update(item_doc, "k", apply) + self.assertEqual(apply.seen, [1]) + self.assertEqual(self.stored().value, 2) + result.document.value = 999 + cache.update(item_doc, "k", apply) + self.assertEqual(apply.seen, [1, 2]) + self.assertEqual(self.stored().value, 3) + + bare = self.cache(store=False) # a cache with no store copies its held instance + bare.put(item_doc, "k", Item("n", 1)) + bare.get(item_doc, "k").value = 999 + seen = Counting(lambda item, calls: Item("n", item.value + 1)) + self.assertEqual(bare.update(item_doc, "k", seen).document.value, 2) + self.assertEqual(seen.seen, [1]) + + def test_none_declines(self) -> None: + cache = self.cache() + cache.put(item_doc, "k", Item("n", 1)) + before = self.bytes() + result = cache.update(item_doc, "k", lambda item: None) + self.assertIs(result.outcome, UpdateOutcome.DECLINED) + self.assertEqual(result.document, cache.get(item_doc, "k")) + self.assertEqual(self.bytes(), before) + absent = cache.update(item_doc, "missing", lambda item: None) + self.assertEqual((absent.outcome, absent.document), (UpdateOutcome.DECLINED, None)) + self.assertEqual(self.bytes(), before) + empty = Fixture.cache(self, path=self.dir / "empty.cc") + self.assertEqual(empty.update(item_doc, "k", lambda item: None).outcome, UpdateOutcome.DECLINED) + self.assertFalse((self.dir / "empty.cc").exists()) + + def test_a_declined_or_lost_attempt_installs_nothing(self) -> None: + cache, rival = self.cache(), self.cache() + cache.put(item_doc, "k", Item("n", 1)) + rival.pull_all_backing_stores() + rival.put(item_doc, "k", Item("n", 5)) + self.assertIs(cache.update(item_doc, "k", lambda item: None).outcome, UpdateOutcome.DECLINED) + self.assertEqual(cache.get(item_doc, "k").value, 1) + cache.pull_all_backing_stores() + self.assertEqual(cache.get(item_doc, "k").value, 5) + + during: list[int] = [] + + def apply(item, calls): + during.append(cache.get(item_doc, "k").value) + if calls == 1: + rival.put(item_doc, "k", Item("n", 9)) + return Item("n", item.value + 1) + + cache.update(item_doc, "k", Counting(apply)) + self.assertEqual(during, [5, 5]) # the lost attempt published nothing mid-update + + def test_absent_key_rival_removal_and_a_store_that_holds_nothing(self) -> None: + a, b = self.cache(), self.cache() + bump = lambda item: Item("n", (item.value if item else 0) + 1) # noqa: E731 + self.assertIs(a.update(item_doc, "k", bump).outcome, UpdateOutcome.UPDATED) + self.assertIs(b.update(item_doc, "k", bump).outcome, UpdateOutcome.UPDATED) + self.assertEqual(self.stored().value, 2) + + cache, rival = self.cache(), self.cache() + cache.put(item_doc, "d", Item("n", 1)) + rival.pull_all_backing_stores() + apply = Counting(lambda item, calls: (rival.delete(item_doc, "d") if calls == 1 else None) or Item("n", (item.value if item else 0) + 1)) + cache.update(item_doc, "d", apply) + self.assertEqual(apply.seen, [1, None]) + self.assertEqual(self.stored("d").value, 1) + + holder = self.cache() + holder.put(item_doc, "h", Item("n", 3)) + rival.pull_all_backing_stores() + rival.delete(item_doc, "h") + nothing = Counting(lambda item, calls: Item("n", 1)) + holder.update(item_doc, "h", nothing) + self.assertEqual(nothing.seen, [None]) + + def test_a_rival_writing_and_restoring_the_value_still_reruns_apply(self) -> None: + cache, rival = self.cache(), self.cache() + cache.put(item_doc, "k", Item("n", 1)) + rival.pull_all_backing_stores() + + def apply(item, calls): + if calls == 1: + rival.put(item_doc, "k", Item("n", 5)) + rival.put(item_doc, "k", Item("n", 1)) + return Item("n", item.value + 1) + + counting = Counting(apply) + cache.update(item_doc, "k", counting) + self.assertEqual(counting.seen, [1, 1]) + self.assertEqual(self.stored().value, 2) + + def test_a_stale_caches_first_attempt_reads_the_store(self) -> None: + a, b = self.cache(), self.cache() + a.put(item_doc, "k", Item("n", 1)) + b.pull_all_backing_stores() + a.put(item_doc, "k", Item("n", 7)) + apply = Counting(lambda item, calls: Item("n", item.value + 1)) + b.update(item_doc, "k", apply) + self.assertEqual(apply.seen, [7]) + self.assertEqual(self.stored().value, 8) + + +class UpdateRefusalTests(Fixture): + def refused(self, cache: CultCache, key: str = "k", document=item_doc, *errors) -> None: + before = self.bytes() + apply = Counting() + with self.assertRaises(errors or (CultCacheError, SchemaConflictError)): + cache.update(document, key, apply) + self.assertEqual(apply.calls, 0) + self.assertEqual(self.bytes(), before) + + def test_refusals_before_apply(self) -> None: + writer = self.cache(item_doc, other_doc) + writer.put(other_doc, "k", Item("n", 1)) + self.refused(writer) # the record at the key is another document's, held + self.refused(self.fresh(item_doc, other_doc)) # unpulled: the store's record is another document's + self.refused(self.cache(item_doc)) # foreign: the store holds a type this cache never registered + self.refused(self.fresh(item_doc)) + storeless = self.cache(item_doc, other_doc, store=False) + storeless.put(other_doc, "k", Item("n", 1)) + apply = Counting() + with self.assertRaises((CultCacheError, SchemaConflictError)): + storeless.update(item_doc, "k", apply) + self.assertEqual(apply.calls, 0) + + def fresh(self, *documents) -> CultCache: + """A cache that has not pulled: whatever it reads, it reads from the store at update.""" + builder = CultCache.builder() + for document in documents: + builder.register_document_type(document) + return builder.add_generic_store(SingleFileMessagePackBackingStore(self.path)).build() + + def test_a_home_without_read_and_commit_if_refuses(self) -> None: + jsonl = JsonLinesBackingStore(self.dir / "state.jsonl") + cache = CultCache.builder().register_document_type(item_doc).add_generic_store(jsonl).build() + cache.pull_all_backing_stores() + cache.put(item_doc, "k", Item("n", 1)) + before = (self.dir / "state.jsonl").read_bytes() + apply = Counting() + with self.assertRaisesRegex(CultCacheError, "cannot write conditionally"): + cache.update(item_doc, "k", apply) + self.assertEqual(apply.calls, 0) + self.assertEqual((self.dir / "state.jsonl").read_bytes(), before) + self.assertFalse(hasattr(JsonLinesBackingStore, "read") or hasattr(JsonLinesBackingStore, "commit_if")) + + class Bare: + def pull_all(self): + return [] + + def push(self, envelope): + raise AssertionError("update wrote through a store that cannot write conditionally") + + def push_all(self, envelopes, soft=False): + raise AssertionError("push_all") + + def delete(self, type, key): + raise AssertionError("delete") + + bare = CultCache.builder().register_document_type(item_doc).add_generic_store(Bare()).build() + with self.assertRaisesRegex(CultCacheError, "Bare"): + bare.update(item_doc, "k", apply) + self.assertEqual(apply.calls, 0) + + def test_read_only_records_refuse_before_apply(self) -> None: + wide = define_database_entry_type( + "tests.rw", + [database_entry_field("name", 0, codec=t.string), database_entry_field("extra", 1, codec=t.string)], + cls=Wide, + schema_id="tests.rw.v1", + ) + narrow = define_database_entry_type("tests.rw", [database_entry_field("name", 0, codec=t.string)], cls=Narrow, schema_id="tests.rw.v1") + self.cache(wide).put(wide, "k", Wide("n", "x")) + before = self.bytes() + for pulled in (True, False): # held read-only, then a store record the cache never loaded + with self.subTest(pulled=pulled): + cache = self.cache(narrow) if pulled else self.fresh(narrow) + apply = Counting() + with self.assertRaises(SchemaConflictError) as caught: + cache.update(narrow, "k", apply) + self.assertEqual(caught.exception.record_key, "k") + self.assertEqual(apply.calls, 0) + self.assertEqual(self.bytes(), before) + + def test_a_non_global_key_for_a_global_document_refuses(self) -> None: + cache = self.cache(global_doc) + apply = Counting() + with self.assertRaises(CultCacheError): + cache.update(global_doc, "not-global", apply) + self.assertEqual(apply.calls, 0) + + def test_refusals_after_apply_are_not_retried(self) -> None: + cache = self.cache() + cache.put(item_doc, "k", Item("n", 1)) + before = self.bytes() + boom = Counting(lambda item, calls: (_ for _ in ()).throw(RuntimeError("boom"))) + with self.assertRaisesRegex(RuntimeError, "boom"): + cache.update(item_doc, "k", boom) + self.assertEqual(boom.calls, 1) + self.assertEqual(self.bytes(), before) + + indexed = self.cache(sku_doc, path=self.dir / "sku.cc") + indexed.put(sku_doc, "a", Item("n", 1, "X")) + indexed.put(sku_doc, "b", Item("n", 1, "Y")) + before = (self.dir / "sku.cc").read_bytes() + clash = Counting(lambda item, calls: Item("n", 2, "X")) + with self.assertRaisesRegex(CultCacheError, "unique index"): + indexed.update(sku_doc, "b", clash) + self.assertEqual(clash.calls, 1) + self.assertEqual((self.dir / "sku.cc").read_bytes(), before) + + +class UpdateOwnWritesTests(Fixture): + def test_own_writes_and_the_held_instance(self) -> None: + cache = self.cache() + cache.put(item_doc, "k", Item("n", 1)) + apply = Counting(lambda item, calls: Item("n", item.value + 1)) + cache.update(item_doc, "k", apply) + second = cache.update(item_doc, "k", apply) + self.assertEqual(apply.calls, 2) + self.assertIs(second.document, cache.get(item_doc, "k")) + self.assertEqual(second.document.value, 3) + + def sideways(item, calls): + cache.put(item_doc, "side", Item("n", 5)) # apply runs outside the cache's hold + return Item("n", item.value + 1) + + sideways_apply = Counting(sideways) + result = cache.update(item_doc, "k", sideways_apply) + self.assertEqual((result.outcome, sideways_apply.calls), (UpdateOutcome.UPDATED, 1)) + self.assertEqual(self.stored("side").value, 5) + + def test_store_less_cache(self) -> None: + cache = self.cache(store=False) + cache.put(item_doc, "k", Item("n", 0)) + results: list[UpdateResult] = [] + + def bump(item): + time.sleep(0) + return Item("n", item.value + 1) + + def run() -> None: + for _ in range(50): + results.append(cache.update(item_doc, "k", bump, max_attempts=500)) + + threads = [threading.Thread(target=run) for _ in range(2)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + self.assertTrue(all(result.outcome is UpdateOutcome.UPDATED for result in results)) + self.assertEqual(cache.get(item_doc, "k").value, 100) + + # A store that becomes the type's home while apply runs holds a record the cache never read: the commit loses. + cache = self.cache(store=False) # holds no record of the type, so a store may become its home + other = self.cache(path=self.dir / "late.cc") + other.put(item_doc, "fresh", Item("n", 40)) + late = SingleFileMessagePackBackingStore(self.dir / "late.cc") + + def attach(item, calls): + if calls == 1: + cache.add_generic_store(late) + return Item("n", (item.value if item else 0) + 1) + + apply = Counting(attach) + result = cache.update(item_doc, "fresh", apply) + self.assertEqual(apply.seen, [None, 40]) + self.assertIs(result.outcome, UpdateOutcome.UPDATED) + self.assertEqual(self.cache(path=self.dir / "late.cc").get(item_doc, "fresh").value, 41) + + def test_ledger_under_contention(self) -> None: + caches = [self.cache() for _ in range(3)] + outcomes: list[UpdateOutcome] = [] + + def run(cache) -> None: + for _ in range(25): + outcomes.append(cache.update(item_doc, "k", lambda item: Item("n", (item.value if item else 0) + 1)).outcome) + + threads = [threading.Thread(target=run, args=(cache,)) for cache in caches] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + self.assertEqual(self.stored().value, outcomes.count(UpdateOutcome.UPDATED)) + self.assertEqual(len(outcomes), 75) + + +class UpdateLegacyGlobalTests(Fixture): + def plant(self, key: str = "legacy") -> None: + entry = global_doc.catalog_entry() + SingleFileMessagePackBackingStore(self.path).push( + CultCacheEnvelope.create(key=key, type=global_doc.type, payload=global_doc.encode_payload(Item("n", 1)), schema_id=entry.schema_id, catalog_entry=entry) + ) + + def test_update_global_updates_the_global_record(self) -> None: + cache = self.cache(global_doc) + result = cache.update_global(global_doc, lambda item: Item("n", (item.value if item else 0) + 1)) + self.assertIs(result.outcome, UpdateOutcome.UPDATED) + self.assertEqual(self.cache(global_doc).get(global_doc, GLOBAL_KEY).value, 1) + + def test_a_legacy_global_is_refused_until_put_global_retires_it(self) -> None: + self.plant() + cache = self.cache(global_doc) # loaded from the legacy key + before = self.bytes() + for call in (lambda a: cache.update_global(global_doc, a), lambda a: cache.update(global_doc, GLOBAL_KEY, a)): + apply = Counting() + with self.assertRaisesRegex(CultCacheError, "put_global"): + call(apply) + self.assertEqual(apply.calls, 0) + self.assertEqual(self.bytes(), before) + # Judged from the cache's load, not only from the store: the legacy record is gone from the store, the cache still holds it. + SingleFileMessagePackBackingStore(self.path).delete(global_doc.type, "legacy") + apply = Counting() + with self.assertRaisesRegex(CultCacheError, "put_global"): + cache.update_global(global_doc, apply) + self.assertEqual(apply.calls, 0) + cache.put_global(global_doc, Item("n", 5)) + self.assertEqual(cache.update_global(global_doc, lambda item: Item("n", item.value + 1)).document.value, 6) + + def test_a_pull_inside_apply_that_loads_a_legacy_global_fails_the_commit(self) -> None: + cache = self.cache(global_doc) + + def apply(item): + self.plant() + cache.pull_all_backing_stores() + return Item("n", 9) + + with self.assertRaisesRegex(CultCacheError, "put_global"): + cache.update_global(global_doc, apply) + self.assertEqual([envelope.key for envelope in SingleFileMessagePackBackingStore(self.path).pull_all()], ["legacy"]) + + def test_an_unpulled_cache_judges_the_legacy_global_from_the_store(self) -> None: + self.plant() + before = self.bytes() + cache = SingleFileMessagePackBackingStore(self.path) + unpulled = CultCache.builder().register_document_type(global_doc).add_generic_store(cache).build() + apply = Counting() + with self.assertRaisesRegex(CultCacheError, "put_global"): + unpulled.update_global(global_doc, apply) + self.assertEqual(apply.calls, 0) + self.assertEqual(self.bytes(), before) + + +class StoreRestampTests(Fixture): + def envelope(self, value: int, stored_at: str = ALPHA, key: str = "k") -> CultCacheEnvelope: + entry = item_doc.catalog_entry() + return CultCacheEnvelope(key, "item", item_doc.encode_payload(Item("n", value)), stored_at, entry.schema_id, entry) + + def stamp_on_disk(self, store, key: str = "k") -> str: + return next(envelope.stored_at for envelope in store.pull_all() if envelope.key == key) + + def later_than_alpha(self, text: str) -> bool: + return _parse_stored_at(text) > _parse_stored_at(ALPHA) + + def test_commit_if_and_the_restamp_at_the_store(self) -> None: + store = SingleFileMessagePackBackingStore(self.path) + self.assertIsNone(store.read("k")) + stamp = store.commit_if(self.envelope(1), None) # None against an absent key commits + self.assertEqual(stamp, ALPHA) + envelope, version = store.read("k") + self.assertEqual(version, record_version(envelope)) + before = self.bytes() + self.assertIsNone(store.commit_if(self.envelope(2, "2100-01-01T00:00:00Z"), None)) # a record is there: not absent + self.assertIsNone(store.commit_if(self.envelope(2, "2100-01-01T00:00:00Z"), "0" * 64)) + self.assertEqual(self.bytes(), before) + written = store.commit_if(self.envelope(2), version) # a staged stamp that is no later is minted over + self.assertTrue(self.later_than_alpha(written)) + self.assertEqual(self.stamp_on_disk(store), written) + self.assertEqual(stamp_of(store, "k"), written) + + def test_every_replacing_write_is_stored_strictly_later(self) -> None: + for make in (SingleFileMessagePackBackingStore, JsonLinesBackingStore): + for name in ("push", "push_all", "commit_if"): + if name == "commit_if" and make is JsonLinesBackingStore: + continue + with self.subTest(store=make.__name__, write=name): + path = self.dir / f"{make.__name__}-{name}" + store = make(path) + store.push(self.envelope(1)) + if name == "push": + reported = store.push(self.envelope(2)) + elif name == "push_all": + reported = store.push_all([self.envelope(2)])["k"] + else: + reported = store.commit_if(self.envelope(2), store.read("k")[1]) + self.assertTrue(self.later_than_alpha(reported)) + self.assertEqual(self.stamp_on_disk(store), reported) + + def test_the_cache_holds_the_stamp_the_store_wrote(self) -> None: + for name in ("put", "put_envelope", "put_envelopes", "update"): + with self.subTest(write=name): + path = self.dir / name + SingleFileMessagePackBackingStore(path).push(self.envelope(1)) + cache = self.cache(path=path) + self.assertEqual(cache.get_envelope(item_doc, "k").stored_at, ALPHA) + if name == "put": + cache.put(item_doc, "k", Item("n", 2)) + elif name == "put_envelope": + cache.put_envelope(item_doc, self.envelope(2)) + elif name == "put_envelopes": + cache.put_envelopes(item_doc, [self.envelope(2)]) + else: + cache.update(item_doc, "k", lambda item: Item("n", 2)) + held = cache.get_envelope(item_doc, "k").stored_at + self.assertTrue(self.later_than_alpha(held)) + self.assertEqual(held, self.stamp_on_disk(SingleFileMessagePackBackingStore(path))) + + def test_a_store_less_cache_stamps_a_replacing_write_later(self) -> None: + cache = self.cache(store=False) + cache.put_envelope(item_doc, self.envelope(1)) + cache.put_envelope(item_doc, self.envelope(2, "2000-01-01T00:00:00Z")) + self.assertTrue(self.later_than_alpha(cache.get_envelope(item_doc, "k").stored_at)) + + def test_stored_at_parses_on_a_python_that_refuses_z_and_seven_digits(self) -> None: + """Python 3.10's fromisoformat takes neither (requires-python is >=3.10): the parse owns both, so it is pinned against a strict + reader on any interpreter.""" + + class Strict(datetime): + @classmethod + def fromisoformat(cls, text): + if text.endswith("Z") or re.search(r"\.\d{4,5}\b|\.\d{7,}", text): + raise ValueError(text) + return datetime.fromisoformat(text) + + with mock.patch("cultcache_py.stores.datetime", Strict): + for text in ("2026-01-01T00:00:00.1234567+00:00", "2026-01-01T00:00:00Z", "2026-01-01T00:00:00.1234567Z"): + with self.subTest(text=text): + parsed = _parse_stored_at(text) + self.assertIsNotNone(parsed) + self.assertEqual(parsed.microsecond, 0 if text.endswith("00Z") else 123456) + + def test_stamp_parsing_and_install(self) -> None: + for text in ("2026-01-01T00:00:00.1234567+00:00", "2026-01-01T00:00:00Z", "2026-01-01T00:00:00.123456+00:00", "2026-01-01T00:00:00"): + with self.subTest(text=text): + self.assertIsNotNone(_parse_stored_at(text)) + self.assertEqual(_parse_stored_at(text).tzinfo, timezone.utc) + self.assertIsNone(_parse_stored_at("garbage")) + self.assertIsNone(_parse_stored_at(None)) + now = datetime(2030, 1, 1, tzinfo=timezone.utc) + self.assertEqual(mint_stored_at("garbage", now), "2030-01-01T00:00:00.000000+00:00") + self.assertEqual(stored_at_over("2000-01-01T00:00:00Z", "2000-01-01T00:00:00Z")[:4], str(datetime.now(timezone.utc).year)) # no later: minted at now + self.assertEqual(mint_stored_at("2030-01-01T00:00:00Z", now), "2030-01-01T00:00:00.000001+00:00") # equal is not later + self.assertEqual(mint_stored_at("2099-01-01T00:00:00.1234567Z", now), "2099-01-01T00:00:00.123457+00:00") + self.assertEqual(stored_at_over("2000-01-01T00:00:00Z", "2099-01-01T00:00:00.1234567Z"), "2099-01-01T00:00:00.123457+00:00") + + for reported in (True, 1, "garbage"): + with self.subTest(reported=reported): + class Lying(SingleFileMessagePackBackingStore): + def push(self, envelope): + super().push(envelope) + return reported + + cache = CultCache.builder().register_document_type(item_doc).add_generic_store(Lying(self.dir / f"lying-{reported}")).build() + cache.pull_all_backing_stores() + cache.put_envelope(item_doc, self.envelope(1, "2026-02-02T00:00:00.000000+00:00")) + self.assertEqual(cache.get_envelope(item_doc, "k").stored_at, "2026-02-02T00:00:00.000000+00:00") + + +def stamp_of(store, key: str) -> str: + return store.read(key)[0].stored_at + + +if __name__ == "__main__": + unittest.main() -- 2.47.3 From dc8919fde20061d5dea0512eeb9373fd75e75934 Mon Sep 17 00:00:00 2001 From: Libby Date: Fri, 13 Nov 2026 20:24:49 +0000 Subject: [PATCH 3/6] cultcache-py: strict stored_at shape, range edges refuse, mesh publishes the stored stamp, home-swap guard pinned (update-with-retry-py s1 fixes) Co-Authored-By: Claude Sonnet 5.5 --- packages/cultcache-py/CHANGELOG.md | 4 +- packages/cultcache-py/README.md | 2 +- .../cultcache-py/src/cultcache_py/stores.py | 40 ++++--- packages/cultcache-py/tests/test_update.py | 101 ++++++++++++++---- packages/cultmesh-py/src/cultmesh_py/node.py | 4 +- packages/cultmesh-py/tests/test_cultmesh.py | 26 +++++ 6 files changed, 142 insertions(+), 35 deletions(-) diff --git a/packages/cultcache-py/CHANGELOG.md b/packages/cultcache-py/CHANGELOG.md index a0d5cbe8..8f3e481d 100644 --- a/packages/cultcache-py/CHANGELOG.md +++ b/packages/cultcache-py/CHANGELOG.md @@ -4,12 +4,12 @@ ### Added -- `CultCache.update` and `update_global` return `UpdateResult(outcome, document)`; `UpdateOutcome` is `UPDATED`, `DECLINED` or `LOST_RACE`. The single-file store gains `read(key)` (the record and its content version, the SHA-256 C# computes) and `commit_if(envelope, expected)`, both looked up by the cache and not `BackingStore` protocol members (variants:cut_spec:cut-update-with-retry-py). +- `CultCache.update` and `update_global` return `UpdateResult(outcome, document)`; `UpdateOutcome` is `UPDATED`, `DECLINED` or `LOST_RACE`. The single-file store gains `read(key)` (the record and its content version, the SHA-256 C# computes) and `commit_if(envelope, expected)`, the version is the SHA-256 of the record re-encoded as `[key, schemaId, storedAt, payload]` with minimal MessagePack encodings, which is the SHA-256 of the record's stored bytes exactly when the writer encoded it minimally (every writer in this library does; a record another writer encoded with a wider-than-needed length or integer has the version of its minimal form, not of its bytes). Both looked up by the cache and not `BackingStore` protocol members (variants:cut_spec:cut-update-with-retry-py). - `read_payload_tree(type_name, payload, decoder=None)`: the exported owner of the one-value, depth-first payload read that `decode_payload` uses, for payloads read outside a definition (`cultnet-py`'s witness bundle decode needs it, so `cultnet-py` now requires `cultcache-py>=0.4.0`). The manifest version is 0.4.0, the minor these Unreleased breaking changes imply, set in this change so the workspace install resolves the new bound; the `## Unreleased` heading becomes `## [0.4.0]` when it is tagged. ### Changed -- A write that replaces a record is stored at a later `stored_at` than the record it replaces, in the single-file store, the JSONL store and a store-less cache, with no exception for identical content (`variants:ruling:py-rs-update-follow-csharp-restamp`); the cache holds the stamp the store wrote. +- A write that replaces a record is stored at a later `stored_at` than the record it replaces, in the single-file store, the JSONL store and a store-less cache, with no exception for identical content (`variants:ruling:py-rs-update-follow-csharp-restamp`); the cache holds the stamp the store wrote. A stored_at is read only in the shape the runtimes write (an RFC 3339 date-time with T, a fraction of 1 to 9 digits, Z or a numeric offset), the same on every Python; any other text, or an instant outside `datetime`'s range, is not a stamp and a write over it is stamped at now. A write over a record stamped at the last instant `datetime` can hold (9999-12-31T23:59:59.999999Z) is refused with `ValueError` and writes nothing. - `store_write_lock` looks for a link at the store path again once the lock is held, so a link put at the path while a writer waited on the lock is refused with `StoreLinkedError` and the lock released, instead of being replaced. `JsonLinesBackingStore` takes no cross-process lock (only its thread lock, under which it already refuses a link), so it has no lock wait to re-check across. Pinned by a test that puts the link while a foreign process holds the lock. - `CultCache.is_writable` does not look at store paths: a link at a store path is not predicted, and the write refuses it with `StoreLinkedError`. Pinned by a test through a cache attached to a linked store. - `JsonLinesBackingStore` writes LF line endings on Windows. It wrote CRLF there, through `write_text` newline translation, before the commit_store_file change (#154, adbbb780). diff --git a/packages/cultcache-py/README.md b/packages/cultcache-py/README.md index 8733682f..0a03a671 100644 --- a/packages/cultcache-py/README.md +++ b/packages/cultcache-py/README.md @@ -142,7 +142,7 @@ used by the TypeScript, Rust, and C# runtimes: canonicalSchemaJson, compatibleSchemaIds, members]` - records: `[key, schemaId, storedAt, payload]` -`update` needs a home store with the optional `read(key)` (the record at the key, of any type, and its content version, from one read) and `commit_if(envelope, expected)` (write only if the record at the key is still that version, under the store's write lock). The content version is the lowercase hex SHA-256 of the record as the store writes it, `[key, schemaId, storedAt, payload]`, the same value C# computes (`tests/vectors/content-version`). The single-file store has both; `JsonLinesBackingStore` and a store without them refuse `update` before `apply` runs. A cache with no store copies its held instance. Every write that replaces a record is stored at a later `stored_at` than the one it replaces, in each store kind and in a store-less cache; `push` and `push_all` return the stamp written, and the cache holds that stamp. +`update` needs a home store with the optional `read(key)` (the record at the key, of any type, and its content version, from one read) and `commit_if(envelope, expected)` (write only if the record at the key is still that version, under the store's write lock). The content version is the lowercase hex SHA-256 of the record re-encoded as the store writes it, `[key, schemaId, storedAt, payload]` with minimal MessagePack encodings, the same value C# computes (`tests/vectors/content-version`). It equals the SHA-256 of the record's bytes in the file exactly when the writer encoded the record minimally, as every writer in this library does; a record encoded with a wider-than-needed length or integer has the version of its minimal form. The single-file store has both; `JsonLinesBackingStore` and a store without them refuse `update` before `apply` runs. A cache with no store copies its held instance. Every write that replaces a record is stored at a later `stored_at` than the one it replaces, in each store kind and in a store-less cache; `push` and `push_all` return the stamp written, and the cache holds that stamp. Large corpora should use a sharded store or a real database. A single snapshot file is a scalpel, not a forklift. diff --git a/packages/cultcache-py/src/cultcache_py/stores.py b/packages/cultcache-py/src/cultcache_py/stores.py index c7211640..c2c3352f 100644 --- a/packages/cultcache-py/src/cultcache_py/stores.py +++ b/packages/cultcache-py/src/cultcache_py/stores.py @@ -775,29 +775,45 @@ def _schema_id_for(envelope: CultCacheEnvelope) -> str: return envelope.schema_id or envelope.type +_STORED_AT_SHAPE = re.compile( + r"([0-9]{4})-([0-9]{2})-([0-9]{2})T([0-9]{2}):([0-9]{2}):([0-9]{2})(?:\.([0-9]{1,9}))?(?:(Z)|([+-])([0-9]{2}):([0-9]{2}))" +) + + def _parse_stored_at(text: Any) -> datetime | None: - """The instant a stored_at text names, or None: a trailing Z is UTC, fraction digits past the sixth are cut (C# writes - seven), a text with no offset is read as UTC. Written to read on Python 3.10, whose fromisoformat takes neither.""" + """The UTC instant a stored_at text names, or None. Only the shape the runtimes write is a stamp: an RFC 3339 date-time with T, + an optional fraction of 1 to 9 digits (cut to the microsecond) and Z or a numeric offset. Checked here, not left to + fromisoformat, whose accepted set differs by interpreter (3.10 and 3.13) and from Rust and TypeScript. An instant outside + datetime's range, or an offset of a day or more, is not a stamp.""" if not isinstance(text, str): return None - body = text.strip() - if body.endswith(("Z", "z")): - body = body[:-1] + "+00:00" - body = re.sub(r"(\.\d{6})\d+", r"\1", body) - try: - parsed = datetime.fromisoformat(body) - except ValueError: + shape = _STORED_AT_SHAPE.fullmatch(text) + if shape is None: + return None + year, month, day, hour, minute, second, fraction, zulu, sign, offset_hours, offset_minutes = shape.groups() + try: + offset = timezone.utc if zulu else timezone( + (-1 if sign == "-" else 1) * timedelta(hours=int(offset_hours), minutes=int(offset_minutes)) + ) + local = datetime( + int(year), int(month), int(day), int(hour), int(minute), int(second), int(((fraction or "") + "000000")[:6]), offset + ) + return local.astimezone(timezone.utc) + except (ValueError, OverflowError): return None - return parsed.replace(tzinfo=timezone.utc) if parsed.tzinfo is None else parsed.astimezone(timezone.utc) def mint_stored_at(previous: str | None, now: datetime) -> str: """A stored_at later than `previous` (C# MintStoredAt): `now`, or the microsecond after `previous` when `now` is not later. - A previous that does not parse leaves `now`.""" + A previous that does not parse leaves `now`. Raises ValueError, nothing minted, when `previous` is the last instant a stamp + can name, so a write over it is refused and writes nothing.""" now = now.astimezone(timezone.utc) last = _parse_stored_at(previous) if last is not None and now <= last: - now = last + timedelta(microseconds=1) + try: + now = last + timedelta(microseconds=1) + except OverflowError: + raise ValueError(f"stored_at {previous!r} is the last instant a stamp can name; a write over it cannot be stored later") from None return now.isoformat(timespec="microseconds") diff --git a/packages/cultcache-py/tests/test_update.py b/packages/cultcache-py/tests/test_update.py index 7ccd5b52..78725ce3 100644 --- a/packages/cultcache-py/tests/test_update.py +++ b/packages/cultcache-py/tests/test_update.py @@ -3,7 +3,6 @@ commit only if the record is unchanged, run the function again on a lost race. M from __future__ import annotations -import re import shutil import tempfile import threading @@ -12,7 +11,6 @@ import unittest from dataclasses import asdict, dataclass from datetime import datetime, timezone from pathlib import Path -from unittest import mock from cultcache_py import ( CultCache, @@ -469,6 +467,35 @@ class UpdateOwnWritesTests(Fixture): self.assertIs(result.outcome, UpdateOutcome.UPDATED) self.assertEqual(self.cache(path=self.dir / "late.cc").get(item_doc, "fresh").value, 41) + def test_a_home_attached_while_apply_runs_is_reread_or_refused(self) -> None: + """The cache's home for the type changes between the read and the commit (attaching is allowed while it holds no record + of the type): the version read from the old home is no condition on the new one.""" + cache = self.cache(path=self.dir / "generic.cc") + late = SingleFileMessagePackBackingStore(self.dir / "late.cc") + + def attach(item, calls): + if calls == 1: + cache.add_backing_store(late, ["item"]) + return Item("n", 1) + + apply = Counting(attach) + self.assertIs(cache.update(item_doc, "k", apply).outcome, UpdateOutcome.UPDATED) + self.assertEqual(apply.seen, [None, None]) # reread from the new home, not committed with the old home's version + self.assertEqual(self.cache(path=self.dir / "late.cc").get(item_doc, "k").value, 1) + + cache = self.cache(path=self.dir / "generic2.cc") + jsonl = JsonLinesBackingStore(self.dir / "late.jsonl") + + def attach_jsonl(item, calls): + cache.add_backing_store(jsonl, ["item"]) + return Item("n", 1) + + apply = Counting(attach_jsonl) + with self.assertRaisesRegex(CultCacheError, "cannot write conditionally"): + cache.update(item_doc, "k", apply) + self.assertEqual(apply.calls, 1) + self.assertFalse((self.dir / "late.jsonl").exists()) + def test_ledger_under_contention(self) -> None: caches = [self.cache() for _ in range(3)] outcomes: list[UpdateOutcome] = [] @@ -612,26 +639,62 @@ class StoreRestampTests(Fixture): cache.put_envelope(item_doc, self.envelope(2, "2000-01-01T00:00:00Z")) self.assertTrue(self.later_than_alpha(cache.get_envelope(item_doc, "k").stored_at)) - def test_stored_at_parses_on_a_python_that_refuses_z_and_seven_digits(self) -> None: - """Python 3.10's fromisoformat takes neither (requires-python is >=3.10): the parse owns both, so it is pinned against a strict - reader on any interpreter.""" + def test_only_the_stamp_shape_the_runtimes_write_parses(self) -> None: + """An RFC 3339 date-time with T, a fraction of 1 to 9 digits and Z or a numeric offset, on every interpreter (the shape is + checked here, not by fromisoformat): Rust and TypeScript read the same set.""" + utc = timezone.utc + accepted = { + "2026-01-01T00:00:00Z": datetime(2026, 1, 1, tzinfo=utc), + "2026-01-01T00:00:00.1Z": datetime(2026, 1, 1, 0, 0, 0, 100000, utc), + "2026-01-01T00:00:00.123456789Z": datetime(2026, 1, 1, 0, 0, 0, 123456, utc), + "2026-01-01T00:00:00.1234567+00:00": datetime(2026, 1, 1, 0, 0, 0, 123456, utc), + "2026-01-01T05:30:00+05:30": datetime(2026, 1, 1, tzinfo=utc), + "2025-12-31T19:00:00-05:00": datetime(2026, 1, 1, tzinfo=utc), + } + for text, instant in accepted.items(): + with self.subTest(text=text): + self.assertEqual(_parse_stored_at(text), instant) + for text in ( + "2026-01-01", "20260101T000000Z", "2026-W01-1T00:00:00Z", "2026-01-01 00:00:00Z", "2026-01-01T00:00:00,5Z", + "2026-01-01T00Z", "2026-01-01T00:00:00", "2026-01-01T00:00:00.Z", "2026-01-01T00:00:00.1234567890Z", + "2026-01-01T00:00:00z", " 2026-01-01T00:00:00Z", "2026-01-01T00:00:00Z ", "2026-01-01T24:00:00Z", + "2026-13-01T00:00:00Z", "2026-01-01T00:00:00+24:00", "2026-01-01T00:00:00+0000", "\u0662026-01-01T00:00:00Z", + "", "garbage", + ): + with self.subTest(text=text): + self.assertIsNone(_parse_stored_at(text)) - class Strict(datetime): - @classmethod - def fromisoformat(cls, text): - if text.endswith("Z") or re.search(r"\.\d{4,5}\b|\.\d{7,}", text): - raise ValueError(text) - return datetime.fromisoformat(text) - - with mock.patch("cultcache_py.stores.datetime", Strict): - for text in ("2026-01-01T00:00:00.1234567+00:00", "2026-01-01T00:00:00Z", "2026-01-01T00:00:00.1234567Z"): - with self.subTest(text=text): - parsed = _parse_stored_at(text) - self.assertIsNotNone(parsed) - self.assertEqual(parsed.microsecond, 0 if text.endswith("00Z") else 123456) + def test_a_stamp_outside_the_range_is_unparseable_and_one_at_the_edge_refuses_a_later_stamp(self) -> None: + for text in ("9999-12-31T23:30:00-05:00", "0001-01-01T00:00:00+05:00"): + with self.subTest(text=text): + self.assertIsNone(_parse_stored_at(text)) + now = datetime(2030, 1, 1, tzinfo=timezone.utc) + self.assertEqual(mint_stored_at("9999-12-31T23:30:00-05:00", now), "2030-01-01T00:00:00.000000+00:00") + edge = "9999-12-31T23:59:59.9999999Z" + with self.assertRaisesRegex(ValueError, "last instant"): + mint_stored_at(edge, now) + with self.assertRaisesRegex(ValueError, "last instant"): + stored_at_over("2000-01-01T00:00:00Z", edge) + # A write over a record stamped at the edge is refused by the store and writes nothing, from push, commit_if and update. + store = SingleFileMessagePackBackingStore(self.path) + entry = item_doc.catalog_entry() + record = lambda value, at: CultCacheEnvelope("k", "item", item_doc.encode_payload(Item("n", value)), at, entry.schema_id, entry) # noqa: E731 + store.push(record(1, edge)) + before = self.bytes() + version = store.read("k")[1] + for write in (lambda: store.push(record(2, ALPHA)), lambda: store.push_all([record(2, ALPHA)]), lambda: store.commit_if(record(2, ALPHA), version)): + with self.assertRaises(ValueError): + write() + cache = self.cache() + apply = Counting() + with self.assertRaises(ValueError): + cache.update(item_doc, "k", apply) + self.assertEqual(apply.calls, 1) + self.assertEqual(self.bytes(), before) + self.assertEqual(cache.get_envelope(item_doc, "k").stored_at, edge) def test_stamp_parsing_and_install(self) -> None: - for text in ("2026-01-01T00:00:00.1234567+00:00", "2026-01-01T00:00:00Z", "2026-01-01T00:00:00.123456+00:00", "2026-01-01T00:00:00"): + for text in ("2026-01-01T00:00:00.1234567+00:00", "2026-01-01T00:00:00Z", "2026-01-01T00:00:00.123456+00:00"): with self.subTest(text=text): self.assertIsNotNone(_parse_stored_at(text)) self.assertEqual(_parse_stored_at(text).tzinfo, timezone.utc) diff --git a/packages/cultmesh-py/src/cultmesh_py/node.py b/packages/cultmesh-py/src/cultmesh_py/node.py index 619aed36..41cd4aec 100644 --- a/packages/cultmesh-py/src/cultmesh_py/node.py +++ b/packages/cultmesh-py/src/cultmesh_py/node.py @@ -570,12 +570,14 @@ class CultMeshDatabase: catalog_entry=catalog_entry, ) value = self.cache.put_envelope(registered, envelope) + # The stamp the write was stored at, which a replacing write mints later than the record it replaces: the cache holds it. + stored_at = self.cache.get_required_envelope(registered, key).stored_at self._publish_local_change(registered, key, "added" if previous is None else "updated", value, previous) message = document_put_raw( message_id=message_id, key=key, schema_id=envelope.schema_id or catalog_entry.schema_id, - stored_at=envelope.stored_at, + stored_at=stored_at, payload=envelope.payload, source_runtime_id=self.runtime_id, shard_id=shard_id, diff --git a/packages/cultmesh-py/tests/test_cultmesh.py b/packages/cultmesh-py/tests/test_cultmesh.py index d96c4ab2..a1e0c35c 100644 --- a/packages/cultmesh-py/tests/test_cultmesh.py +++ b/packages/cultmesh-py/tests/test_cultmesh.py @@ -19,6 +19,7 @@ from typing import Any, Callable import msgpack from cultcache_py import ( CultCache, + CultCacheEnvelope, SingleFileMessagePackBackingStore, define_database_entry_type, define_document_type, @@ -796,6 +797,31 @@ class CultMeshTests(unittest.TestCase): self.assertEqual(node.get_required(document, "note:second").body, "second catalog publication") self.assertIsInstance(node.get_required(alias, "note:second"), UiNote) + def test_raw_put_publishes_and_logs_the_stamp_the_store_wrote(self) -> None: + document = define_database_entry_type("mesh.stamp_item", [("name", 0), ("category", 1), ("value", 2, 0)], cls=Item) + node = create_node(runtime_id="python-node") + node.register_document(document) + cache = node.database.cache + entry = document.catalog_entry() + future = "2099-01-01T00:00:00.000000+00:00" + cache.put_envelope( + document, + CultCacheEnvelope( + key="item:stamp", + type=document.type, + payload=document.encode_payload(Item("old", "gear", 1)), + stored_at=future, + schema_id=entry.schema_id, + catalog_entry=entry, + ), + ) + # A replacing write is stored later than the record it replaces, so the stamp is not the one the write was created with. + wire = node.put_raw_message(document, "item:stamp", Item("new", "gear", 2), shard_id="mesh", shard_epoch=1).to_wire() + held = cache.get_required_envelope(document, "item:stamp").stored_at + self.assertGreater(held, future) + self.assertEqual(wire["document"]["storedAt"], held) + self.assertEqual(node.database._shard_logs["mesh"][-1]["put"]["document"]["storedAt"], held) + def test_cultmesh_node_emits_raw_put_and_delete_messages_for_local_writes(self) -> None: document = define_database_entry_type( "mesh.emit_item", -- 2.47.3 From 14246ad45cd22fba8c2363675c767800a5331e37 Mon Sep 17 00:00:00 2001 From: Libby Date: Fri, 13 Nov 2026 20:32:59 +0000 Subject: [PATCH 4/6] cultmesh-py: replicated put logs the stored stamp; cultcache-py: bound stamp offset minutes, pin shape (update-with-retry-py s2 fixes) Co-Authored-By: Claude Sonnet 5.5 --- .../cultcache-py/src/cultcache_py/stores.py | 6 +++--- packages/cultcache-py/tests/test_update.py | 7 +++++-- packages/cultmesh-py/src/cultmesh_py/node.py | 9 +++++---- packages/cultmesh-py/tests/test_cultmesh.py | 20 +++++++++++++++++++ 4 files changed, 33 insertions(+), 9 deletions(-) diff --git a/packages/cultcache-py/src/cultcache_py/stores.py b/packages/cultcache-py/src/cultcache_py/stores.py index c2c3352f..f081417c 100644 --- a/packages/cultcache-py/src/cultcache_py/stores.py +++ b/packages/cultcache-py/src/cultcache_py/stores.py @@ -776,15 +776,15 @@ def _schema_id_for(envelope: CultCacheEnvelope) -> str: _STORED_AT_SHAPE = re.compile( - r"([0-9]{4})-([0-9]{2})-([0-9]{2})T([0-9]{2}):([0-9]{2}):([0-9]{2})(?:\.([0-9]{1,9}))?(?:(Z)|([+-])([0-9]{2}):([0-9]{2}))" + r"([0-9]{4})-([0-9]{2})-([0-9]{2})T([0-9]{2}):([0-9]{2}):([0-9]{2})(?:\.([0-9]{1,9}))?(?:(Z)|([+-])([01][0-9]|2[0-3]):([0-5][0-9]))" ) def _parse_stored_at(text: Any) -> datetime | None: """The UTC instant a stored_at text names, or None. Only the shape the runtimes write is a stamp: an RFC 3339 date-time with T, - an optional fraction of 1 to 9 digits (cut to the microsecond) and Z or a numeric offset. Checked here, not left to + an optional fraction of 1 to 9 digits (cut to the microsecond) and Z or a numeric offset of hours 00-23 and minutes 00-59. Checked here, not left to fromisoformat, whose accepted set differs by interpreter (3.10 and 3.13) and from Rust and TypeScript. An instant outside - datetime's range, or an offset of a day or more, is not a stamp.""" + datetime's range is not a stamp.""" if not isinstance(text, str): return None shape = _STORED_AT_SHAPE.fullmatch(text) diff --git a/packages/cultcache-py/tests/test_update.py b/packages/cultcache-py/tests/test_update.py index 78725ce3..34f030a8 100644 --- a/packages/cultcache-py/tests/test_update.py +++ b/packages/cultcache-py/tests/test_update.py @@ -641,7 +641,8 @@ class StoreRestampTests(Fixture): def test_only_the_stamp_shape_the_runtimes_write_parses(self) -> None: """An RFC 3339 date-time with T, a fraction of 1 to 9 digits and Z or a numeric offset, on every interpreter (the shape is - checked here, not by fromisoformat): Rust and TypeScript read the same set.""" + checked here, not by fromisoformat), so a stamp reads the same on 3.10 and 3.13. The edges follow RFC 3339: offset hours 00-23, + minutes 00-59.""" utc = timezone.utc accepted = { "2026-01-01T00:00:00Z": datetime(2026, 1, 1, tzinfo=utc), @@ -650,6 +651,7 @@ class StoreRestampTests(Fixture): "2026-01-01T00:00:00.1234567+00:00": datetime(2026, 1, 1, 0, 0, 0, 123456, utc), "2026-01-01T05:30:00+05:30": datetime(2026, 1, 1, tzinfo=utc), "2025-12-31T19:00:00-05:00": datetime(2026, 1, 1, tzinfo=utc), + "2026-01-01T23:59:00+23:59": datetime(2026, 1, 1, 0, 0, tzinfo=utc), } for text, instant in accepted.items(): with self.subTest(text=text): @@ -658,7 +660,8 @@ class StoreRestampTests(Fixture): "2026-01-01", "20260101T000000Z", "2026-W01-1T00:00:00Z", "2026-01-01 00:00:00Z", "2026-01-01T00:00:00,5Z", "2026-01-01T00Z", "2026-01-01T00:00:00", "2026-01-01T00:00:00.Z", "2026-01-01T00:00:00.1234567890Z", "2026-01-01T00:00:00z", " 2026-01-01T00:00:00Z", "2026-01-01T00:00:00Z ", "2026-01-01T24:00:00Z", - "2026-13-01T00:00:00Z", "2026-01-01T00:00:00+24:00", "2026-01-01T00:00:00+0000", "\u0662026-01-01T00:00:00Z", + "2026-13-01T00:00:00Z", "2026-01-01T00:00:00+24:00", "2026-01-01T00:00:00+05:60", "2026-01-01T00:00:00+05:75", "2026-01-01T00:00Z", "2026-01-01T00:00+00:00", + "026-01-01T00:00:00Z", "12026-01-01T00:00:00Z", "2026-01-01T00:00:00+5:00", "2026-01-01T00:00:00+0000", "\u0662026-01-01T00:00:00Z", "", "garbage", ): with self.subTest(text=text): diff --git a/packages/cultmesh-py/src/cultmesh_py/node.py b/packages/cultmesh-py/src/cultmesh_py/node.py index 41cd4aec..5ad917ae 100644 --- a/packages/cultmesh-py/src/cultmesh_py/node.py +++ b/packages/cultmesh-py/src/cultmesh_py/node.py @@ -685,10 +685,11 @@ class CultMeshDatabase: value = self.cache.put_envelope(document, envelope) change_kind = "added" if previous is None else "updated" change = self._publish_local_change(document, record_key, change_kind, value, previous) - self._append_shard_log_put(CultNetMessage( - "cultnet.document_put_raw.v0", - {key: value for key, value in message.items() if key != "schemaVersion"}, - ), change_kind) + # The log carries the stamp the cache holds, which a replacing put mints later than the record it replaces, not the peer's. + stored_at = self.cache.get_required_envelope(document, record_key).stored_at + logged = {key: value for key, value in message.items() if key != "schemaVersion"} + logged["document"] = {**message["document"], "storedAt": stored_at} + self._append_shard_log_put(CultNetMessage("cultnet.document_put_raw.v0", logged), change_kind) return change def raw_put_served_alone(self, message: dict[str, Any], *, message_id: str) -> dict[str, Any] | None: diff --git a/packages/cultmesh-py/tests/test_cultmesh.py b/packages/cultmesh-py/tests/test_cultmesh.py index a1e0c35c..4595f29c 100644 --- a/packages/cultmesh-py/tests/test_cultmesh.py +++ b/packages/cultmesh-py/tests/test_cultmesh.py @@ -822,6 +822,26 @@ class CultMeshTests(unittest.TestCase): self.assertEqual(wire["document"]["storedAt"], held) self.assertEqual(node.database._shard_logs["mesh"][-1]["put"]["document"]["storedAt"], held) + def test_applied_raw_put_logs_the_stamp_the_cache_holds(self) -> None: + document = define_database_entry_type("mesh.applied_stamp", [("name", 0), ("category", 1), ("value", 2, 0)], cls=Item) + node = create_node(runtime_id="python-node") + node.register_document(document) + cache = node.database.cache + entry = document.catalog_entry() + + def envelope(value: int, stored_at: str) -> CultCacheEnvelope: + return CultCacheEnvelope("item:applied", document.type, document.encode_payload(Item("n", "gear", value)), stored_at, entry.schema_id, entry) + + cache.put_envelope(document, envelope(1, "2099-01-01T00:00:00.000000+00:00")) + peer = node.put_raw_message(document, "item:peer", Item("peer", "gear", 2)).to_wire() # a well-formed put to replay + peer["document"] = {**peer["document"], "recordKey": "item:applied", "storedAt": "2000-01-01T00:00:00Z", + "payload": envelope(2, "x").payload} + peer["shardId"], peer["shardEpoch"] = "mesh", 1 + node.database.apply_raw_put_message(peer) + held = cache.get_required_envelope(document, "item:applied").stored_at + self.assertGreater(held, "2099-01-01T00:00:00.000000+00:00") + self.assertEqual(node.database._shard_logs["mesh"][-1]["put"]["document"]["storedAt"], held) + def test_cultmesh_node_emits_raw_put_and_delete_messages_for_local_writes(self) -> None: document = define_database_entry_type( "mesh.emit_item", -- 2.47.3 From b2f87cab7e9da16457b895a8ef4664dc71dbb0eb Mon Sep 17 00:00:00 2001 From: Libby Date: Fri, 13 Nov 2026 20:40:42 +0000 Subject: [PATCH 5/6] cultmesh-py: subscriber notification carries the stamp the cache holds (update-with-retry-py s3 fix) Co-Authored-By: Claude Sonnet 5.5 --- .../cultmesh-py/src/cultmesh_py/server.py | 4 ++- .../tests/test_remote_write_authority.py | 27 ++++++++++++++++--- 2 files changed, 27 insertions(+), 4 deletions(-) diff --git a/packages/cultmesh-py/src/cultmesh_py/server.py b/packages/cultmesh-py/src/cultmesh_py/server.py index ab5267a7..8c8864fc 100644 --- a/packages/cultmesh-py/src/cultmesh_py/server.py +++ b/packages/cultmesh-py/src/cultmesh_py/server.py @@ -686,9 +686,11 @@ class CultMeshLocalServer: message_id=str(message.get("messageId") or ""), code="unregistered_document_put", )] + # Subscribers are sent the stamp the store holds, which a replacing put mints later than the record it replaces, not the peer's. + stored = self.node.database.cache.get_required_envelope(change.document, change.record_key).stored_at return self._database_change_notifications( message, - document_record, + {**document_record, "storedAt": stored}, change.change_kind, subscriptions, resolved_schema_id=change.schema_id, diff --git a/packages/cultmesh-py/tests/test_remote_write_authority.py b/packages/cultmesh-py/tests/test_remote_write_authority.py index 3844e589..23ce8c10 100644 --- a/packages/cultmesh-py/tests/test_remote_write_authority.py +++ b/packages/cultmesh-py/tests/test_remote_write_authority.py @@ -8,6 +8,7 @@ the peer the connection loop would pass, and one test drives the same path over from __future__ import annotations import json +import tempfile import unittest from pathlib import Path from typing import Any @@ -23,7 +24,7 @@ from cultmesh_py import ( CultMeshServerPeer, create_node, ) -from cultmesh_py.server import CultMeshLocalServer +from cultmesh_py.server import CultMeshLocalServer, _DatabaseSubscription from cultnet_py import CultNetPeerError, create_rudp_schema_transport, document_delete, document_put_raw, error_message SCHEMA_ID = "sha256:authority-note" @@ -55,12 +56,12 @@ class Recording: return self._decide(request) -def put_of(key: str, text: str, *, schema_id: str = SCHEMA_ID) -> dict[str, Any]: +def put_of(key: str, text: str, *, schema_id: str = SCHEMA_ID, stored_at: str = "2026-11-07T00:00:00Z") -> dict[str, Any]: return document_put_raw( message_id="wire", key=key, schema_id=schema_id, - stored_at="2026-11-07T00:00:00Z", + stored_at=stored_at, payload=NOTE.encode_payload({"schema_version": "tests.authority_note.v1", "text": text}), shard_id=SHARD_ID, shard_epoch=1, @@ -73,6 +74,26 @@ def delete_of(key: str, *, schema_id: str = SCHEMA_ID) -> dict[str, Any]: ).to_wire() +class NotificationStampTests(unittest.TestCase): + """A subscriber is sent the stamp the store holds after a peer's put, as the shard log is, not the peer's stamp.""" + + def test_the_notification_carries_the_stamp_the_cache_holds(self) -> None: + local, peer_stamp = "2099-01-01T00:00:00.000000+00:00", "2000-01-01T00:00:00Z" + with tempfile.TemporaryDirectory() as directory: + for backing in (None, Path(directory) / "node.cc"): # store-less, then store-backed + with self.subTest(store=backing is not None): + node = create_node(backing, runtime_id="stamp-server") + node.database.register_document(NOTE) + server = CultMeshLocalServer(node=node, port=0, write_authority=CultMeshGrantAllWrites()) + node.database.apply_raw_put_message(put_of(KEY_A, "server", stored_at=local)) + subscriptions = {"s": _DatabaseSubscription("s", set(), set())} + replies = server._handle_connection_message(put_of(KEY_A, "peer", stored_at=peer_stamp), subscriptions, peer=PEER) + held = node.cache.get_required_envelope(NOTE, KEY_A).stored_at + self.assertGreater(held, local) + self.assertEqual([reply["document"]["storedAt"] for reply in replies], [held]) + self.assertEqual(node.database._shard_logs[SHARD_ID][-1]["put"]["document"]["storedAt"], held) + + class RemoteWriteAuthorityTests(unittest.TestCase): def build(self, authority: Any = None) -> CultMeshLocalServer: node = create_node(runtime_id="authority-server") -- 2.47.3 From f0c2eca095f2e1668a9f969d3145a1425967df9a Mon Sep 17 00:00:00 2001 From: Libby Date: Sat, 14 Nov 2026 02:15:17 +0000 Subject: [PATCH 6/6] ci: rerun after run 930 failed with no log access Co-Authored-By: Claude Sonnet 5.5 -- 2.47.3