From 4d577e26f2e3fbf65eadca8b76e7007a79b1884a Mon Sep 17 00:00:00 2001 From: Sashank Bhamidi Date: Wed, 21 Oct 2026 19:27:04 +0530 Subject: [PATCH 1/4] feat: check argument types the same way on every store --- docs/contribute.rst | 3 + news/27.feature.rst | 1 + src/calendaring_sync/_checks.py | 304 +++++++++++++++++++ src/calendaring_sync/_text.py | 66 ---- src/calendaring_sync/objects/change.py | 9 +- src/calendaring_sync/objects/change_set.py | 4 +- src/calendaring_sync/objects/push.py | 6 +- src/calendaring_sync/objects/resolution.py | 7 +- src/calendaring_sync/replica.py | 19 +- src/calendaring_sync/store/base.py | 6 + src/calendaring_sync/store/memory.py | 2 + src/calendaring_sync/store/sqlite.py | 3 + src/calendaring_sync/tests/test_sync_unit.py | 120 +++++++- 13 files changed, 462 insertions(+), 88 deletions(-) create mode 100644 news/27.feature.rst create mode 100644 src/calendaring_sync/_checks.py delete mode 100644 src/calendaring_sync/_text.py diff --git a/docs/contribute.rst b/docs/contribute.rst index e511ce8..51a604e 100644 --- a/docs/contribute.rst +++ b/docs/contribute.rst @@ -125,6 +125,9 @@ No duplication No hardcoding A literal that means something (a status value, a format string, a default path) belongs in a named constant defined once, not inlined at every call site. A one-off literal with no reused meaning (a test fixture's arbitrary ID) doesn't need this. +Argument types + The public methods of ``Replica``, ``Session`` and the stores, and the value classes an adapter builds, check their argument types at runtime, from their annotations, so give every parameter of a new public method an annotation. A built-in type such as ``str`` or ``int`` has to match exactly, so ``True`` isn't taken as ``1``. The unit tests build every check and fail on a parameter without an annotation, or on one the checks can't read. + RFC citations The core doesn't cite RFCs; see :ref:`write-about-sync`. In adapter docstrings (not comments, not test files, see below), cite a real, finalized RFC with Sphinx's ``:rfc:`` role, and always merge the section number into the role itself: ``:rfc:`5545#section-3.6.1```, not ``:rfc:`5545` section 3.6.1``. The latter renders as two disconnected pieces, a link to the RFC's front page and plain text next to it, so a reader has to manually find the section after clicking through. Verify the section number against the actual RFC text before citing it; don't guess from a related section or trust an existing comment's number without checking. diff --git a/news/27.feature.rst b/news/27.feature.rst new file mode 100644 index 0000000..5d03446 --- /dev/null +++ b/news/27.feature.rst @@ -0,0 +1 @@ +Every public method of ``Replica``, ``Session`` and the stores, and every value an adapter or application builds, now checks its argument types and raises ``TypeError`` for a wrong one, the same way on every store. Before, a wrong type could be stored as it was by one store and converted by another, such as ``delete_record("1")``, which deleted record 1 only on SQLite. diff --git a/src/calendaring_sync/_checks.py b/src/calendaring_sync/_checks.py new file mode 100644 index 0000000..db29c9d --- /dev/null +++ b/src/calendaring_sync/_checks.py @@ -0,0 +1,304 @@ +# SPDX-FileCopyrightText: 2026 calendaring-sync contributors +# SPDX-License-Identifier: AGPL-3.0-or-later + +"""The checks on what enters the core: argument types, and strings every store can keep. + +A value's type is checked where it enters the library, so a wrong type fails +the same way on every store instead of being stored as it is by one and +converted by another. The checks are built from the annotations the first +time they're needed, as one plain function each, since they run on every +call. +""" + +from __future__ import annotations + +import dataclasses +import inspect +import types +import typing +from collections.abc import Callable, Iterable, Mapping +from functools import wraps +from typing import Any, TypeVar + +_T = TypeVar("_T") + +#: Built-in types a value must be exactly, not a subclass of: ``True`` is +#: no local key, and a ``str`` subclass could change how two values compare. +_EXACT = (str, int, bytes, bool) + +#: Value dataclasses that check their own fields when built. Being frozen, +#: one inside another value only needs to be of its class. +_CHECKED_WHEN_BUILT: set[type] = set() + +# Marks an argument the caller left to its default. +_MISSING = object() + + +def _kind(hint: Any) -> tuple[str, tuple[Any, ...]]: + """What kind of type a hint asks for, and the hints or class inside it. + + Raises: + TypeError: If the checks can't read the hint. + """ + origin = typing.get_origin(hint) + if hint is type(None): + return "none", () + if origin in (typing.Union, types.UnionType): + return "union", typing.get_args(hint) + if hint in _EXACT: + return "exact", (hint,) + if hint is float: + return "number", () + if origin is frozenset: + return "frozenset", typing.get_args(hint) + if isinstance(origin, type) and issubclass(origin, Mapping): + return "mapping", typing.get_args(hint) + if origin is not None and getattr(origin, "__name__", "") == "Callable": + return "callable", () + cls = origin if isinstance(origin, type) else hint + if not isinstance(cls, type): + raise TypeError(f"Can't check a value against {hint!r}") + if dataclasses.is_dataclass(cls) and cls not in _CHECKED_WHEN_BUILT: + return "fields", (cls,) + return "instance", (cls,) + + +def _describe(hint: Any) -> str: + """The type a hint asks for, as an error message names it.""" + kind, inner = _kind(hint) + match kind: + case "none": + return "None" + case "union": + names = [_describe(part) for part in inner] + return ", ".join(names[:-1]) + " or " + names[-1] + case "number": + return "float or int" + case "frozenset": + return f"frozenset of {_describe(inner[0])}" + case "mapping": + return f"mapping of {_describe(inner[0])} to {_describe(inner[1])}" + case "callable": + return "callable" + # A class, exact or not. + return str(inner[0].__name__) + + +def _test(hint: Any, value: str, env: dict[str, object]) -> str: + """Python source that's true when ``value`` matches ``hint``. + + Names the source needs go into ``env``. + """ + kind, inner = _kind(hint) + match kind: + case "none": + return f"{value} is None" + case "union": + return "(" + " or ".join(_test(part, value, env) for part in inner) + ")" + case "exact": + return f"type({value}) is {inner[0].__name__}" + case "number": + return f"type({value}) in (float, int)" + case "frozenset": + items = f"all({_test(inner[0], '_i', env)} for _i in {value})" + return f"(type({value}) is frozenset and {items})" + case "mapping": + env["Mapping"] = Mapping + pairs = f"{_test(inner[0], '_k', env)} and {_test(inner[1], '_v', env)}" + return f"(isinstance({value}, Mapping) and all({pairs} for _k, _v in {value}.items()))" + case "callable": + return f"callable({value})" + cls = inner[0] + name = f"_{cls.__module__.replace('.', '_')}_{cls.__qualname__.replace('.', '_')}" + env[name] = cls + if kind == "fields": + env[f"{name}_fields"] = _fields(cls) + return f"(isinstance({value}, {name}) and {name}_fields({value}))" + return f"isinstance({value}, {name})" + + +def _compile(source: str, env: dict[str, object]) -> Callable[..., Any]: + """Run the source of one function named ``check`` and return it.""" + exec(source, env) # noqa: S102, the source is built above from annotations only + return env["check"] # type: ignore[return-value] + + +_FIELDS: dict[type, Callable[[object], bool]] = {} + + +def _fields(cls: type) -> Callable[[object], bool]: + """A function that's true when every field of a dataclass instance has its declared type.""" + if cls not in _FIELDS: + env: dict[str, object] = {} + hints = typing.get_type_hints(cls) + tests = [_test(hints[f.name], f"o.{f.name}", env) for f in dataclasses.fields(cls)] + _FIELDS[cls] = _compile("def check(o):\n return " + " and ".join(tests or ["True"]), env) + return _FIELDS[cls] + + +def _field_mismatch(cls: type, value: object, prefix: str) -> str | None: + """The message for the first field of a dataclass instance that has the wrong type.""" + hints = typing.get_type_hints(cls) + found = ( + _mismatch(prefix + f.name, hints[f.name], getattr(value, f.name)) + for f in dataclasses.fields(cls) + ) + return next((message for message in found if message is not None), None) + + +def _inner_mismatch(name: str, hint: Any, value: object) -> str | None: + """For a value of the kind ``hint`` holds, the message naming what inside it is wrong.""" + kind, inner = _kind(hint) + if kind == "fields" and isinstance(value, inner[0]): + return _field_mismatch(inner[0], value, f"{name}.") + if kind == "frozenset" and type(value) is frozenset: + bad = next((each for each in value if _mismatch("item", inner[0], each) is not None), None) + return f"{name} must be {_describe(hint)}, but holds {type(bad).__name__}" + if kind == "mapping" and isinstance(value, Mapping): + key, item = inner + found = ( + _mismatch(f"{name}[{k!r}]", item, v) or _mismatch(f"{name} key", key, k) + for k, v in value.items() + ) + return next((message for message in found if message is not None), None) + return None + + +def _mismatch(name: str, hint: Any, value: object) -> str | None: + """The message for a value that doesn't match its hint, naming the part of it that's wrong.""" + env: dict[str, object] = {} + if _compile(f"def check(v):\n return {_test(hint, 'v', env)}", env)(value): + return None + kind, inner_hints = _kind(hint) + parts = inner_hints if kind == "union" else (hint,) + inner = (_inner_mismatch(name, part, value) for part in parts) + found = next((message for message in inner if message is not None), None) + return found or f"{name} must be {_describe(hint)}, not {type(value).__name__}" + + +def check_types(obj: object) -> None: + """Raise ``TypeError`` unless every field of a dataclass instance has its declared type. + + Fields holding a dataclass are checked in turn, unless that class checks + its own fields when built. + """ + if not _fields(type(obj))(obj): + raise TypeError(_field_mismatch(type(obj), obj, "")) + + +def checked_when_built(cls: type[_T]) -> type[_T]: + """Check a value dataclass's field types every time one is built, before its own checks. + + The class needs a ``__post_init__`` of its own, which runs after the check. + """ + original = cls.__dict__["__post_init__"] + + def __post_init__(self: object) -> None: + check_types(self) + original(self) + + cls.__post_init__ = __post_init__ # type: ignore[attr-defined] + _CHECKED_WHEN_BUILT.add(cls) + return cls + + +#: Checks one call's positional and keyword arguments. +_ArgumentCheck = Callable[[tuple[Any, ...], dict[str, Any]], None] + + +def argument_check(func: Callable[..., object]) -> _ArgumentCheck: + """Build the function that checks a call's arguments against ``func``'s annotations.""" + hints = typing.get_type_hints(func) + env: dict[str, object] = {"_MISSING": _MISSING, "_fail": _fail} + lines = ["def check(args, kwargs):", " n = len(args)"] + for i, param in enumerate(inspect.signature(func).parameters.values()): + if param.name in ("self", "cls"): + continue + if param.kind in (param.VAR_POSITIONAL, param.VAR_KEYWORD): + raise TypeError(f"{func.__qualname__} takes *{param.name}, which isn't checked") + if param.name not in hints: + raise TypeError(f"{func.__qualname__} takes {param.name} without a type to check") + env[f"_h_{param.name}"] = hints[param.name] + if param.kind is param.KEYWORD_ONLY: + lines.append(f" v = kwargs.get({param.name!r}, _MISSING)") + else: + lines.append(f" v = args[{i}] if n > {i} else kwargs.get({param.name!r}, _MISSING)") + test = _test(hints[param.name], "v", env) + lines.append(f" if v is not _MISSING and not ({test}):") + lines.append(f" _fail({param.name!r}, _h_{param.name}, v)") + return _compile("\n".join(lines), env) + + +def _fail(name: str, hint: Any, value: object) -> typing.NoReturn: + raise TypeError(_mismatch(name, hint, value)) + + +def check_argument_types(cls: type[_T]) -> type[_T]: + """Check the argument types of every public method and ``__init__`` a class defines. + + Each check is built from the method's annotations on its first call, so + an annotation may name a class defined later in the module. + """ + for name, func in list(vars(cls).items()): + if inspect.isfunction(func) and (not name.startswith("_") or name == "__init__"): + setattr(cls, name, _checking(func)) + return cls + + +def _checking(func: Callable[..., _T]) -> Callable[..., _T]: + check: _ArgumentCheck | None = None + + @wraps(func) + def wrapper(*args: Any, **kwargs: Any) -> _T: + nonlocal check + if check is None: + check = argument_check(func) + check(args, kwargs) + return func(*args, **kwargs) + + wrapper._checks_arguments = True # type: ignore[attr-defined] + return wrapper + + +def check_text(**values: str | Iterable[str] | None) -> None: + """Raise ``ValueError`` if a string holds a lone surrogate. + + Every store keeps any string exactly, empty strings and whitespace + included, except one kind. Python allows a ``str`` to hold a lone + surrogate, half of the pair UTF-16 uses for characters such as emoji, + for example ``"\\ud800"``. It's not a character, and UTF-8 can't encode + it, so a store that writes UTF-8 fails on it while one that keeps + Python objects doesn't. Refusing it where the string comes in keeps + every store the same. + + Args: + values: Each value by its argument's name: a string, strings, or + ``None``. + """ + for name, value in values.items(): + if value is None: + continue + for text in [value] if isinstance(value, str) else value: + try: + text.encode() + except UnicodeEncodeError as exc: + raise ValueError( + f"{name} holds a lone surrogate at {exc.start}, which can't be stored" + ) from exc + + +def check_data( + fingerprint: str | None, + containers: frozenset[str], + logical_key: str | None, + **values: str | None, +) -> None: + """Check the strings of a copy of an object with :func:`check_text`. + + Args: + fingerprint: Its fingerprint. + containers: Its containers. + logical_key: Its logical key. + values: Any other strings that come with it, by name. + """ + check_text(fingerprint=fingerprint, containers=containers, logical_key=logical_key, **values) diff --git a/src/calendaring_sync/_text.py b/src/calendaring_sync/_text.py deleted file mode 100644 index 90e55e9..0000000 --- a/src/calendaring_sync/_text.py +++ /dev/null @@ -1,66 +0,0 @@ -# SPDX-FileCopyrightText: 2026 calendaring-sync contributors -# SPDX-License-Identifier: AGPL-3.0-or-later - -"""The checks on what the core stores: strings every store can keep, and content as bytes.""" - -from __future__ import annotations - -from collections.abc import Iterable - - -def check_text(**values: str | Iterable[str] | None) -> None: - """Raise ``ValueError`` if a string holds a lone surrogate. - - Every store keeps any string exactly, empty strings and whitespace - included, except one kind. Python allows a ``str`` to hold a lone - surrogate, half of the pair UTF-16 uses for characters such as emoji, - for example ``"\\ud800"``. It's not a character, and UTF-8 can't encode - it, so a store that writes UTF-8 fails on it while one that keeps - Python objects doesn't. Refusing it where the string comes in keeps - every store the same. - - Args: - values: Each value by its argument's name: a string, strings, or - ``None``. - """ - for name, value in values.items(): - if value is None: - continue - for text in [value] if isinstance(value, str) else value: - try: - text.encode() - except UnicodeEncodeError as exc: - raise ValueError( - f"{name} holds a lone surrogate at {exc.start}, which can't be stored" - ) from exc - - -def check_content(content: object) -> None: - """Raise ``TypeError`` if content isn't bytes. - - Content is stored exactly as the server sent it, so it's never decoded. - A ``str`` would be kept as it is by one store and written differently by - another, so it's refused. - """ - if not isinstance(content, bytes): - raise TypeError(f"content must be bytes, not {type(content).__name__}") - - -def check_data( - content: object, - fingerprint: str | None, - containers: frozenset[str], - logical_key: str | None, - **values: str | None, -) -> None: - """Check a copy of an object: its content with :func:`check_content`, and its strings. - - Args: - content: The copy's content. - fingerprint: Its fingerprint. - containers: Its containers. - logical_key: Its logical key. - values: Any other strings that come with it, by name. - """ - check_content(content) - check_text(fingerprint=fingerprint, containers=containers, logical_key=logical_key, **values) diff --git a/src/calendaring_sync/objects/change.py b/src/calendaring_sync/objects/change.py index 1ec7b0c..89721ae 100644 --- a/src/calendaring_sync/objects/change.py +++ b/src/calendaring_sync/objects/change.py @@ -7,9 +7,10 @@ from dataclasses import dataclass -from calendaring_sync._text import check_data +from calendaring_sync._checks import check_data, checked_when_built +@checked_when_built @dataclass(frozen=True) class Changed: """The object changed on the server, or is new there. @@ -25,7 +26,7 @@ class Changed: logical_key: The object's :term:`logical key`, if it has one. Raises: - TypeError: If the content isn't bytes. + TypeError: If a field has the wrong type. ValueError: If a string holds a :ref:`lone surrogate `. Examples: @@ -46,9 +47,7 @@ class Changed: logical_key: str | None = None def __post_init__(self) -> None: - check_data( - self.content, self.fingerprint, self.containers, self.logical_key, version=self.version - ) + check_data(self.fingerprint, self.containers, self.logical_key, version=self.version) @dataclass(frozen=True) diff --git a/src/calendaring_sync/objects/change_set.py b/src/calendaring_sync/objects/change_set.py index 313c8ec..4e9803d 100644 --- a/src/calendaring_sync/objects/change_set.py +++ b/src/calendaring_sync/objects/change_set.py @@ -8,10 +8,11 @@ from collections.abc import Mapping from dataclasses import dataclass -from calendaring_sync._text import check_text +from calendaring_sync._checks import check_text, checked_when_built from calendaring_sync.objects.change import Change +@checked_when_built @dataclass(frozen=True) class ChangeSet: """One :term:`change set`, or one :term:`truncated page` of it, for one :term:`scope`. @@ -33,6 +34,7 @@ class ChangeSet: truncated: Whether more pages follow it. Raises: + TypeError: If a field has the wrong type. ValueError: If the scope id, a token or an object id holds a :ref:`lone surrogate `. """ diff --git a/src/calendaring_sync/objects/push.py b/src/calendaring_sync/objects/push.py index 94479a2..2ed59c3 100644 --- a/src/calendaring_sync/objects/push.py +++ b/src/calendaring_sync/objects/push.py @@ -7,9 +7,10 @@ from dataclasses import dataclass -from calendaring_sync._text import check_text +from calendaring_sync._checks import check_text, checked_when_built +@checked_when_built @dataclass(frozen=True) class Applied: """The server applied the edit. @@ -20,6 +21,7 @@ class Applied: object to another id. ``None`` keeps the record's id. Raises: + TypeError: If the id isn't a string or ``None``. ValueError: If the id holds a :ref:`lone surrogate `. Examples: @@ -55,6 +57,7 @@ class Gone: """ +@checked_when_built @dataclass(frozen=True) class LogicalKeyTaken: """The server refused a create because another object holds its :term:`logical key`. @@ -64,6 +67,7 @@ class LogicalKeyTaken: the server names it; ``None`` when it doesn't. Raises: + TypeError: If the id isn't a string or ``None``. ValueError: If the id holds a :ref:`lone surrogate `. """ diff --git a/src/calendaring_sync/objects/resolution.py b/src/calendaring_sync/objects/resolution.py index c394fbd..1cb17f1 100644 --- a/src/calendaring_sync/objects/resolution.py +++ b/src/calendaring_sync/objects/resolution.py @@ -7,7 +7,7 @@ from dataclasses import KW_ONLY, dataclass -from calendaring_sync._text import check_data +from calendaring_sync._checks import check_data, checked_when_built @dataclass(frozen=True) @@ -28,6 +28,7 @@ class KeepEdit: """ +@checked_when_built @dataclass(frozen=True) class Merge: """Push new content you made from both sides, over the server's current copy. @@ -43,7 +44,7 @@ class Merge: logical_key: Its :term:`logical key`, if it has one. Raises: - TypeError: If the content isn't bytes. + TypeError: If a field has the wrong type. ValueError: If a string holds a :ref:`lone surrogate `. Examples: @@ -62,7 +63,7 @@ class Merge: logical_key: str | None = None def __post_init__(self) -> None: - check_data(self.content, self.fingerprint, self.containers, self.logical_key) + check_data(self.fingerprint, self.containers, self.logical_key) @dataclass(frozen=True) diff --git a/src/calendaring_sync/replica.py b/src/calendaring_sync/replica.py index 30c26a4..132adda 100644 --- a/src/calendaring_sync/replica.py +++ b/src/calendaring_sync/replica.py @@ -17,6 +17,7 @@ from threading import Lock from typing import TypeVar, cast +from calendaring_sync._checks import check_argument_types, check_data, check_text from calendaring_sync._classify import ( ChangeClass, classify, @@ -24,7 +25,6 @@ same_data, same_remote, ) -from calendaring_sync._text import check_data, check_text from calendaring_sync.constants import FEED_PAGE_SIZE from calendaring_sync.error import ( DuplicateObjectError, @@ -254,6 +254,7 @@ class SessionState(StrEnum): PUSHING = "pushing" +@check_argument_types class Session: """One sync of one :term:`scope`, opened with :meth:`Replica.session`. @@ -262,6 +263,9 @@ class Session: record :meth:`to_push` lists, it :meth:`claims ` it, sends it, and :meth:`settles ` it with how the push ended. + Every method raises :class:`TypeError` for an argument of the wrong + type, before it changes anything. + Examples: The adapter pushes an edit after ingesting what changed on the server, here nothing: @@ -503,6 +507,7 @@ def _end(self) -> None: self._restart() +@check_argument_types class Replica: """The sync state of every :term:`scope` in one :term:`store`. @@ -519,7 +524,9 @@ class Replica: methods can't run inside a transaction already open in the same thread; they raise :class:`~calendaring_sync.error.TransactionError`. - Any method can also raise the store's errors: + Every method raises :class:`TypeError` for an argument of the wrong + type, before it changes anything. Any method can also raise the + store's errors: :class:`~calendaring_sync.error.StoreBusyError` when another process kept the store busy, which is safe to retry, :class:`~calendaring_sync.error.StoreError` when the store can't read or @@ -976,7 +983,6 @@ def create( UnknownScopeError: If there's no scope with this id. DuplicateObjectError: If a record that isn't removed holds the object id. - TypeError: If the content isn't bytes. ValueError: If a string holds a :ref:`lone surrogate `. Examples: @@ -986,9 +992,7 @@ def create( >>> replica.record(key).edit.content b'Lunch' """ - check_data( - content, fingerprint, containers, logical_key, scope_id=scope_id, object_id=object_id - ) + check_data(fingerprint, containers, logical_key, scope_id=scope_id, object_id=object_id) with self._store.transaction(): self._existing_scope(scope_id) edit = LocalEdit( @@ -1034,7 +1038,6 @@ def change( OpenConflictError: If the record holds an open :term:`conflict`. RemovedRecordError: If the record was removed from the server. PushInFlightError: If the record's :term:`push` is :term:`in flight`. - TypeError: If the content isn't bytes. ValueError: If a string holds a :ref:`lone surrogate `. Examples: @@ -1049,7 +1052,7 @@ def change( >>> replica.record(key).state """ - check_data(content, fingerprint, containers, logical_key) + check_data(fingerprint, containers, logical_key) with self._editing(local_key) as record: edit = LocalEdit( EditKind.CHANGE, content, fingerprint, containers, logical_key, self._clock() diff --git a/src/calendaring_sync/store/base.py b/src/calendaring_sync/store/base.py index 5e80887..4485943 100644 --- a/src/calendaring_sync/store/base.py +++ b/src/calendaring_sync/store/base.py @@ -18,6 +18,7 @@ from types import TracebackType from typing import Self +from calendaring_sync._checks import check_argument_types from calendaring_sync.constants import ISSUES_URL from calendaring_sync.error import ( CalendaringSyncError, @@ -91,6 +92,7 @@ def _check_record(record: Record) -> None: raise RecordValueError(f"A {conflict.kind} conflict can't name a holder") +@check_argument_types class StateStore(ABC): """Where the core keeps each :term:`scope`, its records, and the :term:`change feed`. @@ -102,6 +104,10 @@ class StateStore(ABC): Besides the errors each method lists, every operation, :meth:`transaction` and :meth:`close` can raise: + - :class:`TypeError`, for an argument of the wrong type, before anything + is read or written; a record or scope to write is checked field by + field + - :class:`~calendaring_sync.error.StoreError`, if the store can't read or write its data - :class:`~calendaring_sync.error.UnexpectedStoreError`, for any other diff --git a/src/calendaring_sync/store/memory.py b/src/calendaring_sync/store/memory.py index f1143d6..3b9efda 100644 --- a/src/calendaring_sync/store/memory.py +++ b/src/calendaring_sync/store/memory.py @@ -8,6 +8,7 @@ from dataclasses import dataclass, field, replace from datetime import datetime +from calendaring_sync._checks import check_argument_types from calendaring_sync.objects.feed import FeedEntry, FeedEntryKind from calendaring_sync.objects.record import Record, RecordState from calendaring_sync.objects.scope import Scope @@ -31,6 +32,7 @@ def copy(self) -> _Data: ) +@check_argument_types class MemoryStore(StateStore): """A store held in memory and lost when the process ends. diff --git a/src/calendaring_sync/store/sqlite.py b/src/calendaring_sync/store/sqlite.py index 0c65ffc..f4d0a44 100644 --- a/src/calendaring_sync/store/sqlite.py +++ b/src/calendaring_sync/store/sqlite.py @@ -16,6 +16,7 @@ from types import UnionType from typing import Any, NamedTuple, TypeVar, Union, cast, get_args, get_origin, get_type_hints +from calendaring_sync._checks import check_argument_types from calendaring_sync.constants import DEFAULT_TIMEOUT, SCHEMA_VERSION from calendaring_sync.error import ( CalendaringSyncError, @@ -307,6 +308,7 @@ def _load(reader: _Reader, row: sqlite3.Row) -> Any: return reader.cls(**values) +@check_argument_types class SQLiteStore(StateStore): """A store in a SQLite database file, shared between threads and processes. @@ -327,6 +329,7 @@ class SQLiteStore(StateStore): timeout: Seconds to wait for another process before giving up. Raises: + TypeError: If ``path`` or ``timeout`` has the wrong type. StoreVersionError: If a newer calendaring-sync wrote the file. StoreBusyError: If another process kept the file busy past ``timeout``. StoreError: If the file can't be opened or isn't a store. diff --git a/src/calendaring_sync/tests/test_sync_unit.py b/src/calendaring_sync/tests/test_sync_unit.py index 1e9060f..9a1ed29 100644 --- a/src/calendaring_sync/tests/test_sync_unit.py +++ b/src/calendaring_sync/tests/test_sync_unit.py @@ -9,8 +9,10 @@ from __future__ import annotations +import dataclasses import doctest import importlib +import inspect import pkgutil import re import runpy @@ -24,7 +26,7 @@ from datetime import UTC, datetime, timedelta from functools import partial from pathlib import Path -from typing import cast +from typing import Any, Literal from unittest.mock import patch import pytest @@ -83,6 +85,7 @@ UnknownScopeError, UpdateHolder, ) +from calendaring_sync._checks import _CHECKED_WHEN_BUILT, _fields, argument_check from calendaring_sync.constants import FEED_PAGE_SIZE, ISSUES_URL, SCHEMA_VERSION from calendaring_sync.store.sqlite import _FEED, _META, _RECORDS, _reader from calendaring_sync.tests import benchmark @@ -609,6 +612,35 @@ def test_benchmarks_run_and_measure_what_they_say(self, monkeypatch, capsys): } assert benchmark.entries_after_pushes(size) == {FeedEntryKind.PUSHED: edited} + def test_every_entry_point_checks_its_types(self): + for cls in (Replica, Session, StateStore, MemoryStore, SQLiteStore): + for name, func in vars(cls).items(): + if inspect.isfunction(func) and (not name.startswith("_") or name == "__init__"): + assert getattr(func, "_checks_arguments", False), f"{cls.__name__}.{name}" + # Builds the check; an annotation it can't read raises here. + argument_check(inspect.unwrap(func)) + exported = [getattr(calendaring_sync, name) for name in calendaring_sync.__all__] + for cls in exported: + if isinstance(cls, type) and dataclasses.is_dataclass(cls): + _fields(cls) + # The values an adapter or application builds check themselves. + assert _CHECKED_WHEN_BUILT == {Changed, ChangeSet, Applied, LogicalKeyTaken, Merge} + + def test_a_check_refuses_what_it_cant_read(self): + def takes_many(*values: str) -> None: ... + + def takes_a_literal(kind: Literal["a"]) -> None: ... + + def takes_anything(value) -> None: ... + + for func, match in [ + (takes_many, r"takes \*values, which isn't checked"), + (takes_a_literal, "^Can't check a value against"), + (takes_anything, "takes value without a type to check$"), + ]: + with pytest.raises(TypeError, match=match): + argument_check(func) + def test_version_is_set(self): assert isinstance(calendaring_sync.__version__, str) assert calendaring_sync.__version__ @@ -1307,8 +1339,13 @@ def _conflicted(edit: EditKind, kind: ConflictKind) -> tuple[object, ...]: # A string no store can keep: a lone surrogate. _BAD_TEXT = "a\ud800b" -# Content that isn't bytes, typed as bytes so the calls refusing it type-check. -_NOT_BYTES = cast(bytes, "text of fa") + +def _wrong(value: object) -> Any: + """A value of the wrong type, passed where the type checker can't object.""" + return value + + +_NOT_BYTES = _wrong("text of fa") class TestLocalEdits: @@ -1962,7 +1999,7 @@ def test_content_and_strings_round_trip_exactly(self, replica, content, text): assert (record.edit.containers, record.edit.logical_key) == (frozenset({text}), text) assert replica.record(created).edit.content == content - def test_invalid_text_and_content_are_refused(self, replica, store): + def test_wrong_types_and_text_are_refused(self, replica, store, db_path): # The values an adapter or the application builds check themselves. built: list[tuple[Callable[[], object], type[Exception], str]] = [ ( @@ -1971,19 +2008,64 @@ def test_invalid_text_and_content_are_refused(self, replica, store): "^fingerprint holds a lone surrogate at 1", ), (lambda: Changed("v1", "fa", _NOT_BYTES), TypeError, "^content must be bytes, not str"), + ( + lambda: Changed(_wrong(1), "fa", b"c"), + TypeError, + "^version must be str or None, not int", + ), + ( + lambda: Changed("v1", _wrong(None), b"c"), + TypeError, + "^fingerprint must be str, not NoneType", + ), + ( + lambda: Changed("v1", "fa", b"c", _wrong("cal")), + TypeError, + "^containers must be frozenset of str, not str", + ), + ( + lambda: Changed("v1", "fa", b"c", frozenset({_wrong(1)})), + TypeError, + "^containers must be frozenset of str, but holds int", + ), + ( + lambda: ChangeSet(_wrong(1), None, "T1", {}), + TypeError, + "^scope must be str, not int", + ), + ( + lambda: ChangeSet(_SCOPE, None, "T1", _wrong([])), + TypeError, + "^changes must be mapping of str to Changed, Removed or Unchanged, not list", + ), (lambda: _page(None, "T1", {_BAD_TEXT: Removed()}), ValueError, "^object_ids "), + ( + lambda: _page(None, "T1", {"a": _wrong("removed")}), + TypeError, + r"^changes\['a'\] must be Changed, Removed or Unchanged, not str", + ), (lambda: Applied(_BAD_TEXT), ValueError, "^object_id "), + (lambda: Applied(_wrong(1)), TypeError, "^object_id must be str or None, not int"), (lambda: LogicalKeyTaken(_BAD_TEXT), ValueError, "^holder "), (lambda: Merge(b"c", fingerprint=_BAD_TEXT), ValueError, "^fingerprint "), (lambda: Merge(_NOT_BYTES, fingerprint="f"), TypeError, "^content "), + ( + lambda: SQLiteStore(db_path, timeout=_wrong("5")), + TypeError, + "^timeout must be float", + ), + (lambda: Replica(store, clock=_wrong(5)), TypeError, "^clock must be callable or None"), ] for build, error, match in built: with pytest.raises(error, match=match): build() + # Calls on the replica and the store refuse them on every store, before a transaction. _seeded(replica) key = _key(replica) + synced = replica.record(key) refused: list[tuple[Callable[[], object], type[Exception], str]] = [ (lambda: _enter(_open_session(replica, _BAD_TEXT)), ValueError, "^scope_id "), + (lambda: _enter(_open_session(replica, _wrong(1))), TypeError, "^scope_id must be str"), ( lambda: replica.create( _SCOPE, b"c", fingerprint="f", containers=frozenset({_BAD_TEXT}) @@ -1998,6 +2080,36 @@ def test_invalid_text_and_content_are_refused(self, replica, store): ), (lambda: replica.create(_SCOPE, _NOT_BYTES, fingerprint="f"), TypeError, "^content "), (lambda: replica.change(key, _NOT_BYTES, fingerprint="f"), TypeError, "^content "), + (lambda: replica.record(_wrong("1")), TypeError, "^local_key must be int, not str"), + (lambda: replica.record(_wrong(True)), TypeError, "^local_key must be int, not bool"), + (lambda: replica.read_feed(_wrong("0")), TypeError, "^after must be int, not str"), + ( + lambda: store.delete_record(_wrong("1")), + TypeError, + "^local_key must be int, not str", + ), + ( + lambda: store.insert_record(replace(synced, containers=_wrong("cal"))), + TypeError, + "^record.containers must be frozenset of str, not str", + ), + ( + lambda: store.update_record( + replace( + synced, + state=RecordState.PENDING, + edit=replace(_EDIT, containers=frozenset({_wrong(1)})), + ) + ), + TypeError, + "^record.edit.containers must be frozenset of str, but holds int", + ), + ( + lambda: store.put_scope(replace(_new_scope(), token=_wrong(5))), + TypeError, + "^scope.token must be str or None, not int", + ), + (lambda: _enter(store.transaction(write=_wrong(1))), TypeError, "^write must be bool"), ] for call, error, match in refused: _assert_refused(replica, store, call, error, match, before_a_transaction=True) From de2c7d8a0a46c68b15eaa039a9196fca0467c079 Mon Sep 17 00:00:00 2001 From: Sashank Bhamidi Date: Wed, 21 Oct 2026 20:16:23 +0530 Subject: [PATCH 2/4] feat: let a memory store keep its data elsewhere --- docs/reference/store.rst | 4 +- news/28.feature.rst | 1 + src/calendaring_sync/store/memory.py | 80 +++++++++++++++++--- src/calendaring_sync/tests/test_sync_unit.py | 50 ++++++++++++ 4 files changed, 121 insertions(+), 14 deletions(-) create mode 100644 news/28.feature.rst diff --git a/docs/reference/store.rst b/docs/reference/store.rst index 4884d14..069feb9 100644 --- a/docs/reference/store.rst +++ b/docs/reference/store.rst @@ -5,9 +5,9 @@ ==================================== .. meta:: - :description: StateStore, the storage interface, and its two implementations, MemoryStore for tests and SQLiteStore for real use. + :description: StateStore, the storage interface, and its two implementations: MemoryStore, for tests or as the base of a store that saves its objects elsewhere, and SQLiteStore for real use. -A :term:`store` keeps every :term:`scope`, its records, and the :term:`change feed`. Use :class:`~calendaring_sync.store.sqlite.SQLiteStore` to keep state between runs, and :class:`~calendaring_sync.store.memory.MemoryStore` in tests. +A :term:`store` keeps every :term:`scope`, its records, and the :term:`change feed`. Use :class:`~calendaring_sync.store.sqlite.SQLiteStore` to keep state between runs, and :class:`~calendaring_sync.store.memory.MemoryStore` in tests. To save the objects themselves somewhere else, such as with :mod:`pickle` or in an object database, subclass ``MemoryStore`` as its documentation shows. .. automodule:: calendaring_sync.store.base :members: diff --git a/news/28.feature.rst b/news/28.feature.rst new file mode 100644 index 0000000..421ec0a --- /dev/null +++ b/news/28.feature.rst @@ -0,0 +1 @@ +``MemoryStore`` reads and saves its data through two methods a subclass can override, ``_load()`` and ``_save()``, so a store that keeps the same objects in a file with ``pickle`` or in an object database is a small subclass. diff --git a/src/calendaring_sync/store/memory.py b/src/calendaring_sync/store/memory.py index 3b9efda..7ffbe8f 100644 --- a/src/calendaring_sync/store/memory.py +++ b/src/calendaring_sync/store/memory.py @@ -1,7 +1,7 @@ # SPDX-FileCopyrightText: 2026 calendaring-sync contributors # SPDX-License-Identifier: AGPL-3.0-or-later -"""A :term:`store` that keeps everything in memory, for tests.""" +"""A :term:`store` that keeps everything in memory, and the base of stores that save it.""" from __future__ import annotations @@ -17,6 +17,8 @@ @dataclass class _Data: + """Everything a memory store holds: plain containers of the library's frozen dataclasses.""" + scopes: dict[str, Scope] = field(default_factory=dict) records: dict[int, Record] = field(default_factory=dict) entries: list[FeedEntry] = field(default_factory=list) @@ -36,27 +38,81 @@ def copy(self) -> _Data: class MemoryStore(StateStore): """A store held in memory and lost when the process ends. - A write transaction works on a copy of the data, which replaces the data - only when the transaction ends normally. Lookups scan every record, so - it suits test data, not large stores. Meant for tests; use - :class:`~calendaring_sync.store.sqlite.SQLiteStore` to keep state. + Each transaction works on its own view of the data: a read on the data + the last write committed, a write on a copy of it, which becomes the + committed data only when the transaction ends normally. Lookups scan + every record, so it suits test data, not large stores. Meant for tests; + use :class:`~calendaring_sync.store.sqlite.SQLiteStore` to keep state. + + To keep the data somewhere else, such as a file written with + :mod:`pickle` or an object database, subclass it and override two + methods: ``_load()`` returns the data the last write committed, and + ``_save(data)`` keeps the data a write commits. ``_load`` runs at the + start of every transaction, reads included, so a store that reads a + file there keeps what it last loaded or saved and reads the file again + only when it changed. The data is one object of plain containers + holding the library's frozen dataclasses, so it pickles. The checks, + the transactions and every lookup come from the base classes. + + If ``_save`` fails, the transaction fails and the last saved data + stays, as long as ``_save`` doesn't stop halfway: write to a new file + and then replace the old one, for example. Turn the errors of the + storage underneath into :class:`~calendaring_sync.error.StoreError` by + overriding ``_store_error``, as + :class:`~calendaring_sync.store.base.StateStore` describes; anything + else is reported as a bug. + + Examples: + A store that keeps its data as pickled bytes, read back by a second + store: + + >>> import pickle + >>> from calendaring_sync import Changed, ChangeSet, MemoryStore, Replica + >>> class PickledStore(MemoryStore): + ... def __init__(self, saved=None): + ... super().__init__() + ... self.saved = saved + ... def _load(self): + ... return pickle.loads(self.saved) if self.saved else super()._load() + ... def _save(self, data): + ... self.saved = pickle.dumps(data) + >>> first = PickledStore() + >>> listing = ChangeSet( + ... "work", None, "t1", {"a": Changed("v1", "fp-a", b"Team meeting")}, full_listing=True + ... ) + >>> with Replica(first).session( + ... "work", fingerprint_scheme="sha256", media_type="text/plain" + ... ) as session: + ... session.ingest(listing) + >>> Replica(PickledStore(first.saved)).find_record("work", "a").content + b'Team meeting' """ def __init__(self) -> None: super().__init__() - self._data = _Data() - self._saved = self._data + self._committed = _Data() + # The open transaction's view of the data. + self._data = self._committed + + def _load(self) -> _Data: + """Return the data the last committed write saved.""" + return self._committed + + def _save(self, data: _Data) -> None: + """Keep the data a write transaction commits, for later transactions to load.""" + self._committed = data def _begin(self, *, write: bool) -> None: - self._saved = self._data - if write: - self._data = self._data.copy() + data = self._load() + self._data = data.copy() if write else data def _commit(self) -> None: - self._saved = self._data + if self._writable: + self._save(self._data) def _rollback(self) -> None: - self._data = self._saved + # A write worked on a copy that's never saved, so there's nothing to undo. + pass def _get_scope(self, scope_id: str) -> Scope | None: return self._data.scopes.get(scope_id) diff --git a/src/calendaring_sync/tests/test_sync_unit.py b/src/calendaring_sync/tests/test_sync_unit.py index 9a1ed29..bad2c5c 100644 --- a/src/calendaring_sync/tests/test_sync_unit.py +++ b/src/calendaring_sync/tests/test_sync_unit.py @@ -13,6 +13,7 @@ import doctest import importlib import inspect +import pickle import pkgutil import re import runpy @@ -87,6 +88,7 @@ ) from calendaring_sync._checks import _CHECKED_WHEN_BUILT, _fields, argument_check from calendaring_sync.constants import FEED_PAGE_SIZE, ISSUES_URL, SCHEMA_VERSION +from calendaring_sync.store.memory import _Data from calendaring_sync.store.sqlite import _FEED, _META, _RECORDS, _reader from calendaring_sync.tests import benchmark @@ -2361,6 +2363,54 @@ def test_store_as_context_manager(self, make_store): with store.transaction(write=False): assert store.scopes() == [] + def test_a_memory_store_can_keep_its_data_elsewhere(self, tmp_path, clock): + path = tmp_path / "store.pickle" + store = _PickledStore(path) + replica = _edited(Replica(store, clock=clock)) + expected = _snapshot(replica, store) + # Reads load the data but never save it. + with patch.object(store, "_save", wraps=store._save) as save: + replica.records(_SCOPE) + replica.read_feed() + assert not save.called + # A write that fails saves nothing. + with pytest.raises(RuntimeError), store.transaction(): + store.put_scope(replace(_new_scope(), id=_OTHER)) + raise RuntimeError + reopened = _PickledStore(path) + again = Replica(reopened, clock=clock) + assert _snapshot(again, reopened) == expected + # A save that fails fails the write, keeps the last data and leaves the store usable. + with patch.object(reopened, "_save", side_effect=OSError("disk full")): + _assert_refused( + again, + reopened, + lambda: _sync(again, _page("T2", "T3", {"a": Removed()})), + UnexpectedStoreError, + "failed unexpectedly with OSError", + ) + _sync(again, _page("T2", "T3", {"a": Removed()})) + expected = _snapshot(again, reopened) + third = _PickledStore(path) + assert _snapshot(Replica(third, clock=clock), third) == expected + + +class _PickledStore(MemoryStore): + """A memory store that saves its data to a file with :mod:`pickle`.""" + + def __init__(self, path: Path) -> None: + super().__init__() + self._path = path + + def _load(self) -> _Data: + if not self._path.exists(): + return super()._load() + data: _Data = pickle.loads(self._path.read_bytes()) + return data + + def _save(self, data: _Data) -> None: + self._path.write_bytes(pickle.dumps(data)) + # Dataclasses the SQLite store can't map: a column holds one type, and the # first field of a nested dataclass marks whether it's there, so it must From 1081bff279ec591febef9258be8e46a4870f4b46 Mon Sep 17 00:00:00 2001 From: Sashank Bhamidi Date: Thu, 22 Oct 2026 00:36:33 +0530 Subject: [PATCH 3/4] feat: let reads run while a write transaction is open --- docs/contribute.rst | 2 +- docs/explanation/design.rst | 2 +- docs/reference/store.rst | 2 +- news/24.feature.rst | 1 + src/calendaring_sync/constants.py | 4 +- src/calendaring_sync/error.py | 2 +- src/calendaring_sync/replica.py | 4 +- src/calendaring_sync/store/base.py | 79 ++++++--- src/calendaring_sync/store/memory.py | 62 +++++-- src/calendaring_sync/store/sqlite.py | 104 +++++++++--- src/calendaring_sync/tests/benchmark.py | 31 +++- src/calendaring_sync/tests/test_sync_unit.py | 168 ++++++++++++++++--- 12 files changed, 369 insertions(+), 92 deletions(-) create mode 100644 news/24.feature.rst diff --git a/docs/contribute.rst b/docs/contribute.rst index 51a604e..289ee56 100644 --- a/docs/contribute.rst +++ b/docs/contribute.rst @@ -39,7 +39,7 @@ If your change touches ingesting, local edits, pushes or a store, run the benchm python -m calendaring_sync.tests.benchmark -It times a full listing of a large scope, a second listing over local edits and conflicts, a run of single edits, and pushes of those edits, each on a fresh SQLite store. Times differ between machines, so only the comparison means anything. Explain any slowdown beyond run-to-run noise in your pull request. ``--size`` and ``--runs`` change the scope's size and the number of runs. +It times a full listing of a large scope, a second listing over local edits and conflicts, a run of single edits, pushes of those edits, and the longest read in another thread while a listing runs, each on a fresh SQLite store. Times differ between machines, so only the comparison means anything. Explain any slowdown beyond run-to-run noise in your pull request. ``--size`` and ``--runs`` change the scope's size and the number of runs. Previewing docs =============== diff --git a/docs/explanation/design.rst b/docs/explanation/design.rst index be1e512..43deac6 100644 --- a/docs/explanation/design.rst +++ b/docs/explanation/design.rst @@ -289,4 +289,4 @@ If a calendar or account disappears from the server, the adapter raises an error Why storage is synchronous ========================== -Storage is a typed interface, ``StateStore``, with an in-memory implementation for tests and a SQLite one for real use. One :term:`store` holds many scopes. A key-value interface couldn't apply a change set and advance a token in one transaction, and an asynchronous store would push synchronous callers through an event loop. Adapters come in synchronous and asynchronous variants that share their logic. The asynchronous ones run each store call in a worker thread, since ingesting a large full listing is one long transaction and would otherwise block the event loop. +Storage is a typed interface, ``StateStore``, with an in-memory implementation for tests and a SQLite one for real use. One :term:`store` holds many scopes. A key-value interface couldn't apply a change set and advance a token in one transaction, and an asynchronous store would push synchronous callers through an event loop. Adapters come in synchronous and asynchronous variants that share their logic. The asynchronous ones run each store call in a worker thread, since ingesting a large full listing is one long transaction and would otherwise block the event loop. Reads from other threads don't wait for that transaction: they see what the store held before it began, so your application can show its records while a sync runs. diff --git a/docs/reference/store.rst b/docs/reference/store.rst index 069feb9..51a675b 100644 --- a/docs/reference/store.rst +++ b/docs/reference/store.rst @@ -7,7 +7,7 @@ .. meta:: :description: StateStore, the storage interface, and its two implementations: MemoryStore, for tests or as the base of a store that saves its objects elsewhere, and SQLiteStore for real use. -A :term:`store` keeps every :term:`scope`, its records, and the :term:`change feed`. Use :class:`~calendaring_sync.store.sqlite.SQLiteStore` to keep state between runs, and :class:`~calendaring_sync.store.memory.MemoryStore` in tests. To save the objects themselves somewhere else, such as with :mod:`pickle` or in an object database, subclass ``MemoryStore`` as its documentation shows. +A :term:`store` keeps every :term:`scope`, its records, and the :term:`change feed`. Use :class:`~calendaring_sync.store.sqlite.SQLiteStore` to keep state between runs, and :class:`~calendaring_sync.store.memory.MemoryStore` in tests. In both, a read in one thread runs while another thread writes. To save the objects themselves somewhere else, such as with :mod:`pickle` or in an object database, subclass ``MemoryStore`` as its documentation shows. .. automodule:: calendaring_sync.store.base :members: diff --git a/news/24.feature.rst b/news/24.feature.rst new file mode 100644 index 0000000..0c0accf --- /dev/null +++ b/news/24.feature.rst @@ -0,0 +1 @@ +A read no longer waits for another thread's write: in ``MemoryStore`` and ``SQLiteStore``, a transaction that only reads runs while a write is open in another thread, and sees the store as the last write left it. On SQLite, a commit still waits for the reads open at that moment, up to the store's ``timeout``, then raises ``StoreBusyError``. diff --git a/src/calendaring_sync/constants.py b/src/calendaring_sync/constants.py index d64a8d2..c6459cb 100644 --- a/src/calendaring_sync/constants.py +++ b/src/calendaring_sync/constants.py @@ -14,11 +14,11 @@ #: Default and largest number of entries one #: :meth:`~calendaring_sync.replica.Replica.read_feed` call returns. Reading #: the feed in pages keeps each read short, so it never holds up a write -#: from another process for long. +#: in another process or thread for long. FEED_PAGE_SIZE = 1000 #: Seconds :class:`~calendaring_sync.store.sqlite.SQLiteStore` waits for -#: another process to finish writing before giving up. +#: another process or thread before giving up. DEFAULT_TIMEOUT = 5.0 #: Where to report a bug, named in the message of diff --git a/src/calendaring_sync/error.py b/src/calendaring_sync/error.py index 3a7c8b6..d6f6cee 100644 --- a/src/calendaring_sync/error.py +++ b/src/calendaring_sync/error.py @@ -139,7 +139,7 @@ class StoreError(CalendaringSyncError): class StoreBusyError(StoreError): - """Another process kept the :term:`store` busy for longer than the store waits. + """Another process or thread kept the :term:`store` busy for longer than the store waits. Nothing was changed. Try again later. """ diff --git a/src/calendaring_sync/replica.py b/src/calendaring_sync/replica.py index 132adda..7a5fd92 100644 --- a/src/calendaring_sync/replica.py +++ b/src/calendaring_sync/replica.py @@ -527,8 +527,8 @@ class Replica: Every method raises :class:`TypeError` for an argument of the wrong type, before it changes anything. Any method can also raise the store's errors: - :class:`~calendaring_sync.error.StoreBusyError` when another process - kept the store busy, which is safe to retry, + :class:`~calendaring_sync.error.StoreBusyError` when another process or + thread kept the store busy, which is safe to retry, :class:`~calendaring_sync.error.StoreError` when the store can't read or write its data, and :class:`~calendaring_sync.error.UnexpectedStoreError` for a bug in the store. None of them changes anything. diff --git a/src/calendaring_sync/store/base.py b/src/calendaring_sync/store/base.py index 4485943..5fd7511 100644 --- a/src/calendaring_sync/store/base.py +++ b/src/calendaring_sync/store/base.py @@ -14,7 +14,7 @@ from collections.abc import Iterator from contextlib import AbstractContextManager, contextmanager from datetime import datetime -from threading import RLock, get_ident +from threading import Condition, Lock, local from types import TracebackType from typing import Self @@ -98,8 +98,10 @@ class StateStore(ABC): Every operation runs inside :meth:`transaction`: one that only reads runs inside any transaction, and one that writes needs a transaction opened - with ``write=True``. One thread's transaction runs to the end before - another's starts, so a store is safe to share between threads. + with ``write=True``. A store is safe to share between threads: one write + transaction runs at a time, and in a store that sets + :attr:`_reads_beside_writes`, transactions that only read run in other + threads meanwhile instead of waiting for it. Besides the errors each method lists, every operation, :meth:`transaction` and :meth:`close` can raise: @@ -107,7 +109,6 @@ class StateStore(ABC): - :class:`TypeError`, for an argument of the wrong type, before anything is read or written; a record or scope to write is checked field by field - - :class:`~calendaring_sync.error.StoreError`, if the store can't read or write its data - :class:`~calendaring_sync.error.UnexpectedStoreError`, for any other @@ -120,13 +121,25 @@ class StateStore(ABC): store implements those, along with :meth:`_begin`, :meth:`_commit` and :meth:`_rollback`, and needs no checks of its own. It turns the errors of the storage it uses into store errors by overriding :meth:`_store_error`. - Its ``__init__`` calls ``super().__init__()`` first. + Its ``__init__`` calls ``super().__init__()`` first. A hook runs in the + thread of the transaction it's for, and :meth:`_writing` says whether + that transaction writes. """ + #: Whether transactions that only read run while another thread's write + #: transaction is open. A store whose storage allows that sets it; with + #: the default, every transaction waits for the one before it to end. + _reads_beside_writes: bool = False + def __init__(self) -> None: - self._lock = RLock() - self._owner: int | None = None - self._writable = False + # Guards the count of open transactions and closing; close() waits on it. + self._state = Condition() + # Held by the write transaction, or by every transaction of a store + # that doesn't read beside a write. + self._turn = Lock() + # This thread's open transaction: whether it writes, or None. + self._thread = local() + self._open = 0 self._closed = False @contextmanager @@ -138,22 +151,28 @@ def transaction(self, *, write: bool = True) -> Iterator[None]: Args: write: ``False`` for a block that only reads. Such a block - can't write, and a store may run it without blocking other - processes' writes. + can't write, and a store may run it beside a write in + another thread or process. Raises: TransactionError: If a transaction is already open in this thread. - StoreBusyError: If another process kept the store busy for too long. + StoreBusyError: If another process or thread kept the store busy + for too long. StoreError: If the store was closed, or can't start or keep the transaction. """ - with self._lock: + with self._state: if self._closed: raise StoreError("The store is closed") if self._in_transaction(): raise TransactionError("A store transaction is already open in this thread") - self._owner, self._writable = get_ident(), write + self._open += 1 + waits = write or not self._reads_beside_writes + try: + if waits: + self._turn.acquire() try: + self._thread.write = write with self._errors(): self._begin(write=write) yield @@ -164,17 +183,27 @@ def transaction(self, *, write: bool = True) -> Iterator[None]: self._rollback() raise finally: - self._owner = None + self._thread.write = None + if waits: + self._turn.release() + finally: + with self._state: + self._open -= 1 + self._state.notify_all() def _in_transaction(self) -> bool: - """Whether this thread holds the store's transaction.""" - return self._owner == get_ident() + """Whether this thread has a transaction open.""" + return getattr(self._thread, "write", None) is not None + + def _writing(self) -> bool: + """Whether this thread's open transaction writes.""" + return getattr(self._thread, "write", None) is True def _operation(self, *, write: bool) -> AbstractContextManager[None]: """Check that this thread's transaction allows the operation, then guard its errors.""" if not self._in_transaction(): raise TransactionRequiredError("Store operations run inside store.transaction()") - if write and not self._writable: + if write and not self._writing(): raise TransactionPermissionError( "A transaction opened with write=False can't change the store" ) @@ -458,20 +487,24 @@ def trim_entries(self, through: int) -> None: def close(self) -> None: """Close the store. Every later transaction raises ``StoreError``. - Closing a closed store does nothing. + Transactions open in other threads end first: closing waits for + them. Closing a closed store does nothing. Raises: TransactionError: If this thread has a transaction open. StoreError: If the store couldn't release what it holds open. It counts as closed anyway. """ - with self._lock: + with self._state: if self._in_transaction(): raise TransactionError("Close the store after its transaction ends") - if not self._closed: - self._closed = True - with self._errors(): - self._close() + if self._closed: + return + # New transactions are refused from here; the open ones end first. + self._closed = True + self._state.wait_for(lambda: self._open == 0) + with self._errors(): + self._close() def __enter__(self) -> Self: return self diff --git a/src/calendaring_sync/store/memory.py b/src/calendaring_sync/store/memory.py index 7ffbe8f..577c4fb 100644 --- a/src/calendaring_sync/store/memory.py +++ b/src/calendaring_sync/store/memory.py @@ -7,6 +7,8 @@ from dataclasses import dataclass, field, replace from datetime import datetime +from threading import local +from typing import cast from calendaring_sync._checks import check_argument_types from calendaring_sync.objects.feed import FeedEntry, FeedEntryKind @@ -40,19 +42,28 @@ class MemoryStore(StateStore): Each transaction works on its own view of the data: a read on the data the last write committed, a write on a copy of it, which becomes the - committed data only when the transaction ends normally. Lookups scan - every record, so it suits test data, not large stores. Meant for tests; - use :class:`~calendaring_sync.store.sqlite.SQLiteStore` to keep state. + committed data only when the transaction ends normally. So a read in + one thread runs while another thread writes, and sees the data as it + was when the read started. Lookups scan every record, so it suits test + data, not large stores. Meant for tests; use + :class:`~calendaring_sync.store.sqlite.SQLiteStore` to keep state. To keep the data somewhere else, such as a file written with - :mod:`pickle` or an object database, subclass it and override two - methods: ``_load()`` returns the data the last write committed, and - ``_save(data)`` keeps the data a write commits. ``_load`` runs at the - start of every transaction, reads included, so a store that reads a - file there keeps what it last loaded or saved and reads the file again - only when it changed. The data is one object of plain containers - holding the library's frozen dataclasses, so it pickles. The checks, - the transactions and every lookup come from the base classes. + :mod:`pickle` or an object database, subclass it and override the two + methods that read and keep it: ``_load()`` returns the data the last + write committed, and ``_save(data)`` keeps the data a write commits. + The data is one object of plain containers holding the library's + frozen dataclasses, so it pickles. The checks, the transactions and + every lookup come from the base classes. + + ``_load`` runs at the start of every transaction, reads included, and + can run in one thread while ``_save`` runs in another. So a store that + keeps its data in a file reads the file once, when it opens, and + leaves ``_load`` as it is: it passes what it read to + ``super()._save()``, and its own ``_save`` writes the file and then + calls ``super()._save()``, so reads come from memory. A store whose + ``_load`` can't run while ``_save`` does sets ``_reads_beside_writes`` + to ``False``. If ``_save`` fails, the transaction fails and the last saved data stays, as long as ``_save`` doesn't stop halfway: write to a new file @@ -88,11 +99,19 @@ class MemoryStore(StateStore): b'Team meeting' """ + _reads_beside_writes = True + def __init__(self) -> None: super().__init__() self._committed = _Data() - # The open transaction's view of the data. - self._data = self._committed + # Each thread's open transaction's view of the data. + self._views = local() + + @property + def _data(self) -> _Data: + """This thread's transaction's view of the data.""" + data: _Data = self._views.data + return data def _load(self) -> _Data: """Return the data the last committed write saved.""" @@ -104,15 +123,22 @@ def _save(self, data: _Data) -> None: def _begin(self, *, write: bool) -> None: data = self._load() - self._data = data.copy() if write else data + self._views.data = data.copy() if write else data def _commit(self) -> None: - if self._writable: - self._save(self._data) + data = self._end_view() + if self._writing(): + # _begin set this thread's view, and only _commit or _rollback drops it. + self._save(cast(_Data, data)) def _rollback(self) -> None: - # A write worked on a copy that's never saved, so there's nothing to undo. - pass + # A write worked on a copy that's never saved, so dropping it undoes everything. + self._end_view() + + def _end_view(self) -> _Data | None: + """Drop and return this thread's view, so an idle thread keeps no old copy of the data.""" + data: _Data | None = vars(self._views).pop("data", None) + return data def _get_scope(self, scope_id: str) -> Scope | None: return self._data.scopes.get(scope_id) diff --git a/src/calendaring_sync/store/sqlite.py b/src/calendaring_sync/store/sqlite.py index f4d0a44..749f0fc 100644 --- a/src/calendaring_sync/store/sqlite.py +++ b/src/calendaring_sync/store/sqlite.py @@ -13,6 +13,7 @@ from datetime import datetime from enum import StrEnum from operator import attrgetter +from threading import Lock, local from types import UnionType from typing import Any, NamedTuple, TypeVar, Union, cast, get_args, get_origin, get_type_hints @@ -308,25 +309,43 @@ def _load(reader: _Reader, row: sqlite3.Row) -> Any: return reader.cls(**values) +# The paths at which SQLite opens a database private to one connection: in +# memory, or in a temporary file. +_PRIVATE = (":memory:", "") + + +def _connect(path: str | os.PathLike[str], timeout: float) -> sqlite3.Connection: + """Open a connection to the store's file the way every one of its connections is opened.""" + conn = sqlite3.connect(path, timeout=timeout, isolation_level=None, check_same_thread=False) + conn.row_factory = sqlite3.Row + return conn + + @check_argument_types class SQLiteStore(StateStore): """A store in a SQLite database file, shared between threads and processes. - One connection serves every thread, one transaction at a time. Writes - take SQLite's write lock for their whole transaction, so two processes - on one file never interleave. Reads don't take that lock, but an open - read still makes another process's commit wait, so the store keeps - reads short. A write waits up to ``timeout`` seconds for another - process's transaction to end. - - Setting the file to SQLite's WAL journal mode lets one process's reads - and another's writes overlap. The mode is kept in the file, so set it - once with ``PRAGMA journal_mode = WAL``; WAL doesn't work on a network - filesystem. + Writes take one connection, one transaction at a time, and SQLite's write + lock for their whole transaction, so two processes on one file never + interleave. Reads take a second connection, one at a time, so a read in + one thread doesn't wait for another thread's write. For that, a write + keeps what it changes in memory until it commits: a full listing takes + about as much memory again as its change set holds. A commit still waits + for the reads open at that moment, in this process or another, so the + store keeps reads short. A write waits up to ``timeout`` seconds for + them, then raises :class:`~calendaring_sync.error.StoreBusyError`, and + a read that starts during a commit waits for it the same way. A + database at ``":memory:"`` or an empty path belongs to one connection, + so it has only that one, and its transactions take turns. + + Setting the file to SQLite's WAL journal mode lets reads and a commit + overlap too. The mode is kept in the file, so set it once with + ``PRAGMA journal_mode = WAL``; WAL doesn't work on a network filesystem. Args: path: The database file. It's created, with the schema, if missing. - timeout: Seconds to wait for another process before giving up. + timeout: Seconds to wait for another process or thread before + giving up. Raises: TypeError: If ``path`` or ``timeout`` has the wrong type. @@ -335,21 +354,43 @@ class SQLiteStore(StateStore): StoreError: If the file can't be opened or isn't a store. """ + _reads_beside_writes = True + def __init__(self, path: str | os.PathLike[str], *, timeout: float = DEFAULT_TIMEOUT) -> None: super().__init__() + self._reader: sqlite3.Connection | None = None + # Read transactions take turns on the reader's connection; each thread + # notes whether it has the turn, so only the thread that took it gives it back. + self._read_turn = Lock() + self._turn_taken = local() with self._errors(): - self._conn = sqlite3.connect( - path, timeout=timeout, isolation_level=None, check_same_thread=False - ) - self._conn.row_factory = sqlite3.Row + self._writer = _connect(path, timeout) try: with self._errors(): - self._conn.execute("PRAGMA foreign_keys = ON") + self._writer.execute("PRAGMA foreign_keys = ON") self._ensure_schema() + if os.fspath(path) in _PRIVATE: + # A second connection would open an empty database of its own. + self._reads_beside_writes = False + else: + self._reader = _connect(path, timeout) + self._reader.execute("PRAGMA query_only = ON") + # A write that outgrows SQLite's page cache would otherwise + # write into the file before it commits, which locks the + # reads out until the commit. + self._writer.execute("PRAGMA cache_spill = OFF") except BaseException: - self._conn.close() + self._close() raise + @property + def _conn(self) -> sqlite3.Connection: + """The connection this thread's transaction uses: the reader's for a read.""" + if self._in_transaction() and self._reads_apart(): + # _reads_apart() found the reader's connection open. + return cast(sqlite3.Connection, self._reader) + return self._writer + def _store_error(self, exc: Exception) -> CalendaringSyncError: if not isinstance(exc, sqlite3.Error): return super()._store_error(exc) @@ -391,17 +432,38 @@ def _ensure_schema(self) -> None: ) def _close(self) -> None: - self._conn.close() + try: + if self._reader is not None: + self._reader.close() + finally: + self._writer.close() def _begin(self, *, write: bool) -> None: + if self._reads_apart(): + self._read_turn.acquire() + self._turn_taken.taken = True self._conn.execute("BEGIN IMMEDIATE" if write else "BEGIN") def _commit(self) -> None: self._conn.execute("COMMIT") + self._end_read() def _rollback(self) -> None: - if self._conn.in_transaction: - self._conn.execute("ROLLBACK") + try: + if self._conn.in_transaction: + self._conn.execute("ROLLBACK") + finally: + self._end_read() + + def _reads_apart(self) -> bool: + """Whether this thread's transaction reads on the reader's connection.""" + return self._reader is not None and not self._writing() + + def _end_read(self) -> None: + """Give the reader's connection to the next read, if this thread's read took it.""" + if getattr(self._turn_taken, "taken", False): + self._turn_taken.taken = False + self._read_turn.release() def _select( self, cls: type[_Row], table: str, clause: str, params: tuple[Any, ...] = () diff --git a/src/calendaring_sync/tests/benchmark.py b/src/calendaring_sync/tests/benchmark.py index ac89d3a..71031fe 100644 --- a/src/calendaring_sync/tests/benchmark.py +++ b/src/calendaring_sync/tests/benchmark.py @@ -2,7 +2,7 @@ # SPDX-License-Identifier: AGPL-3.0-or-later """ -Benchmarks for the work a sync does most: full listings, local edits and pushes. +Benchmarks for the work a sync does most: full listings, local edits, pushes, and reads during them. Run ``python -m calendaring_sync.tests.benchmark`` on ``main`` and on your branch, on the same machine, and compare the two. Times differ between @@ -22,8 +22,10 @@ import time from collections import Counter from collections.abc import Callable, Iterator +from concurrent.futures import ThreadPoolExecutor from contextlib import AbstractContextManager, contextmanager from pathlib import Path +from threading import Event from calendaring_sync import ( Applied, @@ -153,6 +155,32 @@ def pushes(size: int) -> float: return _push_edits(replica) +def read_during_listing(size: int) -> float: + """The longest one read in another thread takes while a listing of ``size`` objects ingests.""" + with _replica() as replica: + _ingest(replica, _listing(size, None, _FIRST)) + key = replica.records(_SCOPE)[0].local_key + reading, done = Event(), Event() + + def read() -> float: + longest = 0.0 + reading.set() + while not done.is_set(): + started = time.perf_counter() + replica.record(key) + longest = max(longest, time.perf_counter() - started) + return longest + + with ThreadPoolExecutor(1) as pool: + reader = pool.submit(read) + reading.wait() + try: + _ingest(replica, _listing(size, _FIRST, _SECOND, changed=size)) + finally: + done.set() + return reader.result() + + def states_over_edits(size: int) -> Counter[RecordState]: """How many records end in each state in ``listing_over_edits``, to check what it measures.""" with _replica() as replica: @@ -175,6 +203,7 @@ def entries_after_pushes(size: int) -> Counter[FeedEntryKind]: "listing over edits and conflicts": listing_over_edits, f"single edits to one in {EDITED_SHARE} objects": single_edits, f"pushes of one in {EDITED_SHARE} objects": pushes, + "longest read during a full listing": read_during_listing, } diff --git a/src/calendaring_sync/tests/test_sync_unit.py b/src/calendaring_sync/tests/test_sync_unit.py index bad2c5c..29d5727 100644 --- a/src/calendaring_sync/tests/test_sync_unit.py +++ b/src/calendaring_sync/tests/test_sync_unit.py @@ -11,6 +11,7 @@ import dataclasses import doctest +import gc import importlib import inspect import pickle @@ -19,6 +20,7 @@ import runpy import sqlite3 import sys +import weakref from collections import Counter from collections.abc import Callable, Iterator, Mapping from concurrent.futures import ThreadPoolExecutor @@ -27,6 +29,7 @@ from datetime import UTC, datetime, timedelta from functools import partial from pathlib import Path +from threading import Event from typing import Any, Literal from unittest.mock import patch @@ -101,6 +104,8 @@ _START = datetime(2026, 10, 1, tzinfo=UTC) # How long a test store waits for another, where the wait is meant to run out. _SHORT_TIMEOUT = 0.1 +# How long a test waits for another thread before it counts as stuck. +_WAIT = 5.0 class _Clock: @@ -242,6 +247,41 @@ def _write_with_scope(store: StateStore) -> Iterator[None]: yield +def _add_scope(store: StateStore) -> None: + """Write the test scope to a store, in a transaction of its own.""" + with _write_with_scope(store): + pass + + +def _scope_ids(store: StateStore) -> list[str]: + """The ids of the scopes a store holds, read in a transaction of their own.""" + with store.transaction(write=False): + return [scope.id for scope in store.scopes()] + + +@contextmanager +def _held_open( + store: StateStore, *, write: bool, work: Callable[[], object] = lambda: None +) -> Iterator[None]: + """A transaction that runs ``work`` in another thread, then stays open for the block.""" + started, finish = Event(), Event() + + def hold() -> None: + with store.transaction(write=write): + work() + started.set() + finish.wait(_WAIT) + + with ThreadPoolExecutor(1) as pool: + held = pool.submit(hold) + assert started.wait(_WAIT) + try: + yield + finally: + finish.set() + held.result(_WAIT) + + @contextmanager def _busy_pair(db_path: Path) -> Iterator[tuple[SQLiteStore, SQLiteStore]]: """Two stores on one file; the second gives up quickly when the first holds it.""" @@ -2217,6 +2257,66 @@ def test_transactions_do_not_nest(self, replica, store): with store.transaction(write=False), pytest.raises(TransactionError, match="already open"): replica.scopes() + @pytest.mark.parametrize( + "open_store, beside", + [ + (lambda path: MemoryStore(), True), + (lambda path: SQLiteStore(path), True), + (lambda path: _OneAtATime(), False), + (lambda path: SQLiteStore(":memory:"), False), + ], + ids=["memory", "sqlite", "memory one at a time", "sqlite in memory"], + ) + def test_a_read_runs_beside_a_write(self, db_path, open_store, beside): + with open_store(db_path) as store: + _add_scope(store) + + def write() -> None: + store.put_scope(replace(_new_scope(), id=_OTHER)) + # More than SQLite's page cache holds, as a large listing writes. + big = replace(_local_record(RecordState.SYNCED), content=b"c" * 4_000_000) + store.insert_record(big) + + with ThreadPoolExecutor(1) as pool, _held_open(store, write=True, work=write): + reader = pool.submit(_scope_ids, store) + if beside: + # It sees what was committed, not the write still open. + assert reader.result(_WAIT) == [_SCOPE] + else: + with pytest.raises(TimeoutError): + reader.result(_SHORT_TIMEOUT) + assert reader.result() == ([_SCOPE] if beside else [_SCOPE, _OTHER]) + # The read left the write's work alone. + assert _scope_ids(store) == [_SCOPE, _OTHER] + + def test_a_write_isnt_starved_by_reads(self, replica): + _seeded(replica) + done = Event() + + def read() -> None: + while not done.is_set(): + replica.records(_SCOPE) + + with ThreadPoolExecutor(3) as pool: + readers = [pool.submit(read) for _ in range(3)] + try: + for i in range(20): + _create(replica, f"n{i}") + finally: + done.set() + for reader in readers: + reader.result(_WAIT) + assert len(replica.records(_SCOPE)) == 22 + + def test_close_waits_for_open_transactions(self, store): + with ThreadPoolExecutor(1) as pool, _held_open(store, write=False): + closing = pool.submit(store.close) + with pytest.raises(TimeoutError): + closing.result(_SHORT_TIMEOUT) + closing.result() + with pytest.raises(StoreError, match="closed"): + _enter(store.transaction(write=False)) + def test_threads_share_one_store(self, replica): _seeded(replica) @@ -2263,8 +2363,7 @@ def test_closing_inside_a_transaction_is_refused(self, store): with _write_with_scope(store): with pytest.raises(TransactionError, match="after its transaction"): store.close() - with store.transaction(write=False): - assert [s.id for s in store.scopes()] == [_SCOPE] + assert _scope_ids(store) == [_SCOPE] def test_object_ids_are_unique_per_scope(self, replica, store): _seeded(replica) @@ -2329,9 +2428,10 @@ def test_updating_a_missing_record_raises(self, store): with store.transaction(), pytest.raises(UnknownRecordError): store.update_record(_local_record(RecordState.SYNCED, local_key=999)) + @pytest.mark.parametrize("write", [True, False], ids=["write", "read"]) @pytest.mark.parametrize("hook", ["_begin", "_commit", "_scopes"]) def test_a_failure_inside_the_store_is_unexpected_and_keeps_nothing( - self, store, monkeypatch, hook + self, store, monkeypatch, hook, write ): bug = RuntimeError(f"{hook} has a bug") @@ -2340,13 +2440,15 @@ def fail(*args: object, **kwargs: object) -> None: monkeypatch.setattr(store, hook, fail) with pytest.raises(UnexpectedStoreError, match=re.escape(ISSUES_URL)) as raised: - with _write_with_scope(store): + with _write_with_scope(store) if write else store.transaction(write=False): store.scopes() assert raised.value.__cause__ is bug assert not isinstance(raised.value, StoreError) monkeypatch.undo() - with store.transaction(write=False): - assert store.scopes() == [] + # Neither kind of transaction is left waiting on the one that failed. + assert _scope_ids(store) == [] + with store.transaction(): + pass def test_interrupts_pass_through_unchanged(self, store, monkeypatch): def interrupt() -> None: @@ -2360,15 +2462,24 @@ def interrupt() -> None: def test_store_as_context_manager(self, make_store): with make_store() as store: - with store.transaction(write=False): - assert store.scopes() == [] + assert _scope_ids(store) == [] + + def test_a_memory_store_lets_go_of_old_data(self): + store = MemoryStore() + with ThreadPoolExecutor(1) as pool: + # The thread stays alive, idle, after its read. + pool.submit(_scope_ids, store).result() + old = weakref.ref(store._load()) + _add_scope(store) + gc.collect() + assert old() is None def test_a_memory_store_can_keep_its_data_elsewhere(self, tmp_path, clock): path = tmp_path / "store.pickle" store = _PickledStore(path) replica = _edited(Replica(store, clock=clock)) expected = _snapshot(replica, store) - # Reads load the data but never save it. + # Reads never save. with patch.object(store, "_save", wraps=store._save) as save: replica.records(_SCOPE) replica.read_feed() @@ -2395,21 +2506,28 @@ def test_a_memory_store_can_keep_its_data_elsewhere(self, tmp_path, clock): assert _snapshot(Replica(third, clock=clock), third) == expected +class _OneAtATime(MemoryStore): + """A memory store that keeps the default: every transaction waits for the one before it.""" + + _reads_beside_writes = False + + class _PickledStore(MemoryStore): - """A memory store that saves its data to a file with :mod:`pickle`.""" + """A memory store that keeps its data in a file with :mod:`pickle`, read when it opens.""" def __init__(self, path: Path) -> None: super().__init__() self._path = path - - def _load(self) -> _Data: - if not self._path.exists(): - return super()._load() - data: _Data = pickle.loads(self._path.read_bytes()) - return data + if path.exists(): + super()._save(pickle.loads(path.read_bytes())) def _save(self, data: _Data) -> None: - self._path.write_bytes(pickle.dumps(data)) + # Written beside the file, then moved over it, so a failure halfway + # leaves the last saved data. + written = self._path.with_suffix(".new") + written.write_bytes(pickle.dumps(data)) + written.replace(self._path) + super()._save(data) # Dataclasses the SQLite store can't map: a column holds one type, and the @@ -2447,8 +2565,7 @@ def test_busy_file_raises_store_busy_error_and_stays_usable(self, db_path): _enter(waiter.transaction()) with waiter.transaction(): waiter.put_scope(_new_scope()) - with holder.transaction(write=False): - assert [s.id for s in holder.scopes()] == [_SCOPE] + assert _scope_ids(holder) == [_SCOPE] def test_ingest_while_busy_changes_nothing_and_can_be_retried(self, db_path, clock): with _busy_pair(db_path) as (holder, waiter): @@ -2572,6 +2689,15 @@ def test_a_dataclass_the_store_cant_map_is_refused(self, cls, message): with pytest.raises(TypeError, match=message): _reader(cls) + def test_a_read_held_open_past_the_timeout_makes_a_write_busy(self, db_path): + with SQLiteStore(db_path, timeout=_SHORT_TIMEOUT) as store: + with _held_open(store, write=False, work=store.scopes): + with pytest.raises(StoreBusyError), _write_with_scope(store): + pass + assert _scope_ids(store) == [] + _add_scope(store) + assert _scope_ids(store) == [_SCOPE] + def test_interrupted_ingest_leaves_the_token_after_reopening(self, db_path, clock): with SQLiteStore(db_path) as store: replica = _seeded(Replica(store, clock=clock)) @@ -2595,8 +2721,8 @@ def version(store: SQLiteStore) -> int: return 0 if len(looks) == 1 else real(store) monkeypatch.setattr(SQLiteStore, "_version", version) - with SQLiteStore(db_path) as store, store.transaction(write=False): - assert store.scopes() == [] + with SQLiteStore(db_path) as store: + assert _scope_ids(store) == [] assert len(looks) == 3 def test_newer_schema_is_refused(self, db_path): From 866376a2e842cd43bcb7b22c0f0ecce0899fef4f Mon Sep 17 00:00:00 2001 From: Sashank Bhamidi Date: Thu, 22 Oct 2026 01:33:18 +0530 Subject: [PATCH 4/4] fix: let a commit wait for reads instead of failing, and skip both in WAL mode Co-authored-by: Copilot <198982749+Copilot@users.noreply.github.com> --- docs/explanation/design.rst | 2 +- news/24.feature.rst | 2 +- news/27.feature.rst | 2 +- src/calendaring_sync/constants.py | 2 +- src/calendaring_sync/error.py | 2 +- src/calendaring_sync/replica.py | 4 +- src/calendaring_sync/store/base.py | 3 +- src/calendaring_sync/store/memory.py | 9 +-- src/calendaring_sync/store/sqlite.py | 65 ++++++++++++-------- src/calendaring_sync/tests/test_sync_unit.py | 56 ++++++++++++----- 10 files changed, 95 insertions(+), 52 deletions(-) diff --git a/docs/explanation/design.rst b/docs/explanation/design.rst index 43deac6..b573581 100644 --- a/docs/explanation/design.rst +++ b/docs/explanation/design.rst @@ -289,4 +289,4 @@ If a calendar or account disappears from the server, the adapter raises an error Why storage is synchronous ========================== -Storage is a typed interface, ``StateStore``, with an in-memory implementation for tests and a SQLite one for real use. One :term:`store` holds many scopes. A key-value interface couldn't apply a change set and advance a token in one transaction, and an asynchronous store would push synchronous callers through an event loop. Adapters come in synchronous and asynchronous variants that share their logic. The asynchronous ones run each store call in a worker thread, since ingesting a large full listing is one long transaction and would otherwise block the event loop. Reads from other threads don't wait for that transaction: they see what the store held before it began, so your application can show its records while a sync runs. +Storage is a typed interface, ``StateStore``, with an in-memory implementation for tests and a SQLite one for real use. One :term:`store` holds many scopes. A key-value interface couldn't apply a change set and advance a token in one transaction, and an asynchronous store would push synchronous callers through an event loop. Adapters come in synchronous and asynchronous variants that share their logic. The asynchronous ones run each store call in a worker thread, since ingesting a large full listing is one long transaction and would otherwise block the event loop. Reads from other threads don't wait for that transaction, at most for its commit: until then they see what the store held before it began, so your application can show its records while a sync runs. diff --git a/news/24.feature.rst b/news/24.feature.rst index 0c0accf..087de2e 100644 --- a/news/24.feature.rst +++ b/news/24.feature.rst @@ -1 +1 @@ -A read no longer waits for another thread's write: in ``MemoryStore`` and ``SQLiteStore``, a transaction that only reads runs while a write is open in another thread, and sees the store as the last write left it. On SQLite, a commit still waits for the reads open at that moment, up to the store's ``timeout``, then raises ``StoreBusyError``. +A read no longer waits for another thread's write: in ``MemoryStore`` and ``SQLiteStore``, a transaction that only reads runs while a write is open in another thread. In SQLite's default journal mode the commit still waits for such reads, and a write keeps its changes in memory until it commits, which for a large listing is about as much again as its change set. diff --git a/news/27.feature.rst b/news/27.feature.rst index 5d03446..1bc5fb5 100644 --- a/news/27.feature.rst +++ b/news/27.feature.rst @@ -1 +1 @@ -Every public method of ``Replica``, ``Session`` and the stores, and every value an adapter or application builds, now checks its argument types and raises ``TypeError`` for a wrong one, the same way on every store. Before, a wrong type could be stored as it was by one store and converted by another, such as ``delete_record("1")``, which deleted record 1 only on SQLite. +Every public method of ``Replica``, ``Session`` and the stores, and every value an adapter or application builds, now checks its argument types and raises ``TypeError`` for a wrong one the same way on every store; a subclass of ``str`` or ``int``, such as an enum, counts as wrong. Before, a wrong type could be stored as it was by one store and converted by another, such as ``delete_record("1")``, which deleted record 1 only on SQLite. diff --git a/src/calendaring_sync/constants.py b/src/calendaring_sync/constants.py index c6459cb..705b67a 100644 --- a/src/calendaring_sync/constants.py +++ b/src/calendaring_sync/constants.py @@ -18,7 +18,7 @@ FEED_PAGE_SIZE = 1000 #: Seconds :class:`~calendaring_sync.store.sqlite.SQLiteStore` waits for -#: another process or thread before giving up. +#: another process before giving up. DEFAULT_TIMEOUT = 5.0 #: Where to report a bug, named in the message of diff --git a/src/calendaring_sync/error.py b/src/calendaring_sync/error.py index d6f6cee..3a7c8b6 100644 --- a/src/calendaring_sync/error.py +++ b/src/calendaring_sync/error.py @@ -139,7 +139,7 @@ class StoreError(CalendaringSyncError): class StoreBusyError(StoreError): - """Another process or thread kept the :term:`store` busy for longer than the store waits. + """Another process kept the :term:`store` busy for longer than the store waits. Nothing was changed. Try again later. """ diff --git a/src/calendaring_sync/replica.py b/src/calendaring_sync/replica.py index 7a5fd92..132adda 100644 --- a/src/calendaring_sync/replica.py +++ b/src/calendaring_sync/replica.py @@ -527,8 +527,8 @@ class Replica: Every method raises :class:`TypeError` for an argument of the wrong type, before it changes anything. Any method can also raise the store's errors: - :class:`~calendaring_sync.error.StoreBusyError` when another process or - thread kept the store busy, which is safe to retry, + :class:`~calendaring_sync.error.StoreBusyError` when another process + kept the store busy, which is safe to retry, :class:`~calendaring_sync.error.StoreError` when the store can't read or write its data, and :class:`~calendaring_sync.error.UnexpectedStoreError` for a bug in the store. None of them changes anything. diff --git a/src/calendaring_sync/store/base.py b/src/calendaring_sync/store/base.py index 5fd7511..c8cd474 100644 --- a/src/calendaring_sync/store/base.py +++ b/src/calendaring_sync/store/base.py @@ -156,8 +156,7 @@ def transaction(self, *, write: bool = True) -> Iterator[None]: Raises: TransactionError: If a transaction is already open in this thread. - StoreBusyError: If another process or thread kept the store busy - for too long. + StoreBusyError: If another process kept the store busy for too long. StoreError: If the store was closed, or can't start or keep the transaction. """ diff --git a/src/calendaring_sync/store/memory.py b/src/calendaring_sync/store/memory.py index 577c4fb..ae9a7c5 100644 --- a/src/calendaring_sync/store/memory.py +++ b/src/calendaring_sync/store/memory.py @@ -74,8 +74,8 @@ class MemoryStore(StateStore): else is reported as a bug. Examples: - A store that keeps its data as pickled bytes, read back by a second - store: + A store that keeps its data as pickled bytes, read once when it + opens, and a second store opened from those bytes: >>> import pickle >>> from calendaring_sync import Changed, ChangeSet, MemoryStore, Replica @@ -83,10 +83,11 @@ class MemoryStore(StateStore): ... def __init__(self, saved=None): ... super().__init__() ... self.saved = saved - ... def _load(self): - ... return pickle.loads(self.saved) if self.saved else super()._load() + ... if saved: + ... super()._save(pickle.loads(saved)) ... def _save(self, data): ... self.saved = pickle.dumps(data) + ... super()._save(data) >>> first = PickledStore() >>> listing = ChangeSet( ... "work", None, "t1", {"a": Changed("v1", "fp-a", b"Team meeting")}, full_listing=True diff --git a/src/calendaring_sync/store/sqlite.py b/src/calendaring_sync/store/sqlite.py index 749f0fc..de64f7d 100644 --- a/src/calendaring_sync/store/sqlite.py +++ b/src/calendaring_sync/store/sqlite.py @@ -328,24 +328,25 @@ class SQLiteStore(StateStore): Writes take one connection, one transaction at a time, and SQLite's write lock for their whole transaction, so two processes on one file never interleave. Reads take a second connection, one at a time, so a read in - one thread doesn't wait for another thread's write. For that, a write - keeps what it changes in memory until it commits: a full listing takes - about as much memory again as its change set holds. A commit still waits - for the reads open at that moment, in this process or another, so the - store keeps reads short. A write waits up to ``timeout`` seconds for - them, then raises :class:`~calendaring_sync.error.StoreBusyError`, and - a read that starts during a commit waits for it the same way. A - database at ``":memory:"`` or an empty path belongs to one connection, - so it has only that one, and its transactions take turns. - - Setting the file to SQLite's WAL journal mode lets reads and a commit - overlap too. The mode is kept in the file, so set it once with - ``PRAGMA journal_mode = WAL``; WAL doesn't work on a network filesystem. + one thread doesn't wait for another thread's write. A database at + ``":memory:"`` or an empty path belongs to one connection, so it has + only that one, and its transactions take turns. + + In SQLite's default journal mode, a commit locks reads out of the file. + So it waits for the reads open in this process, and reads in another + process make it wait up to ``timeout`` seconds, then raise + :class:`~calendaring_sync.error.StoreBusyError`; the store keeps reads + short for that. A write also keeps what it changes in memory until it + commits, so that reads aren't locked out before then: a full listing + takes about as much memory again as its change set holds. In WAL mode + reads and a commit don't lock each other out, so a commit waits for no + read and a write keeps nothing extra in memory. The mode is kept in the + file, so set it once with ``PRAGMA journal_mode = WAL``; WAL doesn't + work on a network filesystem. Args: path: The database file. It's created, with the schema, if missing. - timeout: Seconds to wait for another process or thread before - giving up. + timeout: Seconds to wait for another process before giving up. Raises: TypeError: If ``path`` or ``timeout`` has the wrong type. @@ -359,8 +360,13 @@ class SQLiteStore(StateStore): def __init__(self, path: str | os.PathLike[str], *, timeout: float = DEFAULT_TIMEOUT) -> None: super().__init__() self._reader: sqlite3.Connection | None = None - # Read transactions take turns on the reader's connection; each thread - # notes whether it has the turn, so only the thread that took it gives it back. + # Whether a write locks reads out of the file: in the default journal + # mode it does, in WAL mode it doesn't. + self._locks_reads_out = False + # Read transactions take turns on the reader's connection, and where a + # commit locks reads out, it takes a turn too, so no read in this + # process is open while it runs. Each thread notes whether it has the + # turn, so only the thread that took it gives it back. self._read_turn = Lock() self._turn_taken = local() with self._errors(): @@ -375,10 +381,13 @@ def __init__(self, path: str | os.PathLike[str], *, timeout: float = DEFAULT_TIM else: self._reader = _connect(path, timeout) self._reader.execute("PRAGMA query_only = ON") - # A write that outgrows SQLite's page cache would otherwise - # write into the file before it commits, which locks the - # reads out until the commit. - self._writer.execute("PRAGMA cache_spill = OFF") + mode = self._writer.execute("PRAGMA journal_mode").fetchone()[0] + self._locks_reads_out = mode != "wal" + if self._locks_reads_out: + # A write that outgrows SQLite's page cache would otherwise + # write into the file before it commits, which locks the + # reads out until the commit. + self._writer.execute("PRAGMA cache_spill = OFF") except BaseException: self._close() raise @@ -440,11 +449,14 @@ def _close(self) -> None: def _begin(self, *, write: bool) -> None: if self._reads_apart(): - self._read_turn.acquire() - self._turn_taken.taken = True + self._take_read_turn() self._conn.execute("BEGIN IMMEDIATE" if write else "BEGIN") def _commit(self) -> None: + if self._locks_reads_out and self._writing(): + # The reads open in this process end first, and new ones wait, so + # only another process can keep the commit busy. + self._take_read_turn() self._conn.execute("COMMIT") self._end_read() @@ -459,8 +471,13 @@ def _reads_apart(self) -> bool: """Whether this thread's transaction reads on the reader's connection.""" return self._reader is not None and not self._writing() + def _take_read_turn(self) -> None: + """Take the read turn for this thread's transaction, after the reads ahead of it.""" + self._read_turn.acquire() + self._turn_taken.taken = True + def _end_read(self) -> None: - """Give the reader's connection to the next read, if this thread's read took it.""" + """Give the read turn back, if this thread's transaction took it.""" if getattr(self._turn_taken, "taken", False): self._turn_taken.taken = False self._read_turn.release() diff --git a/src/calendaring_sync/tests/test_sync_unit.py b/src/calendaring_sync/tests/test_sync_unit.py index 29d5727..4ec857b 100644 --- a/src/calendaring_sync/tests/test_sync_unit.py +++ b/src/calendaring_sync/tests/test_sync_unit.py @@ -657,10 +657,16 @@ def test_benchmarks_run_and_measure_what_they_say(self, monkeypatch, capsys): def test_every_entry_point_checks_its_types(self): for cls in (Replica, Session, StateStore, MemoryStore, SQLiteStore): for name, func in vars(cls).items(): - if inspect.isfunction(func) and (not name.startswith("_") or name == "__init__"): - assert getattr(func, "_checks_arguments", False), f"{cls.__name__}.{name}" - # Builds the check; an annotation it can't read raises here. - argument_check(inspect.unwrap(func)) + if name.startswith("_") and name != "__init__": + continue + if isinstance(func, property): + # A property that only reads takes no argument to check. + assert func.fset is None, f"{cls.__name__}.{name} has a setter" + continue + assert inspect.isfunction(func), f"{cls.__name__}.{name} isn't checked" + assert getattr(func, "_checks_arguments", False), f"{cls.__name__}.{name}" + # Builds the check; an annotation it can't read raises here. + argument_check(inspect.unwrap(func)) exported = [getattr(calendaring_sync, name) for name in calendaring_sync.__all__] for cls in exported: if isinstance(cls, type) and dataclasses.is_dataclass(cls): @@ -2559,12 +2565,17 @@ class _HoldsEmpty: class TestSQLiteStore: - def test_busy_file_raises_store_busy_error_and_stays_usable(self, db_path): + # Another process's write makes a write busy as it starts, its read as it commits. + @pytest.mark.parametrize("held", ["write", "read"]) + def test_busy_file_raises_store_busy_error_and_stays_usable(self, db_path, held): with _busy_pair(db_path) as (holder, waiter): - with holder.transaction(), pytest.raises(StoreBusyError): - _enter(waiter.transaction()) - with waiter.transaction(): - waiter.put_scope(_new_scope()) + with ( + _held_open(holder, write=held == "write", work=holder.scopes), + pytest.raises(StoreBusyError), + ): + _add_scope(waiter) + assert _scope_ids(waiter) == [] + _add_scope(waiter) assert _scope_ids(holder) == [_SCOPE] def test_ingest_while_busy_changes_nothing_and_can_be_retried(self, db_path, clock): @@ -2689,13 +2700,28 @@ def test_a_dataclass_the_store_cant_map_is_refused(self, cls, message): with pytest.raises(TypeError, match=message): _reader(cls) - def test_a_read_held_open_past_the_timeout_makes_a_write_busy(self, db_path): + @pytest.mark.parametrize("mode, locks_reads_out", [("delete", True), ("wal", False)]) + def test_a_commit_waits_for_reads_here_where_it_locks_them_out( + self, db_path, mode, locks_reads_out + ): + _edit_file(db_path, f"PRAGMA journal_mode = {mode}", ()) with SQLiteStore(db_path, timeout=_SHORT_TIMEOUT) as store: - with _held_open(store, write=False, work=store.scopes): - with pytest.raises(StoreBusyError), _write_with_scope(store): - pass - assert _scope_ids(store) == [] - _add_scope(store) + # Where a commit locks reads out, a write keeps its changes in memory: + # the pragma gives the pages it holds before spilling, 0 for never. + spills = store._writer.execute("PRAGMA cache_spill").fetchone()[0] + assert bool(spills) is not locks_reads_out + with ( + ThreadPoolExecutor(1) as pool, + _held_open(store, write=False, work=store.scopes), + ): + writing = pool.submit(_add_scope, store) + if locks_reads_out: + # Past the timeout it's still waiting, not busy. + with pytest.raises(TimeoutError): + writing.result(_SHORT_TIMEOUT * 3) + else: + writing.result(_WAIT) + writing.result() assert _scope_ids(store) == [_SCOPE] def test_interrupted_ingest_leaves_the_token_after_reopening(self, db_path, clock):