cultcache-py: strict stored_at shape, range edges refuse, mesh publishes the stored stamp, home-swap guard pinned (update-with-retry-py s1 fixes)
Some checks failed
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (pull_request) Has been cancelled
Some checks failed
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (pull_request) Has been cancelled
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
parent
35b6f8c0fe
commit
cb6e044c0f
6 changed files with 141 additions and 34 deletions
|
|
@ -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).
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
||||
|
|
|
|||
|
|
@ -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")
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
Loading…
Reference in a new issue