diff --git a/docs/conf.py b/docs/conf.py index 75683eb..e14048c 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -158,3 +158,12 @@ "members": True, "show-inheritance": True, } + + +def _public_bases(app, name, obj, options, bases): + """Leave private base classes, such as the one every resolution shares, out of "Bases".""" + bases[:] = [base for base in bases if not base.__name__.startswith("_")] + + +def setup(app): + app.connect("autodoc-process-bases", _public_bases) diff --git a/docs/contribute.rst b/docs/contribute.rst index fa6cefd..2fc63c0 100644 --- a/docs/contribute.rst +++ b/docs/contribute.rst @@ -48,7 +48,7 @@ How the code is laid out * - ``replica.py``, ``session.py``, ``objects/``, ``_classify.py`` - The core: the replica, sync sessions, the classifier, and the values they take and return * - ``store/`` - - The store interface, and the stores that ship + - The store and storage interfaces, and the stores that ship * - ``adapters/base.py`` - The adapter base classes * - ``adapters/.py`` diff --git a/docs/examples/keeper.py b/docs/examples/keeper.py new file mode 100644 index 0000000..d77d293 --- /dev/null +++ b/docs/examples/keeper.py @@ -0,0 +1,44 @@ +# SPDX-FileCopyrightText: 2026 calendaring-sync contributors +# SPDX-License-Identifier: AGPL-3.0-or-later + +# start: keeper +import pickle +from pathlib import Path + +from calendaring_sync import MemoryData, MemoryKeeper, MemoryStore, Replica, StoreError + + +class FileKeeper(MemoryKeeper): + """Keeps a memory store's data in one file, replaced whole on each save.""" + + def __init__(self, path: Path) -> None: + self.path = path + + def load(self) -> MemoryData | None: + if not self.path.exists(): + return None + data: MemoryData = pickle.loads(self.path.read_bytes()) + return data + + def save(self, data: MemoryData) -> None: + # Written beside the file and moved over it, so a failure halfway keeps the last save. + written = self.path.with_name(self.path.name + ".new") + written.write_bytes(pickle.dumps(data)) + written.replace(self.path) + + def error(self, exc: Exception) -> StoreError | None: + return StoreError(str(exc)) if isinstance(exc, OSError) else None + + +replica = Replica(MemoryStore(keeper=FileKeeper(Path("sync.pickle")))) +# end: keeper + +from memory_adapter import MemoryAdapter, MemoryServer + +server = MemoryServer() +server.put("meeting", b"Team meeting", if_version=None) +MemoryAdapter(replica, server, "work").sync() + +# The next process loads what the last one saved. +reopened = Replica(MemoryStore(keeper=FileKeeper(Path("sync.pickle")))) +meeting = reopened.find_record("work", "meeting") diff --git a/docs/explanation/design.rst b/docs/explanation/design.rst index 1a6e7f0..6e82a6e 100644 --- a/docs/explanation/design.rst +++ b/docs/explanation/design.rst @@ -46,7 +46,7 @@ Why the parts around the core can be swapped The :term:`core` holds every sync rule and meets the rest through narrow interfaces, so each part around it can change without the core or the other parts changing: -- where the state lives: a SQLite file, memory, or storage of your own, behind one store interface, as :doc:`../how-to/store` shows +- where the state lives: a SQLite file, memory, or a :term:`storage` of your own, behind one interface, as :doc:`../how-to/store` shows - where changes come from: an adapter that ships with calendaring-sync, one you write, or your own code handing the core :term:`change sets `, as :doc:`../how-to/write-an-adapter` and :doc:`../how-to/without-adapter` show - what reads and edits the data: the core never reads inside :term:`content`, so your application uses whichever library suits the format - how a sync runs: with a synchronous or an asynchronous client, for one :term:`scope` or many across servers and accounts, from one thread or several, as :doc:`../how-to/async` and :doc:`../how-to/parallel` show @@ -108,7 +108,7 @@ You can also ask for a full listing at any time, and each scope records when its Why local edits are recorded, not discovered ============================================ -You tell calendaring-sync about each :term:`local edit` as you make it: a change, a delete, or a create, which returns the new record's :term:`local key`. Repeated edits to one object collapse into one, an edit back to the :term:`synced content` cancels itself, and you can discard an edit. You can't edit an object with an open :term:`conflict` until you resolve it. +You tell calendaring-sync about each :term:`local edit` as you make it: a change, a delete, or a create, which returns the new record's :term:`local key`. Repeated edits to one object collapse into one, an edit back to the :term:`synced copy` cancels itself, and you can discard an edit. You can't edit an object with an open :term:`conflict` until you resolve it. Comparing every object at every sync instead would cost a full comparison each time, and a difference can't tell whether your application edited an object or never loaded its latest version. @@ -183,7 +183,7 @@ Every :term:`record` is in one of five states at a time: synced, pending, in fli in_flight --> removed: delete done in_flight --> pending: rejected in_flight --> conflict: gone, or UID taken - conflict --> synced: resolved: server's side + conflict --> synced: resolved: server's copy conflict --> pending: resolved: push it conflict --> removed: resolved: removal removed --> [*]: purged @@ -210,7 +210,7 @@ The table lists every transition with its conditions, including these, which the - Local edit recorded - Pending * - pending - - Edit back to the synced content, or discarded + - Edit back to the synced copy, or discarded - Synced; a create the server never had is deleted from the store instead * - pending - Remote change or removal ingested @@ -235,7 +235,7 @@ The table lists every transition with its conditions, including these, which the - Settled by the next complete ingest * - conflict - Remote change ingested - - Conflict with the new remote side; synced if it now equals your edit, pending again if the server is back at the synced content, or removed if both sides deleted the object + - Conflict with the new server's copy; synced if it now equals your edit, pending again if the server is back at the synced copy, or removed if both sides deleted the object * - conflict - The object holding the logical key an edit was refused for stops holding it, removed or changed on the server or by your own push - Pending again, unless another object in the scope still holds the key @@ -260,9 +260,9 @@ The table lists every transition with its conditions, including these, which the Why conflicts are reported, not resolved ======================================== -When a :term:`conflict` comes up, calendaring-sync reports it and leaves the decision to your code. Only your application knows whether the local edit, the server's version, or a merge is right. A removal on the server against a local edit is always a conflict, never resolved automatically, because that would delete your edit without asking. +When a :term:`conflict` comes up, calendaring-sync reports it and leaves the decision to your code. Only your application knows whether the local edit, the server's copy, or a merge is right. A removal on the server against a local edit is always a conflict, never resolved automatically, because that would delete your edit without asking. -The core classifies each change by looking only at the :term:`synced content`, the server's change, and your edit. The result is one of six classes: +The core classifies each change by looking only at the :term:`synced copy`, the server's change, and your edit. The result is one of six classes: .. list-table:: :header-rows: 1 @@ -283,9 +283,9 @@ The core classifies each change by looking only at the :term:`synced content`, t * - ``CONFLICT`` - Both sides acted; a conflict record is stored, of one :class:`~calendaring_sync.objects.conflict.ConflictKind` -Later remote changes update the record's remote side. The conflict resolves itself when the server reaches your edit, goes back to the synced content, or removes an object you deleted too. An edit refused for its logical key goes back to pending once the object holding that key no longer holds it. +Later remote changes update the record's :term:`server's copy`. The conflict resolves itself when the server reaches your edit, goes back to the synced copy, or removes an object you deleted too. An edit refused for its logical key goes back to pending once the object holding that key no longer holds it. -The core stores conflicts instead of raising them: an exception for one object would abort every other change in the sync, and a :term:`resolver` called during a sync would run again on every retry. So you resolve them when you choose, outside any sync, with a :data:`~calendaring_sync.objects.resolution.Resolution`. Anything but the server's side becomes a pending edit over the server's current copy, so its push is checked against the server's current version. :meth:`~calendaring_sync.replica.Replica.resolve` refuses a record whose conflict changed since you read it, so a decision is never applied to news you haven't seen. +The core stores conflicts instead of raising them: an exception for one object would abort every other change in the sync, and a :term:`resolver` called during a sync would run again on every retry. So you resolve them when you choose, outside any sync, each with a :data:`~calendaring_sync.objects.resolution.Resolution`: one with :meth:`~calendaring_sync.replica.Replica.resolve`, or every open one through your :term:`resolver` with :meth:`~calendaring_sync.replica.Replica.resolve_conflicts`. Anything but the server's copy becomes a pending edit over the server's current copy, so its push is checked against the server's current version. :meth:`~calendaring_sync.replica.Replica.resolve` refuses a record whose conflict changed since you read it, so a decision is never applied to news you haven't seen. .. _design-feed: @@ -308,4 +308,4 @@ If a calendar or account disappears from the server, the adapter raises an error Why storage is synchronous ========================== -Storage is a typed interface, :class:`~calendaring_sync.store.base.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. +A :term:`store` is a :class:`~calendaring_sync.store.base.StateStore` over a :term:`storage`, a typed interface that a SQLite file, memory or storage of your own implements. The store holds every rule about transactions, threads, checks and errors once, so a storage only reads and writes. One 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/docs/how-to/conflicts.rst b/docs/how-to/conflicts.rst index 63b7368..0827a32 100644 --- a/docs/how-to/conflicts.rst +++ b/docs/how-to/conflicts.rst @@ -28,4 +28,4 @@ Resolve every conflict with a function :start-after: # start: merge :end-before: # end: merge -:meth:`~calendaring_sync.replica.Replica.resolve_conflicts` calls your function with each open conflict, and ``None`` leaves one open. A :class:`~calendaring_sync.objects.resolution.Merge` needs the fields the adapter reads from content, which :meth:`~calendaring_sync.adapters.base.Adapter.edit_fields` returns. :attr:`Record.conflict ` holds the server's side, and :class:`~calendaring_sync.objects.conflict.ConflictKind` says what each side did. +:meth:`~calendaring_sync.replica.Replica.resolve_conflicts` calls your function with each open conflict, and ``None`` leaves one open. It returns a :class:`~calendaring_sync.objects.resolution.ResolvedConflicts` saying which conflicts it resolved, left open, or found changed while your function ran. A :class:`~calendaring_sync.objects.resolution.Merge` needs the fields the adapter reads from content, which :meth:`~calendaring_sync.adapters.base.Adapter.edit_fields` returns. :attr:`Record.conflict ` holds the :term:`server's copy`, and :class:`~calendaring_sync.objects.conflict.ConflictKind` says what each side did. diff --git a/docs/how-to/mirror.rst b/docs/how-to/mirror.rst index 53bcbe4..028f50e 100644 --- a/docs/how-to/mirror.rst +++ b/docs/how-to/mirror.rst @@ -17,6 +17,6 @@ Your application often keeps its own table of events, in the shape its screens a The table and the :term:`cursor` live in one database, and each page of the feed updates both in one transaction. So a crash leaves the table matching the cursor, and the next run continues from there: no entry is handled twice, and none is skipped. -``apply`` reads the record each entry is about, since the record holds its latest synced content, however many entries there were since the last run. A record that's removed, or that was :term:`purged ` meanwhile, loses its row, and a dropped scope loses every row. Your own edits write no feed entries until the server has them, so the table holds each record's synced content, the copy both sides last agreed on. Read :attr:`Record.edit ` as well where your application shows edits that aren't sent yet. +``apply`` reads the record each entry is about, since the record holds its latest synced copy, however many entries there were since the last run. A record that's removed, or that was :term:`purged ` meanwhile, loses its row, and a dropped scope loses every row. Your own edits write no feed entries until the server has them, so the table holds each record's :term:`synced copy`. Read :attr:`Record.edit ` as well where your application shows edits that aren't sent yet. When the feed was trimmed past your cursor, :meth:`~calendaring_sync.replica.Replica.read_feed` raises :class:`~calendaring_sync.error.TrimmedFeedError`: rebuild the table from :meth:`~calendaring_sync.replica.Replica.records`, as :doc:`change-feed` shows. diff --git a/docs/how-to/store.rst b/docs/how-to/store.rst index bc21355..457610d 100644 --- a/docs/how-to/store.rst +++ b/docs/how-to/store.rst @@ -49,4 +49,13 @@ Keep sync state in memory Keep sync state in your own storage =================================== -To keep sync state where your application keeps its other data, subclass :class:`~calendaring_sync.store.memory.MemoryStore` and override how it loads and saves its data, as its description shows with :mod:`pickle`. For a database that should look up records itself, implement :class:`~calendaring_sync.store.base.StateStore`: its description lists the methods a store writes, and the base class adds every check, transaction rule and error a store needs. +To keep sync state in a file of your own, give :class:`~calendaring_sync.store.memory.MemoryStore` a :class:`~calendaring_sync.store.memory.MemoryKeeper`. The store loads its data from the keeper once, when it opens, and hands it to the keeper each time a write commits: + +.. literalinclude:: ../examples/keeper.py + :language: python + :start-after: # start: keeper + :end-before: # end: keeper + +Only load a file your application wrote: unpickling a file runs code from it. Every lookup scans the records in memory, so for a large store use SQLite. + +For a database that looks records up itself, implement :class:`~calendaring_sync.store.base.Storage` and :class:`~calendaring_sync.store.base.StorageTransaction`, and pass the storage to :class:`~calendaring_sync.store.base.StateStore`, which does everything but the reading and writing. diff --git a/docs/reference/glossary.rst b/docs/reference/glossary.rst index 6e4e0b7..9144cdb 100644 --- a/docs/reference/glossary.rst +++ b/docs/reference/glossary.rst @@ -32,7 +32,7 @@ These are the sync terms calendaring-sync's pages use. :doc:`../explanation/desi An object where your :term:`pending edit` and the server disagree: both sides changed it, one side deleted it while the other edited it, or the server refused your create or change because another object holds its :term:`logical key`. calendaring-sync reports it and leaves the decision to you. conflict record - What the core stores for one :term:`conflict`: the :term:`synced content`, your edit, the server's side, and the kind of conflict. It stays open until you resolve it or it resolves itself. + What the core stores for one :term:`conflict`: the :term:`synced copy`, your :term:`local edit`, the :term:`server's copy`, and the kind of conflict. It stays open until you resolve it or it resolves itself. container A calendar, or another grouping, that an object belongs to. One object can belong to more than one. @@ -105,6 +105,12 @@ These are the sync terms calendaring-sync's pages use. :doc:`../explanation/desi session One sync of one :term:`scope`: it :term:`ingests ` every page first, then :term:`pushes ` your pending edits. + server's copy + The server's newer copy of an object you edited, which conflicts with your :term:`local edit`. A record in conflict holds it in ``Record.conflict``, in the fields whose names start with ``remote_``. + + storage + What a :term:`store` keeps its data in, such as a SQLite file, behind the :class:`~calendaring_sync.store.base.Storage` interface. Write one to keep sync state somewhere new. + store Where the core keeps each scope's sync token, its records, your pending edits, open conflicts, and the :term:`change feed`. @@ -112,8 +118,8 @@ These are the sync terms calendaring-sync's pages use. :doc:`../explanation/desi token A marker the server gives out that stands for one point in a scope's history. The adapter sends it back to ask what changed since then. The core only stores and compares it. - synced content - An object's content as both sides last agreed on it, at the end of the last sync. + synced copy + An object as both sides last agreed on it, at the end of the last sync. A :term:`record` holds it in its fields from ``version`` to ``logical_key``. truncated page page diff --git a/docs/reference/guarantees.rst b/docs/reference/guarantees.rst index 551d2eb..50c9180 100644 --- a/docs/reference/guarantees.rst +++ b/docs/reference/guarantees.rst @@ -41,7 +41,7 @@ Guarantees - :ref:`design-feed` * - A scope the server no longer has is never emptied for you. Dropping it returns the edits and conflicts it discards. - :ref:`design-dropped-scope` - * - In the stores that ship, a read in another thread doesn't wait for a sync's write, at most for its commit, and sees the store as it was before the write began. A store of your own does the same where it sets ``_reads_beside_writes``. + * - In the stores that ship, a read in another thread doesn't wait for a sync's write, at most for its commit, and sees the store as it was before the write began. A store over a storage of your own does the same where the storage sets :attr:`~calendaring_sync.store.base.Storage.reads_beside_writes`. - :ref:`design-storage` * - Every store checks arguments and raises errors the same way, since the checks live in the base class. - :class:`~calendaring_sync.store.base.StateStore` diff --git a/docs/reference/store.rst b/docs/reference/store.rst index 8ba4938..9165f05 100644 --- a/docs/reference/store.rst +++ b/docs/reference/store.rst @@ -5,17 +5,19 @@ ====================================================== .. meta:: - :description: StateStore, the storage interface, and its implementations: SQLiteStore for real use and MemoryStore for tests. + :description: StateStore, the storage interface it keeps its data in, and the stores that ship: SQLiteStore, and MemoryStore with an optional keeper. .. automodule:: calendaring_sync.store :no-members: .. automodule:: calendaring_sync.store.base :members: - :private-members: _begin, _commit, _rollback, _scope_records_in, _store_error, _writing, _reads_beside_writes .. automodule:: calendaring_sync.store.memory :members: + :exclude-members: MemoryData + +.. autoclass:: calendaring_sync.store.memory.MemoryData() .. automodule:: calendaring_sync.store.sqlite :members: diff --git a/news/28.feature.rst b/news/28.feature.rst index dfe5cc4..0ea3281 100644 --- a/news/28.feature.rst +++ b/news/28.feature.rst @@ -1 +1 @@ -A :class:`~calendaring_sync.store.memory.MemoryStore` subclass can keep its data elsewhere, such as in a file with ``pickle``, by overriding ``_load()`` and ``_save()``. +A :class:`~calendaring_sync.store.memory.MemoryStore` can keep its data elsewhere, such as in a file, through a :class:`~calendaring_sync.store.memory.MemoryKeeper` that loads it when the store opens and saves it each time a write commits. diff --git a/news/40.feature.rst b/news/40.feature.rst new file mode 100644 index 0000000..af7e3a8 --- /dev/null +++ b/news/40.feature.rst @@ -0,0 +1 @@ +A store is a :class:`~calendaring_sync.store.base.StateStore` over a :class:`~calendaring_sync.store.base.Storage`. To keep sync state somewhere new, implement :class:`~calendaring_sync.store.base.Storage` and :class:`~calendaring_sync.store.base.StorageTransaction`; the store adds every check, transaction rule and error. diff --git a/news/9.feature.rst b/news/9.feature.rst index 7576bf4..3cef1a9 100644 --- a/news/9.feature.rst +++ b/news/9.feature.rst @@ -1 +1 @@ -Added pushes and conflict resolution: a :class:`~calendaring_sync.session.Session` claims each pending edit before it's sent, so no other process sends it at the same time, and records how the push ended. :meth:`Replica.resolve() ` and :meth:`Replica.resolve_conflicts() ` resolve a conflict with your edit, the server's side, merged content, or a delete. +Added pushes and conflict resolution: a :class:`~calendaring_sync.session.Session` claims each pending edit before it's sent, so no other process sends it at the same time, and records how the push ended. :meth:`Replica.resolve() ` resolves one conflict with your edit, the server's copy, merged content, or a delete, and :meth:`Replica.resolve_conflicts() ` resolves every open one with your function and returns which it resolved, left open, or found changed. diff --git a/src/calendaring_sync/__init__.py b/src/calendaring_sync/__init__.py index 20efcf1..663396d 100644 --- a/src/calendaring_sync/__init__.py +++ b/src/calendaring_sync/__init__.py @@ -66,13 +66,14 @@ KeepServer, Merge, Resolution, + ResolvedConflicts, UpdateHolder, ) from calendaring_sync.objects.scope import Scope from calendaring_sync.replica import Replica from calendaring_sync.session import Session, SessionState -from calendaring_sync.store.base import StateStore -from calendaring_sync.store.memory import MemoryStore +from calendaring_sync.store.base import StateStore, Storage, StorageTransaction +from calendaring_sync.store.memory import MemoryData, MemoryKeeper, MemoryStore from calendaring_sync.store.sqlite import SQLiteStore __all__ = [ @@ -97,9 +98,11 @@ "KeepLocal", "KeepServer", "ListingRequiredError", - "LoneSurrogateError", "LocalEdit", "LogicalKeyTaken", + "LoneSurrogateError", + "MemoryData", + "MemoryKeeper", "MemoryStore", "Merge", "ObjectIdAssignedError", @@ -116,6 +119,7 @@ "RemovedRecordError", "Replica", "Resolution", + "ResolvedConflicts", "SQLiteStore", "Scope", "ScopeGoneError", @@ -126,6 +130,8 @@ "StaleChangeSetError", "StaleConflictError", "StateStore", + "Storage", + "StorageTransaction", "StoreBusyError", "StoreError", "StoreVersionError", diff --git a/src/calendaring_sync/_classify.py b/src/calendaring_sync/_classify.py index 1cace3e..15d1b2f 100644 --- a/src/calendaring_sync/_classify.py +++ b/src/calendaring_sync/_classify.py @@ -64,7 +64,7 @@ def same_remote(conflict: Conflict, change: Changed | Removed) -> bool: def same_conflict(a: Conflict, b: Conflict) -> bool: - """Whether two copies of a conflict say the same thing, whatever the server's version.""" + """Whether two copies of a conflict say the same thing, whatever their versions.""" if (a.kind, a.holder) != (b.kind, b.holder): return False return _same(data(a, REMOTE), data(b, REMOTE)) @@ -75,7 +75,7 @@ def classify( remote: Changed | Removed | None, local: LocalEdit | None, ) -> tuple[ChangeClass, ConflictKind | None]: - """Classify one object from its synced side, the server's report and the local edit. + """Classify one object from its synced copy, the server's report and the local edit. Each case below is one kind of local edit meeting one kind of report from the server. A change that holds the synced data counts as no @@ -83,7 +83,7 @@ def classify( the same object. Args: - synced: The record's :term:`synced content`, ``None`` for an object + synced: The record's :term:`synced copy`, ``None`` for an object the store doesn't hold, one it holds as removed, or a create. remote: The server's new copy, ``Removed`` if the server removed it, or ``None`` if the server reported nothing new. diff --git a/src/calendaring_sync/adapters/base.py b/src/calendaring_sync/adapters/base.py index 9f95e11..4bf4395 100644 --- a/src/calendaring_sync/adapters/base.py +++ b/src/calendaring_sync/adapters/base.py @@ -235,6 +235,10 @@ class Adapter(ABC): - :meth:`_new_object_id` picks the :term:`object id` of a create. - :meth:`_edit_fields` reads an edit's fields from its content. + These names start with an underscore because your application calls + the adapter's public methods: a read or a send called on its own would + skip the session that orders them. + Args: replica: Where the sync state is kept. scope_id: The scope's id. diff --git a/src/calendaring_sync/error.py b/src/calendaring_sync/error.py index ce446c7..8093584 100644 --- a/src/calendaring_sync/error.py +++ b/src/calendaring_sync/error.py @@ -130,7 +130,7 @@ class RemovedRecordError(EditRefusedError): class StaleConflictError(CalendaringSyncError): """The :term:`conflict` changed since your application read it, so a resolution isn't applied. - The record ended, or its edit or the server's side moved on. Read it + The record ended, or its edit or the server's copy moved on. Read it again and decide on what it holds now. """ diff --git a/src/calendaring_sync/objects/_data.py b/src/calendaring_sync/objects/_data.py index 96e119b..5e890f7 100644 --- a/src/calendaring_sync/objects/_data.py +++ b/src/calendaring_sync/objects/_data.py @@ -8,12 +8,12 @@ from typing import TypeAlias #: The fields that hold an object's data, in every copy of it: a record's -#: synced side, a change from the server, a local edit, and a conflict's +#: synced copy, a change from the server, a local edit, and a conflict's #: server side. Two copies hold the same data when these match, and a record #: synced to a copy takes these from it. DATA_FIELDS = ("fingerprint", "content", "containers", "logical_key") -#: What a conflict puts in front of each field name for the server's side. +#: What a conflict puts in front of each field name for the server's copy. REMOTE = "remote_" #: A copy's data: its fingerprint, content, containers and logical key. diff --git a/src/calendaring_sync/objects/change_set.py b/src/calendaring_sync/objects/change_set.py index e164f76..eaf2a9a 100644 --- a/src/calendaring_sync/objects/change_set.py +++ b/src/calendaring_sync/objects/change_set.py @@ -7,9 +7,10 @@ from collections.abc import Mapping from dataclasses import dataclass +from types import MappingProxyType from calendaring_sync._checks import check_text, checked_when_built -from calendaring_sync.objects.change import Change +from calendaring_sync.objects.change import Change, Unchanged @checked_when_built @@ -28,7 +29,8 @@ class ChangeSet: token, the one the listing read before it started, and the scope moves to it when the last page is ingested. changes: Each object id mapped to how the object changed. An object - appears once. + appears once. The change set keeps a read-only copy, so changing + the mapping it was built from afterwards changes nothing. full_listing: Whether it lists every object in the scope, rather than only what changed. truncated: Whether more pages follow it. @@ -36,6 +38,8 @@ class ChangeSet: Raises: TypeError: If a field has the wrong type. LoneSurrogateError: If the scope id, a token or an object id holds a lone surrogate. + ValueError: If an incremental change set reports an object ``Unchanged``, + which only a full listing can. """ scope: str @@ -46,4 +50,10 @@ class ChangeSet: truncated: bool = False def __post_init__(self) -> None: + # Its own copy, so what's checked here is what gets ingested. + object.__setattr__(self, "changes", MappingProxyType(dict(self.changes))) check_text(scope=self.scope, since=self.since, token=self.token, object_ids=self.changes) + if not self.full_listing: + for object_id, change in self.changes.items(): + if isinstance(change, Unchanged): + raise ValueError(f"Unchanged object {object_id!r} outside a full listing") diff --git a/src/calendaring_sync/objects/conflict.py b/src/calendaring_sync/objects/conflict.py index 4cc067f..95d7aa4 100644 --- a/src/calendaring_sync/objects/conflict.py +++ b/src/calendaring_sync/objects/conflict.py @@ -1,7 +1,7 @@ # SPDX-FileCopyrightText: 2026 calendaring-sync contributors # SPDX-License-Identifier: AGPL-3.0-or-later -"""The server's side of a :term:`conflict`, kept on its :term:`record` until it's resolved.""" +"""The :term:`server's copy` in a :term:`conflict`, kept on its :term:`record` until resolved.""" from __future__ import annotations @@ -15,7 +15,7 @@ class ConflictKind(StrEnum): - """How the server's side and the :term:`local edit` disagree.""" + """How the :term:`server's copy` and the :term:`local edit` disagree.""" #: Both sides changed the object. BOTH_CHANGED = "both_changed" diff --git a/src/calendaring_sync/objects/edit.py b/src/calendaring_sync/objects/edit.py index 08f696c..6cb4343 100644 --- a/src/calendaring_sync/objects/edit.py +++ b/src/calendaring_sync/objects/edit.py @@ -55,7 +55,7 @@ class EditFields: """What an edit records about an object beside its content. An :term:`adapter` reads these from content the same way for the - server's copy and for an edit, so an edit back to the synced content + server's copy and for an edit, so an edit back to the synced copy cancels, and the server's copy of your own edit isn't a conflict. To build them for content you made, such as a :class:`Merge `, call the adapter's diff --git a/src/calendaring_sync/objects/feed.py b/src/calendaring_sync/objects/feed.py index cad1b97..e70b7a0 100644 --- a/src/calendaring_sync/objects/feed.py +++ b/src/calendaring_sync/objects/feed.py @@ -22,7 +22,7 @@ class FeedEntryKind(StrEnum): SCOPE_DROPPED = "scope_dropped" #: A :term:`conflict` was stored. CONFLICT_OPENED = "conflict_opened" - #: The server's side of an open conflict changed. + #: The server's copy in an open conflict changed. CONFLICT_UPDATED = "conflict_updated" #: An open conflict ended, and the record holds what it ended with. CONFLICT_RESOLVED = "conflict_resolved" diff --git a/src/calendaring_sync/objects/push.py b/src/calendaring_sync/objects/push.py index 2d1bd10..7d495d6 100644 --- a/src/calendaring_sync/objects/push.py +++ b/src/calendaring_sync/objects/push.py @@ -5,15 +5,64 @@ from __future__ import annotations -from dataclasses import dataclass -from typing import TypeAlias +from abc import ABC, abstractmethod +from dataclasses import dataclass, replace +from datetime import datetime +from typing import NamedTuple, TypeAlias, cast from calendaring_sync._checks import check_text, checked_when_built +from calendaring_sync.objects.change import Removed +from calendaring_sync.objects.conflict import Conflict, ConflictKind +from calendaring_sync.objects.edit import EditKind, LocalEdit +from calendaring_sync.objects.feed import FeedEntryKind +from calendaring_sync.objects.record import Record, RecordState + + +class _Settled(NamedTuple): + """What settling a push writes.""" + + #: The record as the outcome leaves it. + record: Record + #: The change feed entry about it. + kind: FeedEntryKind + #: An object id the record takes over from a removed record, if any. + takes: str | None = None + + +def _deleted(record: Record, now: datetime) -> _Settled | None: + """A push that removed the object, if the record's edit was a delete.""" + if cast(LocalEdit, record.edit).kind is EditKind.DELETE: + return _Settled(record.removed_at(now), FeedEntryKind.PUSHED) + return None + + +def _in_conflict( + record: Record, kind: ConflictKind, now: datetime, holder: str | None = None +) -> _Settled: + """A conflict whose server has no copy of the object, opened by a push's outcome.""" + conflict = replace(Conflict.from_change(kind, Removed(), now), holder=holder) + opened = replace(record, state=RecordState.CONFLICT, conflict=conflict) + return _Settled(opened, FeedEntryKind.CONFLICT_OPENED) + + +class _Outcome(ABC): + """What every push outcome says about the record whose push it ends. + + That record is in flight, or pending again after an ingest, so it + always holds its edit; the store checks it. + """ + + def _check(self, kind: EditKind, has_object_id: bool) -> None: # noqa: B027, most fit any push + """Raise ``ValueError`` if no push of an edit of this kind can end this way.""" + + @abstractmethod + def _settled(self, record: Record, now: datetime) -> _Settled | None: + """What the outcome writes for the claimed record, or ``None`` for nothing.""" @checked_when_built @dataclass(frozen=True) -class Applied: +class Applied(_Outcome): """The server applied the edit. Attributes: @@ -39,9 +88,26 @@ class Applied: def __post_init__(self) -> None: check_text(object_id=self.object_id) + def _check(self, kind: EditKind, has_object_id: bool) -> None: + if kind is EditKind.DELETE and self.object_id is not None: + raise ValueError("An applied delete has no object id to report") + if kind is EditKind.CREATE and self.object_id is None and not has_object_id: + raise ValueError("An applied create needs the object id the server gave it") + + def _settled(self, record: Record, now: datetime) -> _Settled | None: + deleted = _deleted(record, now) + if deleted is not None: + return deleted + renamed = self.object_id + # The server may store the edit rewritten, so no version is kept. + pushed = record.synced_to(cast(LocalEdit, record.edit), now) + if renamed is None or renamed == record.object_id: + return _Settled(pushed, FeedEntryKind.PUSHED) + return _Settled(replace(pushed, object_id=renamed), FeedEntryKind.PUSHED, renamed) + @dataclass(frozen=True) -class Rejected: +class Rejected(_Outcome): """The server refused the write because something changed since your last sync. The edit stays pending. The next :term:`ingest` makes it a @@ -49,18 +115,32 @@ class Rejected: otherwise it's pushed again. """ + def _settled(self, record: Record, now: datetime) -> _Settled | None: + if record.state is RecordState.PENDING: + # An ingest already put the push back; the store says what this would. + return None + return _Settled(replace(record, state=RecordState.PENDING), FeedEntryKind.PUSH_FAILED) + @dataclass(frozen=True) -class Gone: +class Gone(_Outcome): """The object no longer exists on the server. A delete is then done; a change becomes a :term:`conflict`. """ + def _check(self, kind: EditKind, has_object_id: bool) -> None: + if kind is EditKind.CREATE: + raise ValueError("A create can't be gone: the server never had the object") + + def _settled(self, record: Record, now: datetime) -> _Settled | None: + deleted = _deleted(record, now) + return deleted or _in_conflict(record, ConflictKind.REMOVED_REMOTELY, now) + @checked_when_built @dataclass(frozen=True) -class LogicalKeyTaken: +class LogicalKeyTaken(_Outcome): """The server refused a create or a change because another object holds its :term:`logical key`. Attributes: @@ -77,15 +157,25 @@ class LogicalKeyTaken: def __post_init__(self) -> None: check_text(holder=self.holder) + def _check(self, kind: EditKind, has_object_id: bool) -> None: + if kind is EditKind.DELETE: + raise ValueError("A delete can't be refused for its logical key") + + def _settled(self, record: Record, now: datetime) -> _Settled | None: + return _in_conflict(record, ConflictKind.LOGICAL_KEY_TAKEN, now, self.holder) + @dataclass(frozen=True) -class Unknown: +class Unknown(_Outcome): """The request may or may not have reached the server. The push stays :term:`in flight`, and the next complete :term:`ingest` settles it from what the server then holds. """ + def _settled(self, record: Record, now: datetime) -> _Settled | None: + return None + #: Any one of the five. PushOutcome: TypeAlias = Applied | Rejected | Gone | LogicalKeyTaken | Unknown diff --git a/src/calendaring_sync/objects/record.py b/src/calendaring_sync/objects/record.py index b952482..a1cc5f1 100644 --- a/src/calendaring_sync/objects/record.py +++ b/src/calendaring_sync/objects/record.py @@ -60,7 +60,7 @@ class Record: fingerprint: The :term:`fingerprint` of :attr:`content`, or ``None`` after a :term:`fingerprint scheme` change until the object is listed again. - content: The :term:`synced content`, the bytes the server sent, in + content: The content of the :term:`synced copy`, the bytes the server sent, in the scope's media type. ``None`` for an object you created that the server doesn't have yet. containers: The :term:`containers ` of the synced copy. @@ -71,7 +71,7 @@ class Record: the record, ``0`` if there's none. edit: The record's :term:`pending edit`, held while it's pending, in flight or in conflict, and ``None`` otherwise. - conflict: The server's side of an open :term:`conflict`, held while + conflict: The :term:`server's copy` in an open :term:`conflict`, held while the record is in conflict, and ``None`` otherwise. claims: How many times a :term:`push` of the record was :term:`claimed `. How a push ended only counts while no diff --git a/src/calendaring_sync/objects/resolution.py b/src/calendaring_sync/objects/resolution.py index 2699071..86db738 100644 --- a/src/calendaring_sync/objects/resolution.py +++ b/src/calendaring_sync/objects/resolution.py @@ -5,25 +5,78 @@ from __future__ import annotations -from dataclasses import dataclass -from typing import TypeAlias +from abc import ABC, abstractmethod +from dataclasses import dataclass, replace +from datetime import datetime +from typing import TypeAlias, cast from calendaring_sync._checks import check_data, checked_when_built -from calendaring_sync.objects.edit import EditFields +from calendaring_sync.objects.conflict import Conflict +from calendaring_sync.objects.edit import EditFields, EditKind, LocalEdit +from calendaring_sync.objects.record import Record, RecordState + +#: What a resolution leaves: the record over the server's current copy, or +#: ``None`` where nothing is left, and the edit it should hold, or ``None`` +#: for the record as it is. Only a record in conflict is resolved, so every +#: function below can rely on its conflict and its edit being there. +_Resolved: TypeAlias = tuple[Record | None, LocalEdit | None] + + +def _on_server(record: Record) -> bool: + """Whether the server has the object the record's conflict is about.""" + return cast(Conflict, record.conflict).remote_content is not None or record.is_refused_change + + +def _over_server_copy(record: Record, now: datetime) -> Record: + """The record synced to the server's current copy, so the next push goes out against it.""" + if record.is_refused_change: + # The server kept the synced copy. + return replace(record, state=RecordState.SYNCED, edit=None, conflict=None) + return record.synced_to(cast(Conflict, record.conflict), now) + + +def _gone(record: Record, now: datetime) -> _Resolved: + """A record whose object the server doesn't have: removed, or nothing if it never had it.""" + return (None if record.is_create else record.removed_at(now)), None + + +def _edit_kind(record: Record) -> EditKind: + """Over a copy the server has, an edit is a change; with none, a create.""" + return EditKind.CHANGE if _on_server(record) else EditKind.CREATE + + +class _Resolution(ABC): + """What every resolution says about a record in conflict: if it applies, what it leaves.""" + + def _check(self, record: Record) -> None: # noqa: B027, most apply to any conflict + """Raise ``ValueError`` if the resolution can't apply to the record's conflict.""" + + @abstractmethod + def _resolve(self, record: Record, now: datetime) -> _Resolved: + """What the resolution leaves of the record, and the edit it should hold.""" + + def _holder_edit(self, record: Record) -> LocalEdit | None: + """The edit to give the object holding the record's logical key, if any.""" + return None @dataclass(frozen=True) -class KeepServer: - """Keep the server's side, and drop your :term:`local edit`. +class KeepServer(_Resolution): + """Keep the :term:`server's copy`, and drop your :term:`local edit`. Where the server has the object, the record takes the server's copy. Where it doesn't, because the server removed the object or refused your create, the record is removed, or deleted if the server never had it. """ + def _resolve(self, record: Record, now: datetime) -> _Resolved: + if not _on_server(record): + return _gone(record, now) + return _over_server_copy(record, now), None + @dataclass(frozen=True) -class KeepLocal: +class KeepLocal(_Resolution): """Push your :term:`local edit` as it is, over the server's current copy. Where the server removed the object, a change goes out as a create. An @@ -31,17 +84,27 @@ class KeepLocal: :term:`logical key` can't go out unchanged, so this refuses it. """ + def _check(self, record: Record) -> None: + if record.is_refused: + raise ValueError("An edit refused for its logical key can't be pushed again unchanged") + + def _resolve(self, record: Record, now: datetime) -> _Resolved: + edit = cast(LocalEdit, record.edit) + if edit.kind is not EditKind.DELETE: + edit = replace(edit, kind=_edit_kind(record)) + return _over_server_copy(record, now), edit + @checked_when_built @dataclass(frozen=True) -class Merge: +class Merge(_Resolution): """Push content you merged from the conflict's versions, over the server's current copy. A record in conflict holds three versions of its object: the - :term:`synced content` both sides last agreed on, in + :term:`synced copy` both sides last agreed on, in :attr:`~calendaring_sync.objects.record.Record.content`; your :term:`local edit`, in :attr:`~calendaring_sync.objects.record.Record.edit`; - and the server's side, in + and the :term:`server's copy`, in :attr:`~calendaring_sync.objects.record.Record.conflict`. Your application merges them into new content, and reads its fields with the :term:`adapter`'s :meth:`~calendaring_sync.adapters.base.Adapter.edit_fields`, @@ -58,7 +121,7 @@ class Merge: Examples: Your application keeps the room your edit set and the time the - server's side moved to, in one copy. An adapter would read its + server's copy moved to, in one copy. An adapter would read its fields; these are made up: >>> from calendaring_sync import EditFields, Merge @@ -75,10 +138,22 @@ def __post_init__(self) -> None: # Every other way in checks these strings where the core takes them. check_data(self.fields.fingerprint, self.fields.containers, self.fields.logical_key) + def _resolve(self, record: Record, now: datetime) -> _Resolved: + fields = self.fields + merged = LocalEdit( + _edit_kind(record), + self.content, + fields.fingerprint, + fields.containers, + fields.logical_key, + now, + ) + return _over_server_copy(record, now), merged + @dataclass(frozen=True) -class Delete: - """Delete the object, whatever the server's side holds. +class Delete(_Resolution): + """Delete the object, whatever the server's copy holds. Where the server has the object, the delete goes out over its current copy, and for a change the server refused for its :term:`logical key`, @@ -86,9 +161,15 @@ class Delete: the record is removed, or deleted if the server never had it. """ + def _resolve(self, record: Record, now: datetime) -> _Resolved: + if not _on_server(record): + return _gone(record, now) + deleted = LocalEdit(EditKind.DELETE, None, None, frozenset(), None, now) + return _over_server_copy(record, now), deleted + @dataclass(frozen=True) -class UpdateHolder: +class UpdateHolder(_Resolution): """Push a refused edit as a change to the object that holds its logical key. Only for a create or a change refused because another object holds its @@ -96,6 +177,36 @@ class UpdateHolder: a refused change was for stays as it was synced. """ + def _check(self, record: Record) -> None: + if not record.is_refused: + raise ValueError("Only an edit refused for its logical key goes to the key's holder") + + def _resolve(self, record: Record, now: datetime) -> _Resolved: + # The edit goes to the holder; this record keeps what the server has. + return KeepServer()._resolve(record, now) + + def _holder_edit(self, record: Record) -> LocalEdit | None: + return record.edit + + +@dataclass(frozen=True) +class ResolvedConflicts: + """What :meth:`~calendaring_sync.replica.Replica.resolve_conflicts` did with each conflict. + + Each field holds :term:`local keys `, in the order the + conflicts were passed to your resolver. + + Attributes: + resolved: The conflicts resolved as your resolver answered. + left_open: The conflicts your resolver answered ``None`` for. + changed: The conflicts that changed while your resolver ran, so its + answer wasn't applied. The next call passes them again. + """ + + resolved: tuple[int, ...] = () + left_open: tuple[int, ...] = () + changed: tuple[int, ...] = () + #: Any one of the five. Resolution: TypeAlias = KeepServer | KeepLocal | Merge | Delete | UpdateHolder diff --git a/src/calendaring_sync/replica.py b/src/calendaring_sync/replica.py index 4d97b4b..67373af 100644 --- a/src/calendaring_sync/replica.py +++ b/src/calendaring_sync/replica.py @@ -43,25 +43,15 @@ ) from calendaring_sync.objects.change import Change, Changed, Removed, Unchanged from calendaring_sync.objects.change_set import ChangeSet -from calendaring_sync.objects.conflict import Conflict, ConflictKind +from calendaring_sync.objects.conflict import Conflict from calendaring_sync.objects.dropped_scope import DroppedScope from calendaring_sync.objects.edit import EditKind, LocalEdit from calendaring_sync.objects.feed import FeedEntry, FeedEntryKind -from calendaring_sync.objects.push import ( - Applied, - Gone, - PushOutcome, - Rejected, - Unknown, -) +from calendaring_sync.objects.push import PushOutcome from calendaring_sync.objects.record import Record, RecordState from calendaring_sync.objects.resolution import ( - Delete, - KeepLocal, - KeepServer, - Merge, Resolution, - UpdateHolder, + ResolvedConflicts, ) from calendaring_sync.objects.scope import Scope from calendaring_sync.store.base import StateStore @@ -96,10 +86,7 @@ def _check_resolution(record: Record, resolution: Resolution) -> None: """Raise ``ValueError`` unless ``record`` is in conflict and ``resolution`` fits its kind.""" if record.state is not RecordState.CONFLICT or record.conflict is None: raise ValueError("resolve() takes a record in conflict") - if isinstance(resolution, KeepLocal) and record.is_refused: - raise ValueError("An edit refused for its logical key can't be pushed again unchanged") - if isinstance(resolution, UpdateHolder) and not record.is_refused: - raise ValueError("Only an edit refused for its logical key goes to the key's holder") + resolution._check(record) def _same_decision(stored: Record | None, record: Record) -> bool: @@ -116,38 +103,13 @@ def _same_decision(stored: Record | None, record: Record) -> bool: def _resolved(record: Record, resolution: Resolution, now: datetime) -> Record | None: """The record a resolution leaves, or ``None`` when it leaves nothing to keep. - The synced side becomes the server's current copy, so whatever is pushed + The synced copy becomes the server's current copy, so whatever is pushed next goes out against the server's current version. """ - conflict, edit = cast(Conflict, record.conflict), cast(LocalEdit, record.edit) - on_server = conflict.remote_content is not None or record.is_refused_change - if not on_server and isinstance(resolution, (KeepServer, Delete)): - # The server doesn't have the object: a create it never had leaves nothing. - return None if record.is_create else record.removed_at(now) - if record.is_refused_change: - # The server kept the synced copy. - base = replace(record, state=RecordState.SYNCED, edit=None, conflict=None) - else: - base = record.synced_to(conflict, now) - # Over a copy the server has, an edit is a change; with none, a create. - kind = EditKind.CHANGE if on_server else EditKind.CREATE - match resolution: - case KeepServer(): - return base - case KeepLocal(): - return _edited(base, edit if edit.kind is EditKind.DELETE else replace(edit, kind=kind)) - case Merge(): - merged = LocalEdit( - kind, - resolution.content, - resolution.fields.fingerprint, - resolution.fields.containers, - resolution.fields.logical_key, - now, - ) - return _edited(base, merged) - # Delete, the one resolution left. - return _edited(base, LocalEdit(EditKind.DELETE, None, None, frozenset(), None, now)) + base, edit = resolution._resolve(record, now) + if base is None or edit is None: + return base + return _edited(base, edit) def _edited(record: Record, edit: LocalEdit) -> Record: @@ -339,8 +301,6 @@ def _ingest( claimed = self._claimed_creates(scope.id) if unique_logical_key else {} freed: set[str] = set() for object_id, change in change_set.changes.items(): - if not change_set.full_listing and isinstance(change, Unchanged): - raise ValueError(f"Unchanged object {object_id!r} outside a full listing") self._apply(scope.id, object_id, change, now, claimed, freed) if change_set.full_listing: mentioned.add(object_id) @@ -631,7 +591,7 @@ def find_record(self, scope_id: str, object_id: str) -> Record | None: def find_by_logical_key(self, scope_id: str, logical_key: str) -> list[Record]: """Return the scope's records with this :term:`logical key`, ordered by local key. - A record matches if its synced side or its :term:`pending edit` has + A record matches if its synced copy or its :term:`local edit` has the key. Several records can share one, and removed records are included until they're :term:`purged `. @@ -730,7 +690,7 @@ def change( ) -> None: """Record a change to an object. - A change back to the :term:`synced content` cancels the edit, a + A change back to the :term:`synced copy` cancels the edit, a change after a delete replaces it, and a change to an object you created is still a create. Build ``fingerprint``, ``containers`` and ``logical_key`` as for :meth:`create`. @@ -751,7 +711,7 @@ def change( >>> replica.record(key).edit.content b'Team meeting, moved' - A change back to the synced content cancels the edit: + A change back to the synced copy cancels the edit: >>> replica.change(key, b"Team meeting", fingerprint="fp-a") >>> replica.record(key).state @@ -864,50 +824,17 @@ def _settle(self, claimed: Record, outcome: PushOutcome) -> bool: or record.state not in (RecordState.IN_FLIGHT, RecordState.PENDING) ): return False - pending_again = record.state is RecordState.PENDING held = record.server_logical_key - # A record in flight or pending always holds an edit; the store checks it. - edit = cast(LocalEdit, record.edit) - match outcome: - case Rejected() | Unknown() if pending_again: - pass - case Rejected(): - self._write( - replace(record, state=RecordState.PENDING), FeedEntryKind.PUSH_FAILED, now - ) - case Unknown(): - pass - case Applied() | Gone() if edit.kind is EditKind.DELETE: - self._write(record.removed_at(now), FeedEntryKind.PUSHED, now) - case Applied(object_id=renamed): - if renamed is not None and renamed != record.object_id: - self._free_object_id(record.scope, renamed, record.local_key) - # The server may store the edit rewritten, so no version is kept. - pushed = replace( - record.synced_to(edit, now), object_id=renamed or record.object_id - ) - self._write(pushed, FeedEntryKind.PUSHED, now) - case Gone(): - self._open_conflict(record, ConflictKind.REMOVED_REMOTELY, None, now) - case _: - # LogicalKeyTaken, the one outcome left. - self._open_conflict(record, ConflictKind.LOGICAL_KEY_TAKEN, outcome.holder, now) - settled = self._store.get_record(record.local_key) - if held is not None and (settled is None or settled.server_logical_key != held): + settled = outcome._settled(record, now) + if settled is not None: + if settled.takes is not None: + self._free_object_id(record.scope, settled.takes, record.local_key) + self._write(settled.record, settled.kind, now) + after = self._store.get_record(record.local_key) + if held is not None and (after is None or after.server_logical_key != held): self._lift_refusals(record.scope, {held}, now) return True - def _open_conflict( - self, record: Record, kind: ConflictKind, holder: str | None, now: datetime - ) -> None: - """Store a conflict whose server has no copy of the object, from a push's outcome.""" - conflict = replace(Conflict.from_change(kind, Removed(), now), holder=holder) - self._write( - replace(record, state=RecordState.CONFLICT, conflict=conflict), - FeedEntryKind.CONFLICT_OPENED, - now, - ) - def _free_object_id(self, scope_id: str, object_id: str, owner: int) -> None: """Make ``object_id`` free for record ``owner``. @@ -995,7 +922,7 @@ def records( def known_versions(self, scope_id: str) -> dict[str, str]: """Return the version of the server's copy of every object the store knows one for. - For a record in :term:`conflict` that's the server's side. An + For a record in :term:`conflict` that's the :term:`server's copy`. An :term:`adapter` compares these with the server's during a :term:`full listing`, and reports objects whose version matches as ``Unchanged`` instead of fetching them. Removed records aren't included. @@ -1074,7 +1001,7 @@ def drop_scope(self, scope_id: str) -> DroppedScope: def resolve(self, record: Record, resolution: Resolution) -> None: """Resolve one :term:`conflict`, as your application decided from ``record``. - Anything but the server's side becomes a :term:`pending edit` over the + Anything but the server's copy becomes a :term:`pending edit` over the server's current copy, so the next sync pushes it against the server's current version. Writes a ``CONFLICT_RESOLVED`` entry to the :term:`change feed`. Runs in a write transaction of its own. @@ -1086,8 +1013,8 @@ def resolve(self, record: Record, resolution: Resolution) -> None: Raises: TransactionError: If a store transaction is already open in this thread. StaleConflictError: If the conflict changed since ``record`` was - read: it ended, or its edit or the server's side moved on. A - move of the server's version alone doesn't count. + read: it ended, or its edit or the server's copy moved on. A + move of that copy's version alone doesn't count. UnknownHolderError: For ``UpdateHolder``, if the server named no record that's still there, and no single record holds the refused edit's logical key, or it was sent without one. @@ -1122,18 +1049,17 @@ def resolve(self, record: Record, resolution: Resolution) -> None: if not _same_decision(stored, record): raise StaleConflictError(f"Record {record.local_key}'s conflict changed") stored = cast(Record, stored) - if isinstance(resolution, UpdateHolder): + holder_edit = resolution._holder_edit(stored) + if holder_edit is not None: holder = self._editable(self._holder(stored).local_key) if holder.edit is not None: raise PendingEditError( f"Record {holder.local_key} holds a pending edit of its own" ) - edit = cast(LocalEdit, stored.edit) - self._store.update_record(_changed_to(holder, replace(edit, recorded_at=now))) - # A refused create leaves nothing; a refused change, the synced copy. - resolved = None if stored.is_create else _resolved(stored, KeepServer(), now) - else: - resolved = _resolved(stored, resolution, now) + self._store.update_record( + _changed_to(holder, replace(holder_edit, recorded_at=now)) + ) + resolved = _resolved(stored, resolution, now) if resolved is None: self._store.append_entry( stored.scope, @@ -1146,7 +1072,9 @@ def resolve(self, record: Record, resolution: Resolution) -> None: else: self._write(resolved, FeedEntryKind.CONFLICT_RESOLVED, now) - def resolve_conflicts(self, resolver: Callable[[Record], Resolution | None]) -> int: + def resolve_conflicts( + self, resolver: Callable[[Record], Resolution | None] + ) -> ResolvedConflicts: """Pass every open :term:`conflict` to ``resolver``, and resolve each as it answers. The conflicts are read in one read transaction, then ``resolver`` runs @@ -1162,13 +1090,13 @@ def resolve_conflicts(self, resolver: Callable[[Record], Resolution | None]) -> scope, and returns a resolution or ``None``. Returns: - How many conflicts were resolved. + Which conflicts were resolved, left open, or changed meanwhile. Raises: TransactionError: If a store transaction is already open in this thread. Examples: - An application that takes the server's side of every conflict, + An application that takes the server's copy in every conflict, without asking anyone, resolves them all in one call: >>> from calendaring_sync import KeepServer @@ -1179,8 +1107,8 @@ def resolve_conflicts(self, resolver: Callable[[Record], Resolution | None]) -> ... "work", fingerprint_scheme="sha256", media_type="text/plain" ... ) as session: ... session.ingest(ChangeSet("work", "t1", "t2", theirs)) - >>> replica.resolve_conflicts(lambda record: KeepServer()) - 1 + >>> replica.resolve_conflicts(lambda record: KeepServer()).resolved == (key,) + True >>> replica.record(key).content b'Team meeting, room 3' """ @@ -1190,17 +1118,21 @@ def resolve_conflicts(self, resolver: Callable[[Record], Resolution | None]) -> for scope in self._store.scopes() for record in self._store.scope_records(scope.id, states=_CONFLICT) ] - resolved = 0 + resolved: list[int] = [] + left_open: list[int] = [] + changed: list[int] = [] for record in open_conflicts: resolution = resolver(record) if resolution is None: + left_open.append(record.local_key) continue try: self.resolve(record, resolution) except StaleConflictError: + changed.append(record.local_key) continue - resolved += 1 - return resolved + resolved.append(record.local_key) + return ResolvedConflicts(tuple(resolved), tuple(left_open), tuple(changed)) def _holder(self, record: Record) -> Record: """The one live record holding a refused create's logical key. diff --git a/src/calendaring_sync/session.py b/src/calendaring_sync/session.py index 3dd1802..ca646d5 100644 --- a/src/calendaring_sync/session.py +++ b/src/calendaring_sync/session.py @@ -10,8 +10,7 @@ from calendaring_sync._checks import check_argument_types, check_text from calendaring_sync.error import OtherScopeError, SessionError, StaleChangeSetError from calendaring_sync.objects.change_set import ChangeSet -from calendaring_sync.objects.edit import EditKind -from calendaring_sync.objects.push import Applied, Gone, LogicalKeyTaken, PushOutcome +from calendaring_sync.objects.push import PushOutcome from calendaring_sync.objects.record import Record, RecordState from calendaring_sync.replica import Replica @@ -24,16 +23,7 @@ def _check_outcome(claimed_keys: set[int], claimed: Record, outcome: PushOutcome or claimed.edit is None ): raise ValueError("settle() takes a record this session's claim() returned") - kind = claimed.edit.kind - match outcome: - case Gone() if kind is EditKind.CREATE: - raise ValueError("A create can't be gone: the server never had the object") - case LogicalKeyTaken() if kind is EditKind.DELETE: - raise ValueError("A delete can't be refused for its logical key") - case Applied(object_id=str()) if kind is EditKind.DELETE: - raise ValueError("An applied delete has no object id to report") - case Applied(object_id=None) if kind is EditKind.CREATE and claimed.object_id is None: - raise ValueError("An applied create needs the object id the server gave it") + outcome._check(claimed.edit.kind, claimed.object_id is not None) class SessionState(StrEnum): @@ -135,7 +125,6 @@ def ingest(self, change_set: ChangeSet) -> None: scope with no token. UnknownVersionError: If an ``Unchanged`` entry names a record with no version. - ValueError: If an ``Unchanged`` entry is outside a listing. """ if change_set.scope != self._scope_id: raise OtherScopeError( diff --git a/src/calendaring_sync/store/__init__.py b/src/calendaring_sync/store/__init__.py index 08fb1b0..f0a0fb1 100644 --- a/src/calendaring_sync/store/__init__.py +++ b/src/calendaring_sync/store/__init__.py @@ -3,5 +3,5 @@ """Where the core keeps sync state. -The interface every :term:`store` implements, and the stores that ship. +The :term:`store`, the :term:`storage` interface it keeps its data in, and the stores that ship. """ diff --git a/src/calendaring_sync/store/base.py b/src/calendaring_sync/store/base.py index 6fe354f..c733f18 100644 --- a/src/calendaring_sync/store/base.py +++ b/src/calendaring_sync/store/base.py @@ -1,11 +1,13 @@ # SPDX-FileCopyrightText: 2026 calendaring-sync contributors # SPDX-License-Identifier: AGPL-3.0-or-later -"""The storage interface every :term:`store` implements. +"""The :term:`store` interface, and the :term:`storage` interface a store keeps its data in. A store only keeps data. Every sync rule lives in -:class:`~calendaring_sync.replica.Replica`, so each store implements the -same small set of typed operations and nothing more. +:class:`~calendaring_sync.replica.Replica`, and every rule about how a store +behaves, such as its transactions, its checks and its errors, lives in +:class:`StateStore`. Where the data goes is a :class:`Storage`, which +implements the same small set of typed operations and nothing more. """ from __future__ import annotations @@ -54,11 +56,174 @@ } +class StorageTransaction(ABC): + """One transaction on a :class:`Storage`, which :meth:`Storage.begin` returns. + + It holds whatever its transaction needs, such as a connection or a view + of the data, so a storage keeps no state per thread. Its data operations + are the ones :class:`StateStore` offers, without the checks: + :class:`StateStore` calls them only inside this transaction, after it + checked the arguments, the records and whether the transaction may + write, and turns every exception they raise into a store error. Each + returns what the :class:`StateStore` method of the same name returns. + + Args: + write: Whether the transaction may write. + """ + + def __init__(self, *, write: bool) -> None: + #: Whether the transaction may write. + self.write = write + + @abstractmethod + def commit(self) -> None: + """Keep what the transaction wrote, and end it.""" + + @abstractmethod + def rollback(self) -> None: + """Discard what the transaction wrote, and end it. + + It's also called after :meth:`commit` raised, so it ends whatever + the transaction still holds open. + """ + + @abstractmethod + def get_scope(self, scope_id: str) -> Scope | None: + """Do what :meth:`StateStore.get_scope` does, without its checks.""" + + @abstractmethod + def put_scope(self, scope: Scope) -> None: + """Do what :meth:`StateStore.put_scope` does, without its checks.""" + + @abstractmethod + def delete_scope(self, scope_id: str) -> None: + """Do what :meth:`StateStore.delete_scope` does, without its checks.""" + + @abstractmethod + def scopes(self) -> list[Scope]: + """Do what :meth:`StateStore.scopes` does, without its checks.""" + + @abstractmethod + def insert_record(self, record: Record) -> int: + """Do what :meth:`StateStore.insert_record` does, without its checks.""" + + @abstractmethod + def update_record(self, record: Record) -> None: + """Do what :meth:`StateStore.update_record` does, without its checks.""" + + @abstractmethod + def delete_record(self, local_key: int) -> None: + """Do what :meth:`StateStore.delete_record` does, without its checks.""" + + @abstractmethod + def get_record(self, local_key: int) -> Record | None: + """Do what :meth:`StateStore.get_record` does, without its checks.""" + + @abstractmethod + def find_record(self, scope_id: str, object_id: str) -> Record | None: + """Do what :meth:`StateStore.find_record` does, without its checks.""" + + @abstractmethod + def find_by_logical_key(self, scope_id: str, logical_key: str) -> list[Record]: + """Do what :meth:`StateStore.find_by_logical_key` does, without its checks.""" + + @abstractmethod + def scope_records(self, scope_id: str) -> list[Record]: + """Return every record of the scope, ordered by local key.""" + + def scope_records_in(self, scope_id: str, states: frozenset[RecordState]) -> list[Record]: + """Return the scope's records in one of ``states``, ordered by local key. + + This filters what :meth:`scope_records` returns. A storage that can + filter where it keeps its records, such as with an index, overrides + it. It never gets an empty set. + """ + return [record for record in self.scope_records(scope_id) if record.state in states] + + @abstractmethod + def removed_records(self, through: int) -> list[Record]: + """Do what :meth:`StateStore.removed_records` does, without its checks.""" + + @abstractmethod + def append_entry( + self, + scope_id: str, + kind: FeedEntryKind, + local_key: int | None, + object_id: str | None, + at: datetime, + ) -> FeedEntry: + """Do what :meth:`StateStore.append_entry` does, without its checks.""" + + @abstractmethod + def entries(self, after: int, limit: int) -> list[FeedEntry]: + """Do what :meth:`StateStore.entries` does, without its checks.""" + + @abstractmethod + def feed_bounds(self) -> tuple[int, int]: + """Do what :meth:`StateStore.feed_bounds` does, without its checks.""" + + @abstractmethod + def trim_entries(self, through: int) -> None: + """Do what :meth:`StateStore.trim_entries` does, without its checks.""" + + +class Storage(ABC): + """Where a :class:`StateStore` keeps its data. + + To keep sync state somewhere new, such as in another database, + implement this and a :class:`StorageTransaction`, and pass the storage + to :class:`StateStore`. The store adds every check, transaction rule, + thread rule and error a store needs, so a storage only reads and writes. + """ + + #: Whether a transaction that only reads can run while another thread's + #: write transaction is open. With ``False``, every transaction waits for + #: the one before it to end. Set it to ``True`` only where a read sees the + #: data as the last commit left it until it ends, whatever a write does + #: meanwhile; reads in other threads then run at the same time too, so + #: :meth:`begin` and the transactions it returns must be safe for that. + reads_beside_writes: bool = False + + @abstractmethod + def begin(self, *, write: bool) -> StorageTransaction: + """Start a transaction and return it. + + The store runs one write transaction at a time, and transactions + that only read beside it only if :attr:`reads_beside_writes` is + set. If this raises, it leaves nothing open. + """ + + def close(self) -> None: # noqa: B027, a storage may hold nothing to release + """Release what the storage holds open. + + The store calls it once, after every transaction ended. + """ + + def error(self, exc: Exception) -> StoreError | None: + """Return the store error for an exception the storage raised, or ``None``. + + Turn the failures of whatever the storage uses, such as a file it + can't open, into :class:`~calendaring_sync.error.StoreError` or one + of its subclasses. The store reports any exception this returns + ``None`` for as a bug in the storage. + """ + return None + + +def _unexpected(cls: type, exc: Exception) -> UnexpectedStoreError: + """The error for an exception ``cls`` didn't expect: a bug, reported to whoever wrote it.""" + name = f"{cls.__module__}.{cls.__qualname__}" + ours = cls.__module__.startswith(f"{__name__.partition('.')[0]}.") + where = f"; please report it at {ISSUES_URL}" if ours else f" in {name}" + return UnexpectedStoreError(f"{name} failed unexpectedly with {exc!r}. This is a bug{where}") + + class _Transaction(local): """This thread's open transaction.""" - #: Whether it writes, or ``None`` when the thread has none open. - write: bool | None = None + #: The storage's transaction, or ``None`` when the thread has none open. + current: StorageTransaction | None = None def _check_record(record: Record) -> None: @@ -68,13 +233,13 @@ def _check_record(record: Record) -> None: if (edit is not None) != (state in _EDIT_STATES): need = "needs" if state in _EDIT_STATES else "can't hold" raise RecordValueError(f"A {state} record {need} a local edit") - # The server's side of a conflict in exactly the conflict state. + # The server's copy in a conflict in exactly the conflict state. if (conflict is not None) != (state is RecordState.CONFLICT): need = "needs" if state is RecordState.CONFLICT else "can't hold" raise RecordValueError(f"A {state} record {need} a conflict") - # No synced content for exactly an object the server doesn't have yet. + # No synced copy for exactly an object the server doesn't have yet. if (record.content is None) != record.is_create: - raise RecordValueError("A record has no synced content exactly when its edit is a create") + raise RecordValueError("A record has no synced copy exactly when its edit is a create") # A delete holds no copy of the object; a change or a create holds one. if edit is not None: if edit.kind is EditKind.DELETE and data(edit) != NO_DATA: @@ -88,7 +253,7 @@ def _check_record(record: Record) -> None: if edit is None or edit.kind not in edit_kinds: kinds = " or ".join(sorted(edit_kinds)) raise RecordValueError(f"A {conflict.kind} conflict needs a {kinds} edit") - # The server's side holds a copy exactly when the server still has the object. + # The server's copy is there exactly when the server still has the object. removed = data(conflict, REMOTE) == NO_DATA and conflict.remote_version is None if conflict.remote_content is None and not removed: raise RecordValueError("A conflict without the server's content can't hold the rest of it") @@ -101,15 +266,17 @@ def _check_record(record: Record) -> None: @check_argument_types -class StateStore(ABC): +class StateStore: """Where the core keeps each :term:`scope`, its records, and the :term:`change feed`. - 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``. 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. + It keeps them in a :class:`Storage`, and adds every rule a store + follows, whichever storage it uses. 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``. A + store is safe to share between threads: one write transaction runs at + a time, and where the storage sets + :attr:`~Storage.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: @@ -120,31 +287,19 @@ class StateStore(ABC): - :class:`~calendaring_sync.error.StoreError`, if the store can't read or write its data - :class:`~calendaring_sync.error.UnexpectedStoreError`, for any other - failure inside the store, which is a bug - - No exception of the storage underneath a store reaches the caller. - - The public methods check that they run inside a transaction of the right - kind, then call a method of the same name with a leading underscore. A - store implements those, along with :meth:`_begin`, :meth:`_commit` and - :meth:`_rollback`, and needs no checks of its own; it may also override - :meth:`_scope_records_in` to filter records by state itself. It turns - the errors of the storage it uses into store errors by overriding - :meth:`_store_error`. - 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. - """ + failure inside the storage, which is a bug - #: 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 + No exception of the storage underneath reaches the caller. - def __init__(self) -> None: + Args: + storage: Where the store keeps its data. + """ + + def __init__(self, storage: Storage) -> None: + self._storage = storage # 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 + # Held by the write transaction, or by every transaction of a storage # that doesn't read beside a write. self._turn = Lock() self._thread = _Transaction() @@ -176,22 +331,29 @@ def transaction(self, *, write: bool = True) -> Iterator[None]: raise TransactionError("A store transaction is already open in this thread") self._open += 1 try: - waits = write or not self._reads_beside_writes + waits = write or not self._storage.reads_beside_writes if waits: self._turn.acquire() try: - self._thread.write = write - with self._errors(): - self._begin(write=write) - yield with self._errors(): - self._commit() - except BaseException: - with self._errors(): - self._rollback() - raise + current = self._storage.begin(write=write) + self._thread.current = current + try: + yield + with self._errors(): + current.commit() + except BaseException as exc: + # What ended the block stays the error, interrupts included; + # a rollback that fails too is noted on it. + try: + with self._errors(): + current.rollback() + except CalendaringSyncError as failed: + exc.add_note(f"Rolling the transaction back failed too: {failed}") + raise + finally: + self._thread.current = None finally: - self._thread.write = None if waits: self._turn.release() finally: @@ -201,21 +363,20 @@ def transaction(self, *, write: bool = True) -> Iterator[None]: def _in_transaction(self) -> bool: """Whether this thread has a transaction open.""" - return self._thread.write is not None - - def _writing(self) -> bool: - """Whether this thread's open transaction writes.""" - return self._thread.write is True + return self._thread.current is not None - 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(): + @contextmanager + def _operation(self, *, write: bool) -> Iterator[StorageTransaction]: + """Check this thread's transaction allows the operation, then yield it, errors guarded.""" + current = self._thread.current + if current is None: raise TransactionRequiredError("Store operations run inside store.transaction()") - if write and not self._writing(): + if write and not current.write: raise TransactionPermissionError( "A transaction opened with write=False can't change the store" ) - return self._errors() + with self._errors(): + yield current @contextmanager def _errors(self) -> Iterator[None]: @@ -228,17 +389,8 @@ def _errors(self) -> Iterator[None]: raise self._store_error(exc) from exc def _store_error(self, exc: Exception) -> CalendaringSyncError: - """Return the error to raise for an exception from inside the store. - - A store overrides this to turn the errors of the storage it uses - into :class:`~calendaring_sync.error.StoreError`, and passes every - other exception on to this method. - """ - cls = type(self) - return UnexpectedStoreError( - f"{cls.__module__}.{cls.__qualname__} failed unexpectedly with {exc!r}. " - f"This is a bug; please report it at {ISSUES_URL}" - ) + """Return the error to raise for an exception from inside the storage.""" + return self._storage.error(exc) or _unexpected(type(self._storage), exc) def get_scope(self, scope_id: str) -> Scope | None: """Return the scope with this id, or ``None``. @@ -248,8 +400,8 @@ def get_scope(self, scope_id: str) -> Scope | None: Raises: TransactionRequiredError: If no transaction is open in this thread. """ - with self._operation(write=False): - return self._get_scope(scope_id) + with self._operation(write=False) as current: + return current.get_scope(scope_id) def put_scope(self, scope: Scope) -> None: """Insert the scope, or replace the one with its id. @@ -260,8 +412,8 @@ def put_scope(self, scope: Scope) -> None: TransactionRequiredError: If no transaction is open in this thread. TransactionPermissionError: If the transaction only reads. """ - with self._operation(write=True): - self._put_scope(scope) + with self._operation(write=True) as current: + current.put_scope(scope) def delete_scope(self, scope_id: str) -> None: """Delete the scope and every record in it. @@ -272,8 +424,8 @@ def delete_scope(self, scope_id: str) -> None: TransactionRequiredError: If no transaction is open in this thread. TransactionPermissionError: If the transaction only reads. """ - with self._operation(write=True): - self._delete_scope(scope_id) + with self._operation(write=True) as current: + current.delete_scope(scope_id) def scopes(self) -> list[Scope]: """Return every scope, ordered by id. @@ -283,8 +435,8 @@ def scopes(self) -> list[Scope]: Raises: TransactionRequiredError: If no transaction is open in this thread. """ - with self._operation(write=False): - return self._scopes() + with self._operation(write=False) as current: + return current.scopes() def insert_record(self, record: Record) -> int: """Insert a new record and return the :term:`local key` it got. @@ -296,10 +448,10 @@ def insert_record(self, record: Record) -> int: Raises: RecordValueError: If the record holds what its state can't. A record holds an edit exactly when it's pending, in flight or - in conflict, the server's side of a conflict exactly when - it's in conflict, and no synced content exactly when its + in conflict, the server's copy in a conflict exactly when + it's in conflict, and no synced copy exactly when its edit is a create. An edit holds a copy of the object unless - it's a delete, a conflict's edit and server's side match its + it's a delete, a conflict's edit and server's copy match its kind, and only a refused create's conflict names a holder. The record is checked first, so this is raised outside a transaction too. @@ -309,11 +461,11 @@ def insert_record(self, record: Record) -> int: DuplicateObjectError: If another record in the scope has the record's object id. """ - with self._record_write(record): - if self._get_scope(record.scope) is None: + with self._record_write(record) as current: + if current.get_scope(record.scope) is None: raise UnknownScopeError(f"No scope {record.scope!r}") - self._check_unique(record) - return self._insert_record(record) + _check_unique(current, record) + return current.insert_record(record) def update_record(self, record: Record) -> None: """Replace the record with the same local key. @@ -329,15 +481,15 @@ def update_record(self, record: Record) -> None: DuplicateObjectError: If another record in the scope has the record's object id. """ - with self._record_write(record): - stored = self._get_record(record.local_key) + with self._record_write(record) as current: + stored = current.get_record(record.local_key) if stored is None: raise UnknownRecordError(f"No record with local key {record.local_key}") if (stored.scope, stored.object_id) != (record.scope, record.object_id): - self._check_unique(record) - self._update_record(record) + _check_unique(current, record) + current.update_record(record) - def _record_write(self, record: Record) -> AbstractContextManager[None]: + def _record_write(self, record: Record) -> AbstractContextManager[StorageTransaction]: """Check the record, then the transaction, and guard the write that follows. The record comes first: checking it costs less than anything the @@ -346,16 +498,6 @@ def _record_write(self, record: Record) -> AbstractContextManager[None]: _check_record(record) return self._operation(write=True) - def _check_unique(self, record: Record) -> None: - """Raise if another record in the record's scope has its object id.""" - if record.object_id is None: - return - other = self._find_record(record.scope, record.object_id) - if other is not None and other.local_key != record.local_key: - raise DuplicateObjectError( - f"Scope {record.scope!r} already has object {record.object_id!r}" - ) - def delete_record(self, local_key: int) -> None: """Delete the record with this local key. @@ -365,8 +507,8 @@ def delete_record(self, local_key: int) -> None: TransactionRequiredError: If no transaction is open in this thread. TransactionPermissionError: If the transaction only reads. """ - with self._operation(write=True): - self._delete_record(local_key) + with self._operation(write=True) as current: + current.delete_record(local_key) def get_record(self, local_key: int) -> Record | None: """Return the record with this local key, or ``None``. @@ -376,8 +518,8 @@ def get_record(self, local_key: int) -> Record | None: Raises: TransactionRequiredError: If no transaction is open in this thread. """ - with self._operation(write=False): - return self._get_record(local_key) + with self._operation(write=False) as current: + return current.get_record(local_key) def find_record(self, scope_id: str, object_id: str) -> Record | None: """Return the scope's record with this :term:`object id`, or ``None``. @@ -387,20 +529,20 @@ def find_record(self, scope_id: str, object_id: str) -> Record | None: Raises: TransactionRequiredError: If no transaction is open in this thread. """ - with self._operation(write=False): - return self._find_record(scope_id, object_id) + with self._operation(write=False) as current: + return current.find_record(scope_id, object_id) def find_by_logical_key(self, scope_id: str, logical_key: str) -> list[Record]: """Return the scope's records with this :term:`logical key`, ordered by local key. - A record matches if its synced side or its :term:`pending edit` has + A record matches if its synced copy or its :term:`local edit` has the key. Several records can share one. Runs inside any transaction. Raises: TransactionRequiredError: If no transaction is open in this thread. """ - with self._operation(write=False): - return self._find_by_logical_key(scope_id, logical_key) + with self._operation(write=False) as current: + return current.find_by_logical_key(scope_id, logical_key) def scope_records( self, scope_id: str, *, states: frozenset[RecordState] | None = None @@ -417,10 +559,10 @@ def scope_records( Raises: TransactionRequiredError: If no transaction is open in this thread. """ - with self._operation(write=False): + with self._operation(write=False) as current: if states is None: - return self._scope_records(scope_id) - return self._scope_records_in(scope_id, states) if states else [] + return current.scope_records(scope_id) + return current.scope_records_in(scope_id, states) if states else [] def removed_records(self, through: int) -> list[Record]: """Return the removed records no feed entry after ``through`` is about. @@ -433,8 +575,8 @@ def removed_records(self, through: int) -> list[Record]: Raises: TransactionRequiredError: If no transaction is open in this thread. """ - with self._operation(write=False): - return self._removed_records(through) + with self._operation(write=False) as current: + return current.removed_records(through) def append_entry( self, @@ -453,8 +595,8 @@ def append_entry( TransactionRequiredError: If no transaction is open in this thread. TransactionPermissionError: If the transaction only reads. """ - with self._operation(write=True): - return self._append_entry(scope_id, kind, local_key, object_id, at) + with self._operation(write=True) as current: + return current.append_entry(scope_id, kind, local_key, object_id, at) def entries(self, after: int, limit: int) -> list[FeedEntry]: """Return up to ``limit`` kept entries numbered above ``after``, in order. @@ -464,8 +606,8 @@ def entries(self, after: int, limit: int) -> list[FeedEntry]: Raises: TransactionRequiredError: If no transaction is open in this thread. """ - with self._operation(write=False): - return self._entries(after, limit) + with self._operation(write=False) as current: + return current.entries(after, limit) def feed_bounds(self) -> tuple[int, int]: """Return the number the feed is trimmed through, and the last number written. @@ -475,8 +617,8 @@ def feed_bounds(self) -> tuple[int, int]: Raises: TransactionRequiredError: If no transaction is open in this thread. """ - with self._operation(write=False): - return self._feed_bounds() + with self._operation(write=False) as current: + return current.feed_bounds() def trim_entries(self, through: int) -> None: """Delete every entry numbered at or below ``through``. @@ -488,8 +630,8 @@ def trim_entries(self, through: int) -> None: TransactionRequiredError: If no transaction is open in this thread. TransactionPermissionError: If the transaction only reads. """ - with self._operation(write=True): - self._trim_entries(through) + with self._operation(write=True) as current: + current.trim_entries(through) def close(self) -> None: """Close the store. Every later transaction raises ``StoreError``. @@ -511,7 +653,7 @@ def close(self) -> None: self._closed = True self._state.wait_for(lambda: self._open == 0) with self._errors(): - self._close() + self._storage.close() def __enter__(self) -> Self: return self @@ -524,81 +666,13 @@ def __exit__( ) -> None: self.close() - def _close(self) -> None: # noqa: B027, most stores hold nothing to release - """Release what the store holds open.""" - - @abstractmethod - def _begin(self, *, write: bool) -> None: - """Start a transaction.""" - - @abstractmethod - def _commit(self) -> None: - """Keep what the transaction wrote.""" - @abstractmethod - def _rollback(self) -> None: - """Discard what the transaction wrote.""" - - @abstractmethod - def _get_scope(self, scope_id: str) -> Scope | None: ... - - @abstractmethod - def _put_scope(self, scope: Scope) -> None: ... - - @abstractmethod - def _delete_scope(self, scope_id: str) -> None: ... - - @abstractmethod - def _scopes(self) -> list[Scope]: ... - - @abstractmethod - def _insert_record(self, record: Record) -> int: ... - - @abstractmethod - def _update_record(self, record: Record) -> None: ... - - @abstractmethod - def _delete_record(self, local_key: int) -> None: ... - - @abstractmethod - def _get_record(self, local_key: int) -> Record | None: ... - - @abstractmethod - def _find_record(self, scope_id: str, object_id: str) -> Record | None: ... - - @abstractmethod - def _find_by_logical_key(self, scope_id: str, logical_key: str) -> list[Record]: ... - - @abstractmethod - def _scope_records(self, scope_id: str) -> list[Record]: ... - - def _scope_records_in(self, scope_id: str, states: frozenset[RecordState]) -> list[Record]: - """The scope's records in one of ``states``, ordered by local key. - - It filters what ``_scope_records`` returns. A store that can - filter where it keeps its records, such as with an index, overrides - it. It never gets an empty set. - """ - return [record for record in self._scope_records(scope_id) if record.state in states] - - @abstractmethod - def _removed_records(self, through: int) -> list[Record]: ... - - @abstractmethod - def _append_entry( - self, - scope_id: str, - kind: FeedEntryKind, - local_key: int | None, - object_id: str | None, - at: datetime, - ) -> FeedEntry: ... - - @abstractmethod - def _entries(self, after: int, limit: int) -> list[FeedEntry]: ... - - @abstractmethod - def _feed_bounds(self) -> tuple[int, int]: ... - - @abstractmethod - def _trim_entries(self, through: int) -> None: ... +def _check_unique(current: StorageTransaction, record: Record) -> None: + """Raise if another record in the record's scope has its object id.""" + if record.object_id is None: + return + other = current.find_record(record.scope, record.object_id) + if other is not None and other.local_key != record.local_key: + raise DuplicateObjectError( + f"Scope {record.scope!r} already has object {record.object_id!r}" + ) diff --git a/src/calendaring_sync/store/memory.py b/src/calendaring_sync/store/memory.py index db1e49d..c71b311 100644 --- a/src/calendaring_sync/store/memory.py +++ b/src/calendaring_sync/store/memory.py @@ -1,25 +1,30 @@ # SPDX-FileCopyrightText: 2026 calendaring-sync contributors # SPDX-License-Identifier: AGPL-3.0-or-later -"""A :term:`store` that keeps everything in memory, and the base of stores that save it.""" +"""A :term:`store` that keeps everything in memory, and can hand its data to a keeper to save.""" from __future__ import annotations +from abc import ABC, abstractmethod 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.error import StoreError, UnexpectedStoreError from calendaring_sync.objects.feed import FeedEntry, FeedEntryKind from calendaring_sync.objects.record import Record, RecordState from calendaring_sync.objects.scope import Scope -from calendaring_sync.store.base import StateStore +from calendaring_sync.store.base import StateStore, Storage, StorageTransaction, _unexpected @dataclass -class _Data: - """Everything a memory store holds: plain containers of the library's frozen dataclasses.""" +class MemoryData: + """Everything a :class:`MemoryStore` holds, as one object. + + A :class:`MemoryKeeper` saves and loads it whole. It's made of plain + containers holding the library's frozen dataclasses, so it pickles; + what's inside isn't part of the API. + """ scopes: dict[str, Scope] = field(default_factory=dict) records: dict[int, Record] = field(default_factory=dict) @@ -28,7 +33,7 @@ class _Data: feed_head: int = 0 feed_trimmed: int = 0 - def copy(self) -> _Data: + def _copy(self) -> MemoryData: # Scopes, records and entries are frozen, so copying the containers # is enough for a snapshot. return replace( @@ -36,146 +41,122 @@ def copy(self) -> _Data: ) -class _View(local): - """This thread's open transaction's view of the data.""" - - #: The view, or ``None`` when the thread has no transaction open. - data: _Data | None = None - +class MemoryKeeper(ABC): + """Keeps a :class:`MemoryStore`'s data somewhere else, such as in a file. -@check_argument_types -class MemoryStore(StateStore): - """A store held in memory and lost when the process ends. + Pass one to :class:`MemoryStore`. It loads the data once, when the + store opens, and saves it each time a write commits; reads come from + memory. If :meth:`save` raises, the write fails and the store keeps + the data it saved last, as long as :meth:`save` doesn't stop halfway: + write to a new file and then replace the old one, for example. - 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. 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 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 - 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 once when it - opens, and a second store opened from those bytes: - - >>> import pickle - >>> from calendaring_sync import Changed, ChangeSet, MemoryStore, Replica - >>> class PickledStore(MemoryStore): - ... def __init__(self, saved=None): - ... super().__init__() - ... self.saved = saved - ... 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 - ... ) - >>> 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' + What :class:`MemoryData` holds can change between releases, so data a + keeper saved with one release may not load in another. Keep sync state + that has to outlive an upgrade in + :class:`~calendaring_sync.store.sqlite.SQLiteStore`, whose file a newer + release reads. """ - _reads_beside_writes = True - - def __init__(self) -> None: - super().__init__() - self._committed = _Data() - self._view = _View() - - @property - def _data(self) -> _Data: - """This thread's transaction's view of the data.""" - # Every hook runs inside a transaction, after _begin set the view. - return cast(_Data, self._view.data) + @abstractmethod + def load(self) -> MemoryData | None: + """Return the data saved last, or ``None`` for a store with nothing saved yet.""" + + @abstractmethod + def save(self, data: MemoryData) -> None: + """Keep the data a write commits, in place of what was saved before.""" + + def error(self, exc: Exception) -> StoreError | None: + """Return the store error for an exception :meth:`load` or :meth:`save` raised. + + Turn the failures of the storage underneath, such as a full disk, + into :class:`~calendaring_sync.error.StoreError`; anything this + returns ``None`` for is reported as a bug. + """ + return None + + +class _MemoryStorage(Storage): + """The data of a memory store, and the keeper it's saved with.""" + + # A read works on the data the last write committed, which nothing changes. + reads_beside_writes = True + + def __init__(self, keeper: MemoryKeeper | None) -> None: + self._keeper = keeper + self._committed = MemoryData() + + def load(self) -> None: + """Take the data the keeper saved last.""" + if self._keeper is not None: + try: + self._committed = self._keeper.load() or MemoryData() + except Exception as exc: + raise self._keeper_error(self._keeper, exc) from exc + + def begin(self, *, write: bool) -> StorageTransaction: + data = self._committed + return _MemoryTransaction(self, data._copy() if write else data, write=write) + + def keep(self, data: MemoryData) -> None: + """Make a write's data the committed data, once the keeper saved it.""" + if self._keeper is not None: + try: + self._keeper.save(data) + except Exception as exc: + raise self._keeper_error(self._keeper, exc) from exc + self._committed = data - def _load(self) -> _Data: - """Return the data the last committed write saved.""" - return self._committed + @staticmethod + def _keeper_error(keeper: MemoryKeeper, exc: Exception) -> StoreError | UnexpectedStoreError: + """The error for a keeper's failure: its own store error, or a bug in the keeper.""" + return keeper.error(exc) or _unexpected(type(keeper), exc) - 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: - data = self._load() - self._view.data = data.copy() if write else data +class _MemoryTransaction(StorageTransaction): + """A transaction over the committed data for a read, or over a copy of it for a write.""" - def _commit(self) -> None: - 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 __init__(self, storage: _MemoryStorage, data: MemoryData, *, write: bool) -> None: + super().__init__(write=write) + self._storage = storage + self._data = data - def _rollback(self) -> None: - # A write worked on a copy that's never saved, so dropping it undoes everything. - self._end_view() + def commit(self) -> None: + if self.write: + self._storage.keep(self._data) - 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, self._view.data = self._view.data, None - return data + def rollback(self) -> None: + # A write worked on a copy that's never kept, so dropping it undoes everything. + pass - def _get_scope(self, scope_id: str) -> Scope | None: + def get_scope(self, scope_id: str) -> Scope | None: return self._data.scopes.get(scope_id) - def _put_scope(self, scope: Scope) -> None: + def put_scope(self, scope: Scope) -> None: self._data.scopes[scope.id] = scope - def _delete_scope(self, scope_id: str) -> None: + def delete_scope(self, scope_id: str) -> None: self._data.scopes.pop(scope_id, None) self._data.records = {k: r for k, r in self._data.records.items() if r.scope != scope_id} - def _scopes(self) -> list[Scope]: + def scopes(self) -> list[Scope]: return sorted(self._data.scopes.values(), key=lambda s: s.id) - def _insert_record(self, record: Record) -> int: + def insert_record(self, record: Record) -> int: self._data.last_key += 1 key = self._data.last_key self._data.records[key] = replace(record, local_key=key) return key - def _update_record(self, record: Record) -> None: + def update_record(self, record: Record) -> None: self._data.records[record.local_key] = record - def _delete_record(self, local_key: int) -> None: + def delete_record(self, local_key: int) -> None: self._data.records.pop(local_key, None) - def _get_record(self, local_key: int) -> Record | None: + def get_record(self, local_key: int) -> Record | None: return self._data.records.get(local_key) - def _find_record(self, scope_id: str, object_id: str) -> Record | None: + def find_record(self, scope_id: str, object_id: str) -> Record | None: return next( ( r @@ -185,17 +166,17 @@ def _find_record(self, scope_id: str, object_id: str) -> Record | None: None, ) - def _find_by_logical_key(self, scope_id: str, logical_key: str) -> list[Record]: + def find_by_logical_key(self, scope_id: str, logical_key: str) -> list[Record]: return [ r - for r in self._scope_records(scope_id) + for r in self.scope_records(scope_id) if logical_key in (r.logical_key, r.edit and r.edit.logical_key) ] - def _scope_records(self, scope_id: str) -> list[Record]: + def scope_records(self, scope_id: str) -> list[Record]: return [r for r in self._ordered_records() if r.scope == scope_id] - def _removed_records(self, through: int) -> list[Record]: + def removed_records(self, through: int) -> list[Record]: return [ r for r in self._ordered_records() @@ -205,7 +186,7 @@ def _removed_records(self, through: int) -> list[Record]: def _ordered_records(self) -> list[Record]: return [record for _, record in sorted(self._data.records.items())] - def _append_entry( + def append_entry( self, scope_id: str, kind: FeedEntryKind, @@ -218,12 +199,65 @@ def _append_entry( self._data.entries.append(entry) return entry - def _entries(self, after: int, limit: int) -> list[FeedEntry]: + def entries(self, after: int, limit: int) -> list[FeedEntry]: return [e for e in self._data.entries if e.number > after][:limit] - def _feed_bounds(self) -> tuple[int, int]: + def feed_bounds(self) -> tuple[int, int]: return self._data.feed_trimmed, self._data.feed_head - def _trim_entries(self, through: int) -> None: + def trim_entries(self, through: int) -> None: self._data.entries = [e for e in self._data.entries if e.number > through] self._data.feed_trimmed = max(self._data.feed_trimmed, through) + + +@check_argument_types +class MemoryStore(StateStore): + """A store held in memory, lost when the process ends unless a keeper saves it. + + 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. 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 and small stores; use + :class:`~calendaring_sync.store.sqlite.SQLiteStore` for large ones. + + Args: + keeper: Where to save the data, such as a file, so the next process + can load it. ``None`` keeps it only in memory. + + Raises: + StoreError: If the keeper can't load the data, as its + :meth:`~MemoryKeeper.error` says. + + Examples: + A keeper that saves the data as pickled bytes, and a second store + loaded from them: + + >>> import pickle + >>> from calendaring_sync import Changed, ChangeSet, MemoryKeeper, MemoryStore, Replica + >>> class Pickled(MemoryKeeper): + ... def __init__(self, saved=None): + ... self.saved = saved + ... def load(self): + ... return pickle.loads(self.saved) if self.saved else None + ... def save(self, data): + ... self.saved = pickle.dumps(data) + >>> keeper = Pickled() + >>> listing = ChangeSet( + ... "work", None, "t1", {"a": Changed("v1", "fp-a", b"Team meeting")}, full_listing=True + ... ) + >>> with Replica(MemoryStore(keeper=keeper)).session( + ... "work", fingerprint_scheme="sha256", media_type="text/plain" + ... ) as session: + ... session.ingest(listing) + >>> store = MemoryStore(keeper=Pickled(keeper.saved)) + >>> Replica(store).find_record("work", "a").content + b'Team meeting' + """ + + def __init__(self, *, keeper: MemoryKeeper | None = None) -> None: + storage = _MemoryStorage(keeper) + super().__init__(storage) + with self._errors(): + storage.load() diff --git a/src/calendaring_sync/store/sqlite.py b/src/calendaring_sync/store/sqlite.py index 9da3975..71a0f8d 100644 --- a/src/calendaring_sync/store/sqlite.py +++ b/src/calendaring_sync/store/sqlite.py @@ -13,14 +13,13 @@ from datetime import datetime from enum import StrEnum from operator import attrgetter -from threading import Lock, local +from threading import Lock 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, StoreBusyError, StoreError, StoreVersionError, @@ -28,7 +27,7 @@ from calendaring_sync.objects.feed import FeedEntry, FeedEntryKind from calendaring_sync.objects.record import Record, RecordState from calendaring_sync.objects.scope import Scope -from calendaring_sync.store.base import StateStore +from calendaring_sync.store.base import StateStore, Storage, StorageTransaction _META = "store_meta" _SCOPES = "scopes" @@ -314,12 +313,6 @@ def _load(reader: _Reader, row: sqlite3.Row) -> Any: _PRIVATE = (":memory:", "") -class _ReadTurn(local): - """Whether this thread's transaction holds the read turn.""" - - taken: bool = False - - 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) @@ -327,166 +320,152 @@ def _connect(path: str | os.PathLike[str], timeout: float) -> sqlite3.Connection return conn -@check_argument_types -class SQLiteStore(StateStore): - """A store in a SQLite database file, shared between threads and processes. +def _meta(conn: sqlite3.Connection, columns: str) -> sqlite3.Row: + """Read ``columns`` from the one row of the store's own data.""" + row: sqlite3.Row | None = conn.execute(f"SELECT {columns} FROM {_META}").fetchone() + if row is None: + raise StoreError(f"The store's {_META} table has lost its row") + return row - 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. 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. +def _version(conn: sqlite3.Connection) -> int: + """The schema version in ``store_meta``, ``0`` for a file with no schema yet.""" + exists = conn.execute( + "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?", (_META,) + ).fetchone() + if exists is None: + return 0 + version: int = _meta(conn, "schema_version")[0] + return version - Args: - path: The database file. It's created, with the schema, if missing. - 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. - """ - - _reads_beside_writes = True +class _SQLiteStorage(Storage): + """A SQLite database file: a connection for writes, and one for reads where the file allows.""" - def __init__(self, path: str | os.PathLike[str], *, timeout: float = DEFAULT_TIMEOUT) -> None: - super().__init__() + def __init__(self, path: str | os.PathLike[str], timeout: float) -> None: + self._path = path + self._timeout = timeout + self._writer: sqlite3.Connection | None = None self._reader: sqlite3.Connection | None = None + self.reads_beside_writes = True # 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. + # process is open while it runs. self._read_turn = Lock() - self._turn_taken = _ReadTurn() - with self._errors(): - self._writer = _connect(path, timeout) - try: - with self._errors(): - 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") - 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 - - @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) - # Errors from the sqlite3 module itself, such as a closed connection, carry no code. - if getattr(exc, "sqlite_errorcode", None) == sqlite3.SQLITE_BUSY: - return StoreBusyError(str(exc)) - return StoreError(str(exc)) - def _meta(self, columns: str) -> sqlite3.Row: - """Read ``columns`` from the one row of the store's own data.""" - row: sqlite3.Row | None = self._conn.execute(f"SELECT {columns} FROM {_META}").fetchone() - if row is None: - raise StoreError(f"The store's {_META} table has lost its row") - return row - - def _version(self) -> int: - """The schema version in ``store_meta``, ``0`` for a file with no schema yet.""" - exists = self._conn.execute( - "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?", (_META,) - ).fetchone() - if exists is None: - return 0 - version: int = self._meta("schema_version")[0] - return version - - def _ensure_schema(self) -> None: - if self._version() == 0: + def open(self) -> None: + """Connect, and create the schema if the file has none.""" + self._writer = _connect(self._path, self._timeout) + self._writer.execute("PRAGMA foreign_keys = ON") + self._ensure_schema(self._writer) + if os.fspath(self._path) in _PRIVATE: + # A second connection would open an empty database of its own. + self.reads_beside_writes = False + return + self._reader = _connect(self._path, self._timeout) + self._reader.execute("PRAGMA query_only = ON") + 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") + + @staticmethod + def _ensure_schema(conn: sqlite3.Connection) -> None: + if _version(conn) == 0: # Checked again under the write lock: another process may have # created the schema in the meantime. - with self.transaction(): - if self._version() == 0: + conn.execute("BEGIN IMMEDIATE") + try: + if _version(conn) == 0: for statement in _SCHEMA: - self._conn.execute(statement) - version = self._version() + conn.execute(statement) + conn.execute("COMMIT") + except BaseException: + if conn.in_transaction: + conn.execute("ROLLBACK") + raise + version = _version(conn) if version > SCHEMA_VERSION: raise StoreVersionError( f"The store has schema version {version}; " f"this calendaring-sync reads up to {SCHEMA_VERSION}" ) - def _close(self) -> None: + def begin(self, *, write: bool) -> StorageTransaction: + writer = cast(sqlite3.Connection, self._writer) + if write or self._reader is None: + writer.execute("BEGIN IMMEDIATE" if write else "BEGIN") + return _SQLiteTransaction(self, writer, write=write) + # Reads take turns on the reader's connection. + self._read_turn.acquire() + try: + self._reader.execute("BEGIN") + except BaseException: + self._read_turn.release() + raise + return _SQLiteTransaction(self, self._reader, write=False, read_turn=True) + + def close(self) -> None: try: if self._reader is not None: self._reader.close() finally: - self._writer.close() + if self._writer is not None: + self._writer.close() - def _begin(self, *, write: bool) -> None: - if self._reads_apart(): - self._take_read_turn() - self._conn.execute("BEGIN IMMEDIATE" if write else "BEGIN") + def error(self, exc: Exception) -> StoreError | None: + if not isinstance(exc, sqlite3.Error): + return None + # Errors from the sqlite3 module itself, such as a closed connection, carry no code. + if getattr(exc, "sqlite_errorcode", None) == sqlite3.SQLITE_BUSY: + return StoreBusyError(str(exc)) + return StoreError(str(exc)) - def _commit(self) -> None: - if self._locks_reads_out and self._writing(): + +class _SQLiteTransaction(StorageTransaction): + """A transaction on one of the storage's connections.""" + + def __init__( + self, + storage: _SQLiteStorage, + conn: sqlite3.Connection, + *, + write: bool, + read_turn: bool = False, + ) -> None: + super().__init__(write=write) + self._storage = storage + self._conn = conn + # Whether the transaction holds the storage's read turn, and so gives it back. + self._read_turn = read_turn + + def commit(self) -> None: + if self.write and self._storage._locks_reads_out: # 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._storage._read_turn.acquire() + self._read_turn = True self._conn.execute("COMMIT") self._end_read() - def _rollback(self) -> None: + def rollback(self) -> None: 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 _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 read turn back, if this thread's transaction took it.""" - if self._turn_taken.taken: - self._turn_taken.taken = False - self._read_turn.release() + """Give the read turn back, if the transaction holds it.""" + if self._read_turn: + self._read_turn = False + self._storage._read_turn.release() def _select( self, cls: type[_Row], table: str, clause: str, params: tuple[Any, ...] = () @@ -502,38 +481,38 @@ def _select_one( found = self._select(cls, table, clause, params) return found[0] if found else None - def _get_scope(self, scope_id: str) -> Scope | None: + def get_scope(self, scope_id: str) -> Scope | None: return self._select_one(Scope, _SCOPES, "WHERE id = ?", (scope_id,)) - def _put_scope(self, scope: Scope) -> None: + def put_scope(self, scope: Scope) -> None: self._conn.execute(_PUT_SCOPE, _values(scope, _SCOPE_COLUMNS)) - def _delete_scope(self, scope_id: str) -> None: + def delete_scope(self, scope_id: str) -> None: self._conn.execute(f"DELETE FROM {_SCOPES} WHERE id = ?", (scope_id,)) - def _scopes(self) -> list[Scope]: + def scopes(self) -> list[Scope]: return self._select(Scope, _SCOPES, "ORDER BY id") - def _insert_record(self, record: Record) -> int: + def insert_record(self, record: Record) -> int: cursor = self._conn.execute(_INSERT_RECORD, _values(record, _RECORD_VALUES)) # Typed as optional, but an INSERT always sets it to the new row's id. return cast(int, cursor.lastrowid) - def _update_record(self, record: Record) -> None: + def update_record(self, record: Record) -> None: self._conn.execute(_UPDATE_RECORD, (*_values(record, _RECORD_VALUES), record.local_key)) - def _delete_record(self, local_key: int) -> None: + def delete_record(self, local_key: int) -> None: self._conn.execute(f"DELETE FROM {_RECORDS} WHERE local_key = ?", (local_key,)) - def _get_record(self, local_key: int) -> Record | None: + def get_record(self, local_key: int) -> Record | None: return self._select_one(Record, _RECORDS, "WHERE local_key = ?", (local_key,)) - def _find_record(self, scope_id: str, object_id: str) -> Record | None: + def find_record(self, scope_id: str, object_id: str) -> Record | None: return self._select_one( Record, _RECORDS, "WHERE scope = ? AND object_id = ?", (scope_id, object_id) ) - def _find_by_logical_key(self, scope_id: str, logical_key: str) -> list[Record]: + def find_by_logical_key(self, scope_id: str, logical_key: str) -> list[Record]: # A UNION, not an OR: SQLite answers each half from its own index, # and would scan the table for the OR. return self._select( @@ -545,10 +524,10 @@ def _find_by_logical_key(self, scope_id: str, logical_key: str) -> list[Record]: (scope_id, logical_key, scope_id, logical_key), ) - def _scope_records(self, scope_id: str) -> list[Record]: + def scope_records(self, scope_id: str) -> list[Record]: return self._select(Record, _RECORDS, "WHERE scope = ? ORDER BY local_key", (scope_id,)) - def _scope_records_in(self, scope_id: str, states: frozenset[RecordState]) -> list[Record]: + def scope_records_in(self, scope_id: str, states: frozenset[RecordState]) -> list[Record]: marks = ", ".join("?" for _ in states) return self._select( Record, @@ -557,7 +536,7 @@ def _scope_records_in(self, scope_id: str, states: frozenset[RecordState]) -> li (scope_id, *states), ) - def _removed_records(self, through: int) -> list[Record]: + def removed_records(self, through: int) -> list[Record]: return self._select( Record, _RECORDS, @@ -565,7 +544,7 @@ def _removed_records(self, through: int) -> list[Record]: (RecordState.REMOVED, through), ) - def _append_entry( + def append_entry( self, scope_id: str, kind: FeedEntryKind, @@ -574,20 +553,65 @@ def _append_entry( at: datetime, ) -> FeedEntry: self._conn.execute(f"UPDATE {_META} SET feed_head = feed_head + 1") - number: int = self._meta("feed_head")[0] + number: int = _meta(self._conn, "feed_head")[0] entry = FeedEntry(number, scope_id, kind, local_key, object_id, at) self._conn.execute(_INSERT_ENTRY, _values(entry, _FEED_COLUMNS)) return entry - def _entries(self, after: int, limit: int) -> list[FeedEntry]: + def entries(self, after: int, limit: int) -> list[FeedEntry]: return self._select( FeedEntry, _FEED, "WHERE number > ? ORDER BY number LIMIT ?", (after, limit) ) - def _feed_bounds(self) -> tuple[int, int]: - trimmed, head = self._meta("feed_trimmed, feed_head") + def feed_bounds(self) -> tuple[int, int]: + trimmed, head = _meta(self._conn, "feed_trimmed, feed_head") return trimmed, head - def _trim_entries(self, through: int) -> None: + def trim_entries(self, through: int) -> None: self._conn.execute(f"DELETE FROM {_FEED} WHERE number <= ?", (through,)) self._conn.execute(f"UPDATE {_META} SET feed_trimmed = max(feed_trimmed, ?)", (through,)) + + +@check_argument_types +class SQLiteStore(StateStore): + """A store in a SQLite database file, shared between threads and processes. + + 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. 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 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. + """ + + def __init__(self, path: str | os.PathLike[str], *, timeout: float = DEFAULT_TIMEOUT) -> None: + storage = _SQLiteStorage(path, timeout) + super().__init__(storage) + try: + with self._errors(): + storage.open() + except BaseException: + storage.close() + raise diff --git a/src/calendaring_sync/tests/_storage.py b/src/calendaring_sync/tests/_storage.py new file mode 100644 index 0000000..e67dff3 --- /dev/null +++ b/src/calendaring_sync/tests/_storage.py @@ -0,0 +1,25 @@ +# SPDX-FileCopyrightText: 2026 calendaring-sync contributors +# SPDX-License-Identifier: AGPL-3.0-or-later + +"""A storage written against the storage interface alone, as a third party would.""" + +from __future__ import annotations + +from calendaring_sync.store.base import Storage, StorageTransaction +from calendaring_sync.store.memory import _MemoryStorage + + +class OneAtATime(Storage): + """A storage that keeps every default of :class:`Storage`. + + So every transaction waits for the one before it, nothing is released + on close, and every exception is reported as a bug. It hands the data + operations to a memory storage, so what's under test is what the store + adds to a storage it knows nothing about. + """ + + def __init__(self) -> None: + self._data = _MemoryStorage(None) + + def begin(self, *, write: bool) -> StorageTransaction: + return self._data.begin(write=write) diff --git a/src/calendaring_sync/tests/conftest.py b/src/calendaring_sync/tests/conftest.py index e9b9484..560a070 100644 --- a/src/calendaring_sync/tests/conftest.py +++ b/src/calendaring_sync/tests/conftest.py @@ -1,7 +1,11 @@ # SPDX-FileCopyrightText: 2026 calendaring-sync contributors # SPDX-License-Identifier: AGPL-3.0-or-later -"""What every test file runs on: both stores, and the sync and async adapters.""" +"""What every test file runs on: the stores, and the sync and async adapters. + +The stores are the two that ship and one over a storage written as a third +party would write it. +""" from __future__ import annotations @@ -11,6 +15,7 @@ import pytest from calendaring_sync import MemoryStore, SQLiteStore, StateStore +from calendaring_sync.tests._storage import OneAtATime @pytest.fixture @@ -19,15 +24,22 @@ def db_path(tmp_path: Path) -> Path: return tmp_path / "s.db" -@pytest.fixture(params=[MemoryStore, SQLiteStore], ids=lambda cls: cls.__name__) +@pytest.fixture(params=[MemoryStore, SQLiteStore, OneAtATime], ids=lambda cls: cls.__name__) def make_store(request: pytest.FixtureRequest, db_path: Path) -> Iterator[Callable[[], StateStore]]: - """Open handles on one store, as separate processes would: one shared - ``MemoryStore``, or a new connection to one SQLite file each time.""" + """Open handles on one store, as separate processes would: a new + connection to one SQLite file each time, or one shared store in memory.""" opened: list[StateStore] = [] - memory = MemoryStore() + shared: list[StateStore] = [] def make() -> StateStore: - store: StateStore = memory if request.param is MemoryStore else SQLiteStore(db_path) + if request.param is SQLiteStore: + store: StateStore = SQLiteStore(db_path) + else: + if not shared: + shared.append( + MemoryStore() if request.param is MemoryStore else StateStore(OneAtATime()) + ) + store = shared[0] opened.append(store) return store diff --git a/src/calendaring_sync/tests/test_sync_unit.py b/src/calendaring_sync/tests/test_sync_unit.py index e8a3aed..664d67f 100644 --- a/src/calendaring_sync/tests/test_sync_unit.py +++ b/src/calendaring_sync/tests/test_sync_unit.py @@ -31,7 +31,7 @@ from pathlib import Path from threading import Event from types import SimpleNamespace -from typing import Any, Literal +from typing import Any, Literal, cast from unittest.mock import patch import pytest @@ -58,6 +58,8 @@ LocalEdit, LogicalKeyTaken, LoneSurrogateError, + MemoryData, + MemoryKeeper, MemoryStore, Merge, ObjectIdAssignedError, @@ -73,6 +75,7 @@ Removed, RemovedRecordError, Replica, + ResolvedConflicts, Scope, ScopeGoneError, ScopeMismatchError, @@ -83,6 +86,7 @@ StaleChangeSetError, StaleConflictError, StateStore, + StorageTransaction, StoreBusyError, StoreError, StoreVersionError, @@ -104,10 +108,20 @@ from calendaring_sync.adapters.caldav import AsyncCalDAVAdapter, CalDAVAdapter, _CalDAVFlow from calendaring_sync.adapters.jmap import AsyncJMAPAdapter, JMAPAdapter, _JMAPFlow 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.store.memory import _MemoryStorage +from calendaring_sync.store.sqlite import ( + _FEED, + _META, + _RECORDS, + _SCHEMA, + _reader, + _SQLiteStorage, + _SQLiteTransaction, + _version, +) from calendaring_sync.tests import _caldav_scenarios, _examples, _jmap_scenarios, benchmark from calendaring_sync.tests._scenarios import done, run +from calendaring_sync.tests._storage import OneAtATime from calendaring_sync.tests._tape import Tape _SCOPE = "cal" @@ -327,7 +341,7 @@ def _local_record(state: RecordState, object_id: str | None = None, local_key: i ) -# The server's side of a conflict where the server removed the object. +# The server's copy in a conflict where the server removed the object. _GONE = Conflict(ConflictKind.REMOVED_REMOTELY, None, None, None, frozenset(), None, _START) _PENDING = _local_record(RecordState.PENDING) _CONFLICTED = _local_record(RecordState.CONFLICT) @@ -338,8 +352,8 @@ def _local_record(state: RecordState, object_id: str | None = None, local_key: i "synced with an edit": replace(_local_record(RecordState.SYNCED), edit=_EDIT), "conflict without its server side": replace(_CONFLICTED, conflict=None), "synced with a server side": replace(_local_record(RecordState.SYNCED), conflict=_CONFLICT), - "no synced content for a change": replace(_PENDING, content=None), - "synced content for a create": replace(_PENDING, edit=replace(_EDIT, kind=EditKind.CREATE)), + "no synced copy for a change": replace(_PENDING, content=None), + "synced copy for a create": replace(_PENDING, edit=replace(_EDIT, kind=EditKind.CREATE)), "a delete holding content": replace(_PENDING, edit=replace(_EDIT, kind=EditKind.DELETE)), "a change without content": replace(_PENDING, edit=replace(_EDIT, content=None)), "a conflict about the wrong edit": replace( @@ -654,7 +668,7 @@ def _assert_refused( """ before = _snapshot(replica, store) with ( - patch.object(store, "_begin", wraps=store._begin) as begin, + patch.object(store._storage, "begin", wraps=store._storage.begin) as begin, pytest.raises(error, match=match) as raised, ): call() @@ -666,6 +680,40 @@ def _assert_refused( assert _snapshot(replica, store) == before +def _current(store: StateStore) -> StorageTransaction: + """The storage's transaction this thread has open in ``store``.""" + current = store._thread.current + assert current is not None + return current + + +def _writer(store: StateStore) -> sqlite3.Connection: + """The connection a SQLite store writes on.""" + return cast(sqlite3.Connection, cast(_SQLiteStorage, store._storage)._writer) + + +def _break( + store: StateStore, monkeypatch: pytest.MonkeyPatch, step: str, error: BaseException +) -> None: + """Make ``step`` raise ``error``: the storage's ``begin``, or each transaction's method.""" + storage = store._storage + + def fail(*args: object, **kwargs: object) -> None: + raise error + + if step == "begin": + monkeypatch.setattr(storage, "begin", fail) + return + begin = storage.begin + + def begin_broken(*, write: bool) -> StorageTransaction: + current = begin(write=write) + setattr(current, step, fail) + return current + + monkeypatch.setattr(storage, "begin", begin_broken) + + # Examples that continue from the replica the Replica class example builds. _CONTINUING = ("calendaring_sync.replica.Replica.", "calendaring_sync.session.Session") @@ -919,6 +967,13 @@ def test_sqlite(self, tmp_path, monkeypatch, failing, calls, scopes): with SQLiteStore(tmp_path / "sync.db") as store: assert [scope.id for scope in Replica(store).scopes()] == scopes + def test_keeper(self, tmp_path, monkeypatch): + ran = _examples.run("keeper", monkeypatch, tmp_path) + assert ran["meeting"].content == b"Team meeting" + keeper = ran["FileKeeper"](tmp_path / "sync.pickle") + assert isinstance(keeper.error(OSError("disk full")), StoreError) + assert keeper.error(ValueError("a bug")) is None + class TestPackage: @pytest.mark.parametrize("test", _docstring_examples(), ids=lambda test: test.name) @@ -1155,11 +1210,6 @@ def test_incremental_without_token_is_refused(self, replica, store): "needs a full listing", ) - def test_unchanged_outside_listing_is_refused(self, replica, store): - _seeded(replica) - page = _page("T1", "T2", {"a": Unchanged()}) - _assert_refused(replica, store, lambda: _sync(replica, page), ValueError, "outside a full") - @pytest.mark.parametrize("object_id", ["zz", "a"]) def test_unchanged_needs_a_live_record_with_a_version(self, replica, store, object_id): _a_removed(replica) @@ -1965,7 +2015,7 @@ def test_a_refused_change_is_not_the_holder_of_its_own_key(self, replica): def test_a_holder_in_conflict_holds_the_key_of_the_server_side(self, replica, refused): _sync(_seeded(replica), _page("T1", "T2", {"b": _changed("fb", "v2", logical_key="u")})) key = _refused_edit(replica, refused) - # b's synced copy keeps u, but the server's side of its conflict moved to w. + # b's synced copy keeps u, but the server's copy in its conflict moved to w. _change(replica, _key(replica, "b"), "fl") head = replica.feed_head() _sync(replica, _page("T3", "T4", {"b": _changed("fx", "v3", logical_key="w")})) @@ -2581,14 +2631,14 @@ def resolver(record: Record) -> KeepServer | None: seen.append(record.local_key) return None if record.local_key == a else KeepServer() - assert replica.resolve_conflicts(resolver) == 2 + assert replica.resolve_conflicts(resolver) == ResolvedConflicts((b, other), (a,)) assert seen == [a, b, other] assert [_summary(replica.record(key)) for key in (a, b, other)] == [ _conflicted(EditKind.CHANGE, ConflictKind.BOTH_CHANGED), _SYNCED, _SYNCED, ] - assert replica.resolve_conflicts(lambda record: None) == 0 + assert replica.resolve_conflicts(lambda record: None) == ResolvedConflicts(left_open=(a,)) def test_resolve_conflicts_skips_a_conflict_that_moved_on(self, replica): a, b = _two_conflicts(replica) @@ -2599,9 +2649,9 @@ def resolver(record: Record) -> KeepServer: _sync(replica, _page("T3", "T4", {"b": _changed("fy", "v4")})) return KeepServer() - assert replica.resolve_conflicts(resolver) == 1 + assert replica.resolve_conflicts(resolver) == ResolvedConflicts((a,), changed=(b,)) assert replica.record(b).state is RecordState.CONFLICT - assert replica.resolve_conflicts(lambda record: KeepServer()) == 1 + assert replica.resolve_conflicts(lambda record: KeepServer()) == ResolvedConflicts((b,)) assert replica.record(b).content == _content("fy") def test_an_error_from_the_resolver_keeps_earlier_resolutions(self, replica): @@ -2646,6 +2696,15 @@ 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_a_change_set_keeps_its_own_changes(self): + changes: dict[str, Change] = {"a": Removed()} + change_set = _page("T1", "T2", changes) + # What's added to the caller's mapping afterwards was never checked, so it isn't ingested. + changes["b"] = Unchanged() + assert dict(change_set.changes) == {"a": Removed()} + with pytest.raises(TypeError): + change_set.changes["b"] = Unchanged() # type: ignore[index] + 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]] = [ @@ -2686,6 +2745,11 @@ def test_wrong_types_and_text_are_refused(self, replica, store, db_path): "^changes must be mapping of str to Changed, Removed or Unchanged, not list", ), (lambda: _page(None, "T1", {_BAD_TEXT: Removed()}), LoneSurrogateError, "^object_ids "), + ( + lambda: _page("T1", "T2", {"a": Unchanged()}), + ValueError, + "^Unchanged object 'a' outside a full listing$", + ), ( lambda: _page(None, "T1", {"a": _wrong("removed")}), TypeError, @@ -2842,7 +2906,11 @@ def test_default_clock_is_utc_now(self, store): class TestStores: def test_every_operation_is_in_the_tables(self): public = {name for name in vars(StateStore) if not name.startswith("_")} - assert public - {"transaction", "close"} == _READS.keys() | _WRITES.keys() + operations = public - {"transaction", "close"} + assert operations == _READS.keys() | _WRITES.keys() + # A storage's transaction offers the same operations, plus its own ends. + offered = {name for name in vars(StorageTransaction) if not name.startswith("_")} + assert offered - {"commit", "rollback", "scope_records_in"} == operations @pytest.mark.parametrize("call", [*_READS.values(), *_WRITES.values()], ids=[*_READS, *_WRITES]) def test_operations_outside_a_transaction_raise(self, store, call): @@ -2873,7 +2941,7 @@ def test_transactions_do_not_nest(self, replica, store): [ (lambda path: MemoryStore(), True), (lambda path: SQLiteStore(path), True), - (lambda path: _OneAtATime(), False), + (lambda path: StateStore(OneAtATime()), False), (lambda path: SQLiteStore(":memory:"), False), ], ids=["memory", "sqlite", "memory one at a time", "sqlite in memory"], @@ -2996,15 +3064,16 @@ def test_object_ids_are_unique_per_scope(self, replica, store): ) def test_scope_records_by_state(self, replica, store, states, expected): _edited(replica) - with ( - store.transaction(write=False), - patch.object(store, "_scope_records", wraps=store._scope_records) as every, - patch.object(store, "_scope_records_in", wraps=store._scope_records_in) as some, - ): - records = store.scope_records(_SCOPE, states=states) + with store.transaction(write=False): + current = _current(store) + with ( + patch.object(current, "scope_records", wraps=current.scope_records) as every, + patch.object(current, "scope_records_in", wraps=current.scope_records_in) as some, + ): + records = store.scope_records(_SCOPE, states=states) assert [r.object_id for r in records] == expected assert replica.records(_SCOPE, states=states) == records - # An empty set is answered by the base class, so no store sees one. + # An empty set is answered by the store, so no storage sees one. assert (every.called or some.called) == (states != frozenset()) assert some.called == bool(states) @@ -3043,16 +3112,12 @@ def test_updating_a_missing_record_raises(self, store): 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"]) + @pytest.mark.parametrize("step", ["begin", "commit", "scopes"]) def test_a_failure_inside_the_store_is_unexpected_and_keeps_nothing( - self, store, monkeypatch, hook, write + self, store, monkeypatch, step, write ): - bug = RuntimeError(f"{hook} has a bug") - - def fail(*args: object, **kwargs: object) -> None: - raise bug - - monkeypatch.setattr(store, hook, fail) + bug = RuntimeError(f"{step} has a bug") + _break(store, monkeypatch, step, bug) with pytest.raises(UnexpectedStoreError, match=re.escape(ISSUES_URL)) as raised: with _write_with_scope(store) if write else store.transaction(write=False): store.scopes() @@ -3064,16 +3129,34 @@ def fail(*args: object, **kwargs: object) -> None: with store.transaction(): pass - def test_interrupts_pass_through_unchanged(self, store, monkeypatch): - def interrupt() -> None: - raise KeyboardInterrupt + def test_a_bug_in_someone_elses_storage_is_reported_to_them(self, monkeypatch): + storage = OneAtATime() + monkeypatch.setattr(type(storage), "__module__", "their_storage") + store = StateStore(storage) + _break(store, monkeypatch, "scopes", RuntimeError("a bug")) + with pytest.raises(UnexpectedStoreError) as raised, store.transaction(write=False): + store.scopes() + assert str(raised.value).endswith("This is a bug in their_storage.OneAtATime") + assert ISSUES_URL not in str(raised.value) - monkeypatch.setattr(store, "_scopes", interrupt) + def test_interrupts_pass_through_unchanged(self, store, monkeypatch): + _break(store, monkeypatch, "scopes", KeyboardInterrupt()) with pytest.raises(KeyboardInterrupt), store.transaction(write=False): store.scopes() + monkeypatch.undo() with store.transaction(write=False): assert store.feed_bounds() == (0, 0) + @pytest.mark.parametrize("first", [KeyboardInterrupt(), RuntimeError("the block failed")]) + def test_a_failed_rollback_keeps_what_ended_the_block(self, store, monkeypatch, first): + _break(store, monkeypatch, "rollback", RuntimeError("rollback broke")) + with pytest.raises(type(first)) as raised, store.transaction(): + raise first + assert raised.value is first + assert "Rolling the transaction back failed too" in raised.value.__notes__[0] + monkeypatch.undo() + assert _scope_ids(store) == [] + def test_store_as_context_manager(self, make_store): with make_store() as store: assert _scope_ids(store) == [] @@ -3083,65 +3166,89 @@ def test_a_memory_store_lets_go_of_old_data(self): with ThreadPoolExecutor(1) as pool: # The thread stays alive, idle, after its read. pool.submit(_scope_ids, store).result() - old = weakref.ref(store._load()) + old = weakref.ref(cast(_MemoryStorage, store._storage)._committed) _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) + keeper = _Pickled(path) + store = MemoryStore(keeper=keeper) replica = _edited(Replica(store, clock=clock)) expected = _snapshot(replica, store) - # Reads never save. - with patch.object(store, "_save", wraps=store._save) as save: + # Reads never save, and never load again. + with ( + patch.object(keeper, "save", wraps=keeper.save) as save, + patch.object(keeper, "load", wraps=keeper.load) as load, + ): replica.records(_SCOPE) replica.read_feed() - assert not save.called + assert not (save.called or load.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) + keeper = _Pickled(path) + reopened = MemoryStore(keeper=keeper) 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", - ) + + # A save that fails fails the write, keeps the last data and leaves the store + # usable; the keeper says which of its failures are store errors. + def add_scope() -> None: + with reopened.transaction(): + reopened.put_scope(replace(_new_scope(), id=_OTHER)) + + with patch.object(keeper, "save", side_effect=OSError("disk full")): + _assert_refused(again, reopened, add_scope, StoreError, "^disk full$") _sync(again, _page("T2", "T3", {"a": Removed()})) expected = _snapshot(again, reopened) - third = _PickledStore(path) + third = MemoryStore(keeper=_Pickled(path)) 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 + @pytest.mark.parametrize("step", ["load", "save"]) + def test_a_keepers_failure_it_doesnt_name_is_a_bug(self, step): + class Failing(MemoryKeeper): + def load(self) -> MemoryData | None: + if step == "load": + raise OSError("disk full") + return None + + def save(self, data: MemoryData) -> None: + raise OSError("disk full") + + bug = r"\.Failing failed unexpectedly with OSError" + if step == "load": + with pytest.raises(UnexpectedStoreError, match=bug): + MemoryStore(keeper=Failing()) + return + store = MemoryStore(keeper=Failing()) + with pytest.raises(UnexpectedStoreError, match=bug): + _add_scope(store) -class _PickledStore(MemoryStore): - """A memory store that keeps its data in a file with :mod:`pickle`, read when it opens.""" +class _Pickled(MemoryKeeper): + """A keeper that saves a memory store's data in a file with :mod:`pickle`.""" def __init__(self, path: Path) -> None: - super().__init__() self._path = path - if path.exists(): - super()._save(pickle.loads(path.read_bytes())) - def _save(self, data: _Data) -> None: + def load(self) -> MemoryData | None: + loaded: MemoryData | None = ( + pickle.loads(self._path.read_bytes()) if self._path.exists() else None + ) + return loaded + + def save(self, data: MemoryData) -> None: # Written beside the file, then moved over it, so a failure halfway # leaves the last saved data. - written = self._path.with_suffix(".new") + written = self._path.with_name(self._path.name + ".new") written.write_bytes(pickle.dumps(data)) written.replace(self._path) - super()._save(data) + + def error(self, exc: Exception) -> StoreError | None: + return StoreError(str(exc)) if isinstance(exc, OSError) else None # Dataclasses the SQLite store can't map: a column holds one type, and the @@ -3246,20 +3353,23 @@ def test_a_full_disk_raises_store_error_and_keeps_nothing(self, db_path, clock): replica = _seeded(Replica(store, clock=clock)) before = _snapshot(replica, store) # SQLite reports a file at its page limit as a full disk. - pages = store._conn.execute("PRAGMA page_count").fetchone()[0] - store._conn.execute(f"PRAGMA max_page_count = {pages}") + writer = _writer(store) + pages = writer.execute("PRAGMA page_count").fetchone()[0] + writer.execute(f"PRAGMA max_page_count = {pages}") many = {f"o{n}": _changed(f"f{n}", content=b"x" * 10_000) for n in range(20)} with pytest.raises(StoreError, match="full"): _sync(replica, _page("T1", "T2", many)) assert _snapshot(replica, store) == before - def test_a_failed_close_raises_store_error_and_still_closes(self, db_path): - class FailingClose(SQLiteStore): - def _close(self) -> None: - super()._close() - raise sqlite3.OperationalError("disk I/O error") + def test_a_failed_close_raises_store_error_and_still_closes(self, db_path, monkeypatch): + store = SQLiteStore(db_path) + close = store._storage.close - store = FailingClose(db_path) + def failing_close() -> None: + close() + raise sqlite3.OperationalError("disk I/O error") + + monkeypatch.setattr(store._storage, "close", failing_close) with pytest.raises(StoreError, match="disk I/O"): store.close() store.close() @@ -3316,7 +3426,7 @@ def test_a_commit_waits_for_reads_here_where_it_locks_them_out( with SQLiteStore(db_path, timeout=_SHORT_TIMEOUT) as 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] + spills = _writer(store).execute("PRAGMA cache_spill").fetchone()[0] assert bool(spills) is not locks_reads_out with ( ThreadPoolExecutor(1) as pool, @@ -3344,17 +3454,46 @@ def test_interrupted_ingest_leaves_the_token_after_reopening(self, db_path, cloc assert replica.scope(_SCOPE).token == "T1" assert _states(replica) == {"a": RecordState.SYNCED, "b": RecordState.SYNCED} + # A statement that fails rolls the schema back; one that ends the + # transaction early leaves the schema, and its own COMMIT fails. + @pytest.mark.parametrize( + "last, kept", [("CREATE TABLE broken (", False), ("COMMIT", True)], ids=["fails", "ends"] + ) + def test_a_schema_that_fails_halfway_leaves_none(self, db_path, monkeypatch, last, kept): + monkeypatch.setattr("calendaring_sync.store.sqlite._SCHEMA", (*_SCHEMA, last)) + with pytest.raises(StoreError): + SQLiteStore(db_path) + monkeypatch.undo() + with closing(sqlite3.connect(db_path)) as connection: + tables = connection.execute("SELECT count(*) FROM sqlite_master").fetchone()[0] + assert bool(tables) is kept + with SQLiteStore(db_path) as store: + assert _scope_ids(store) == [] + + def test_a_read_that_cant_begin_gives_its_turn_back(self, db_path, monkeypatch): + class Refusing: + def execute(self, *args: object) -> None: + raise sqlite3.OperationalError("disk I/O error") + + with SQLiteStore(db_path) as store: + monkeypatch.setattr(store._storage, "_reader", Refusing()) + with pytest.raises(StoreError, match="disk I/O"): + _enter(store.transaction(write=False)) + monkeypatch.undo() + # A turn kept by the failed read would hold every later read forever. + assert not cast(_SQLiteStorage, store._storage)._read_turn.locked() + assert _scope_ids(store) == [] + def test_schema_created_by_another_process_first(self, db_path, monkeypatch): SQLiteStore(db_path).close() - real = SQLiteStore._version looks: list[int] = [] - def version(store: SQLiteStore) -> int: + def version(conn: sqlite3.Connection) -> int: # The first look sees an empty file; under the write lock the schema is there. looks.append(0) - return 0 if len(looks) == 1 else real(store) + return 0 if len(looks) == 1 else _version(conn) - monkeypatch.setattr(SQLiteStore, "_version", version) + monkeypatch.setattr("calendaring_sync.store.sqlite._version", version) with SQLiteStore(db_path) as store: assert _scope_ids(store) == [] assert len(looks) == 3 @@ -3368,5 +3507,5 @@ def test_newer_schema_is_refused(self, db_path): def test_rollback_after_the_transaction_already_ended(self, db_path): with SQLiteStore(db_path) as store: with pytest.raises(OSError), store.transaction(): - store._conn.execute("COMMIT") + cast(_SQLiteTransaction, _current(store))._conn.execute("COMMIT") raise OSError("after an implicit end")