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",