cultcache-py: update-with-retry (update-with-retry-py) (#172)
This commit is contained in:
parent
afab2d5a12
commit
c77d734da5
16 changed files with 1198 additions and 63 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)`, 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 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).
|
||||
|
|
@ -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`.
|
||||
|
|
|
|||
|
|
@ -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 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.
|
||||
|
||||
## Store format and rollback
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
@ -609,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
|
||||
|
|
@ -663,18 +718,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 +778,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:
|
||||
|
|
|
|||
|
|
@ -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 `<path>.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,65 @@ 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)|([+-])([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 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 is not a stamp."""
|
||||
if not isinstance(text, str):
|
||||
return None
|
||||
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
|
||||
|
||||
|
||||
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`. 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:
|
||||
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")
|
||||
|
||||
|
||||
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(
|
||||
|
|
|
|||
38
packages/cultcache-py/tests/test_content_version_vector.py
Normal file
38
packages/cultcache-py/tests/test_content_version_vector.py
Normal file
|
|
@ -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()
|
||||
|
|
@ -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"))
|
||||
|
|
|
|||
|
|
@ -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__":
|
||||
|
|
|
|||
|
|
@ -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):
|
||||
|
|
|
|||
731
packages/cultcache-py/tests/test_update.py
Normal file
731
packages/cultcache-py/tests/test_update.py
Normal file
|
|
@ -0,0 +1,731 @@
|
|||
"""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 shutil
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
import unittest
|
||||
from dataclasses import asdict, dataclass
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
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_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] = []
|
||||
|
||||
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_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), 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),
|
||||
"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),
|
||||
"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):
|
||||
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+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):
|
||||
self.assertIsNone(_parse_stored_at(text))
|
||||
|
||||
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"):
|
||||
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()
|
||||
|
|
@ -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,
|
||||
|
|
@ -683,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:
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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,51 @@ 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_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",
|
||||
|
|
|
|||
|
|
@ -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")
|
||||
|
|
|
|||
Loading…
Reference in a new issue