replica: Sessions, ingesting, local edits, pushing, resolving conflicts, and the change feed#

Replica is the entry point. Record your local edits as you make them. To sync, open a session for a scope and ingest the change sets an adapter fetched. After the last page, the adapter pushes your edits in the same session: it claims each one, sends it, and settles it with how the push ended. Then read the change feed to learn what changed and which conflicts came up, and resolve them with resolve() or resolve_conflicts().

The core’s entry point: sessions, pushes, local edits, conflicts, and the change feed.

Every sync rule lives here, on top of any StateStore.

class calendaring_sync.replica.Replica(store: StateStore, *, clock: Callable[[], datetime] | None = None)[source]#

Bases: object

The sync state of every scope in one store.

This is the class your application and the adapter use: the adapter ingests what the server changed through a session(), and your application records its edits and reads records and the change feed. The store underneath only keeps the data.

Each method runs in a store transaction of its own, so two calls can see another thread’s or process’s write between them. To read several things at one point in time, open the store’s transaction() with write=False and call the store’s own methods inside it. A replica’s methods can’t run inside a transaction already open in the same thread; they raise TransactionError.

Any method can also raise the store’s errors: StoreBusyError when another process kept the store busy, which is safe to retry, StoreError when the store can’t read or write its data, and UnexpectedStoreError for a bug in the store. None of them changes anything.

Parameters:

  • store – Where the state is kept.

  • clock – Returns the current time for every timestamp. Defaults to the current time in UTC.

Examples

A replica on a store in memory, with one scope that a full listing filled with one object. The examples on each method start from this replica. An adapter computes fingerprints from the content; these are short made-up ones.

>>> from calendaring_sync import Changed, ChangeSet, MemoryStore, Replica
>>> replica = Replica(MemoryStore())
>>> listing = ChangeSet(
...     "work", None, "t1", {"a": Changed("v1", "fp-a", b"Team meeting")}, full_listing=True
... )
>>> with replica.session(
...     "work", fingerprint_scheme="sha256", media_type="text/plain"
... ) as session:
...     session.ingest(listing)
>>> replica.find_record("work", "a").content
b'Team meeting'
>>> [(entry.kind, entry.object_id) for entry in replica.read_feed()]
[(<FeedEntryKind.CHANGED: 'changed'>, 'a')]
change(local_key: int, content: bytes, *, fingerprint: str, containers: frozenset[str] = frozenset({}), logical_key: str | None = None) → None[source]#

Record a change to an object.

A change back to the synced content 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 create().

Runs in a write transaction of its own.

Raises:

Examples

>>> key = replica.find_record("work", "a").local_key
>>> replica.change(key, b"Team meeting, moved", fingerprint="fp-a2")
>>> replica.record(key).edit.content
b'Team meeting, moved'

A change back to the synced content cancels the edit:

>>> replica.change(key, b"Team meeting", fingerprint="fp-a")
>>> replica.record(key).state
<RecordState.SYNCED: 'synced'>
create(scope_id: str, content: bytes, *, fingerprint: str, containers: frozenset[str] = frozenset({}), logical_key: str | None = None, object_id: str | None = None) → int[source]#

Record the creation of an object, and return its local key.

The record is pending until a push creates the object on the server. A create at the object id of a record the server removed takes that record over, as the server does when it has an object at that id again: it keeps its local key and drops the removed copy. Build fingerprint, containers and logical_key with the adapter’s function for them, the same one it uses for the server’s changes; built any other way, an identical copy on the server looks different.

Runs in a write transaction of its own.

Parameters:

  • scope_id – The scope the object belongs to.

  • content – The object’s content.

  • fingerprint – The content’s fingerprint.

  • containers – The containers the object belongs to.

  • logical_key – The object’s logical key, if it has one.

  • object_id – The object id the create chose, where the server takes ids from its clients; None where the server picks one.

Raises:

Examples

>>> key = replica.create("work", b"Lunch", fingerprint="fp-l", object_id="b")
>>> replica.record(key).state
<RecordState.PENDING: 'pending'>
>>> replica.record(key).edit.content
b'Lunch'
delete(local_key: int) → None[source]#

Record the deletion of an object.

Deleting an object you created that the server doesn’t have yet removes its record, leaving nothing to push.

Runs in a write transaction of its own.

Raises:

Examples

>>> key = replica.find_record("work", "a").local_key
>>> replica.delete(key)
>>> replica.record(key).edit.kind
<EditKind.DELETE: 'delete'>
discard(local_key: int) → None[source]#

Discard an object’s pending edit, so it matches the server’s copy again.

Discarding a create removes its record. A record with no edit is left as it is.

Runs in a write transaction of its own.

Raises:

Examples

>>> key = replica.find_record("work", "a").local_key
>>> replica.delete(key)
>>> replica.discard(key)
>>> replica.record(key).state
<RecordState.SYNCED: 'synced'>
drop_scope(scope_id: str) → DroppedScope[source]#

Drop a scope and every record in it.

Runs in a write transaction of its own.

Returns:

The records holding local work that the drop discarded.

Raises:

Examples

>>> dropped = replica.drop_scope("work")
>>> dropped.pending, dropped.conflicts
((), ())
>>> replica.scopes()
[]
feed_head() → int[source]#

Return the number of the last change feed entry written, 0 for none.

Start a new cursor here after reading every record.

Runs in a read transaction of its own.

Raises:

TransactionError – If a store transaction is already open in this thread.

Examples

>>> replica.feed_head()
1
find_by_logical_key(scope_id: str, logical_key: str) → list[Record][source]#

Return the scope’s records with this logical key, ordered by local key.

A record matches if its synced side or its pending edit has the key. Several records can share one, and removed records are included until they’re purged.

Runs in a read transaction of its own.

Raises:

Examples

>>> key = replica.create("work", b"Lunch", fingerprint="fp-l", logical_key="uid-1")
>>> found = replica.find_by_logical_key("work", "uid-1")
>>> [record.local_key for record in found] == [key]
True
find_record(scope_id: str, object_id: str) → Record | None[source]#

Return the scope’s record with this object id, or None.

A removed record is returned until it’s purged.

Runs in a read transaction of its own.

Raises:

Examples

>>> replica.find_record("work", "a").state
<RecordState.SYNCED: 'synced'>
>>> replica.find_record("work", "z") is None
True
known_versions(scope_id: str) → dict[str, str][source]#

Return the version of the server’s copy of every object the store knows one for.

For a record in conflict that’s the server’s side. An adapter compares these with the server’s during a full listing, and reports objects whose version matches as Unchanged instead of fetching them. Removed records aren’t included.

Runs in a read transaction of its own.

Raises:

Examples

>>> replica.known_versions("work")
{'a': 'v1'}
purge(through: int) → int[source]#

Purge removed records whose last feed entry is at or below through.

Their feed entries stay until trimmed.

Runs in a write transaction of its own.

Parameters:

through – The number of a change feed entry, such as the oldest cursor your application still reads from, so every part of it has seen these removals.

Returns:

How many records were purged.

Raises:

  • TransactionError – If a store transaction is already open in this thread.

  • ValueError – If through is negative or past the last entry.

Examples

>>> from calendaring_sync import Removed
>>> with replica.session(
...     "work", fingerprint_scheme="sha256", media_type="text/plain"
... ) as session:
...     session.ingest(ChangeSet("work", "t1", "t2", {"a": Removed()}))
>>> replica.purge(replica.feed_head())
1
>>> replica.records("work")
[]
read_feed(after: int = 0, limit: int = 1000) → list[FeedEntry][source]#

Return the change feed entries after a cursor, oldest first.

Save the new cursor only after handling the entries, so a crash in between reads them again instead of skipping them.

Runs in a read transaction of its own.

Parameters:

  • after – The number of the last entry already handled.

  • limit – The most entries to return, at most FEED_PAGE_SIZE.

Raises:

  • TransactionError – If a store transaction is already open in this thread.

  • TrimmedFeedError – If entries after after were trimmed.

  • ValueError – If after is negative or past the last entry, or limit is outside 1 to FEED_PAGE_SIZE.

Examples

>>> [(entry.number, entry.kind) for entry in replica.read_feed(after=0)]
[(1, <FeedEntryKind.CHANGED: 'changed'>)]
>>> replica.read_feed(after=replica.feed_head())
[]
record(local_key: int) → Record | None[source]#

Return the record with this local key, or None.

Runs in a read transaction of its own.

Raises:

TransactionError – If a store transaction is already open in this thread.

Examples

>>> key = replica.find_record("work", "a").local_key
>>> replica.record(key).object_id
'a'
records(scope_id: str) → list[Record][source]#

Return every record in a scope, removed ones included, ordered by local key.

Runs in a read transaction of its own.

Raises:

Examples

>>> [record.object_id for record in replica.records("work")]
['a']
reset(scope_id: str) → None[source]#

Reset a scope, so its next sync is a full listing.

Records keep their state.

Runs in a write transaction of its own.

Raises:

Examples

>>> replica.reset("work")
>>> replica.scope("work").token is None
True
resolve(record: Record, resolution: TakeServer | KeepEdit | Merge | Delete | UpdateHolder) → None[source]#

Resolve one conflict, as your application decided from record.

Anything but the server’s side becomes a 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 change feed. Runs in a write transaction of its own.

Parameters:

  • record – The record in conflict, as your application read it.

  • resolution – What to do with it.

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.

  • UnknownHolderError – For UpdateHolder, if the server named no record that’s still there, and no single record holds the refused create’s logical key, or it was sent without one.

  • OpenConflictError – For UpdateHolder, if the holder is in conflict.

  • PushInFlightError – For UpdateHolder, if the holder’s push is in flight.

  • PendingEditError – For UpdateHolder, if the holder holds a pending edit of its own, which this would replace.

  • ValueError – If record isn’t in conflict, or the resolution doesn’t fit its kind: KeepEdit for a create the server refused for its logical key, or UpdateHolder for any other.

Examples

>>> from calendaring_sync import KeepEdit
>>> key = replica.find_record("work", "a").local_key
>>> replica.change(key, b"Team meeting, room 2", fingerprint="fp-mine")
>>> theirs = {"a": Changed("v2", "fp-theirs", b"Team meeting, room 3")}
>>> with replica.session(
...     "work", fingerprint_scheme="sha256", media_type="text/plain"
... ) as session:
...     session.ingest(ChangeSet("work", "t1", "t2", theirs))
>>> record = replica.record(key)
>>> record.conflict.kind
<ConflictKind.BOTH_CHANGED: 'both_changed'>
>>> replica.resolve(record, KeepEdit())
>>> replica.record(key).state, replica.record(key).version
(<RecordState.PENDING: 'pending'>, 'v2')
resolve_conflicts(resolver: Callable[[Record], TakeServer | KeepEdit | Merge | Delete | UpdateHolder | None]) → int[source]#

Pass every open conflict to resolver, and resolve each as it answers.

The conflicts are read in one read transaction, then resolver runs outside any transaction, so it can take its time, and each answer is resolved with resolve() in a write transaction of its own. None leaves a record open. A record that changed after it was read is skipped, and comes back on the next call. An exception from resolver or resolve() stops the loop and keeps every resolution before it.

Parameters:

resolver – Your function. It gets each record in conflict, from every scope, and returns a resolution or None.

Returns:

How many conflicts were resolved.

Raises:

TransactionError – If a store transaction is already open in this thread.

Examples

>>> from calendaring_sync import TakeServer
>>> key = replica.find_record("work", "a").local_key
>>> replica.change(key, b"Team meeting, room 2", fingerprint="fp-mine")
>>> theirs = {"a": Changed("v2", "fp-theirs", b"Team meeting, room 3")}
>>> with replica.session(
...     "work", fingerprint_scheme="sha256", media_type="text/plain"
... ) as session:
...     session.ingest(ChangeSet("work", "t1", "t2", theirs))
>>> replica.resolve_conflicts(lambda record: TakeServer())
1
>>> replica.record(key).content
b'Team meeting, room 3'
scope(scope_id: str) → Scope[source]#

Return one scope.

Runs in a read transaction of its own.

Raises:

Examples

>>> replica.scope("work").token
't1'
scopes() → list[Scope][source]#

Return every scope, ordered by id.

Runs in a read transaction of its own.

Raises:

TransactionError – If a store transaction is already open in this thread.

Examples

>>> [scope.id for scope in replica.scopes()]
['work']
session(scope_id: str, *, fingerprint_scheme: str, media_type: str, guarded_by_version: bool = False, unique_logical_key: bool = False) → Iterator[Session][source]#

Open a session for one scope, for the length of the with block.

The first session creates the scope. A different fingerprint scheme from the stored one resets the scope and clears every record’s version and fingerprint, so the next full listing fetches every object again. Objects whose content is the same as before aren’t reported as changed.

Opening the session and each Session.ingest() run in a write transaction of their own.

Parameters:

  • scope_id – The scope’s id, chosen by the adapter.

  • fingerprint_scheme – The name of the adapter’s fingerprint scheme.

  • media_type – The media type of the scope’s content.

  • guarded_by_version – Whether the adapter’s pushes only apply if the object still has the version they were based on. A change or delete then waits for an ingest whenever its record has no version, so it never goes out unguarded.

  • unique_logical_key – Whether no two objects in the scope can share a logical key. A create whose answer never arrived is then recognized at the next ingest by its key and the data it was sent with, which a create the server picks the id for needs.

Raises:

Examples

>>> changes = {"a": Changed("v2", "fp-a2", b"Team meeting, moved")}
>>> with replica.session(
...     "work", fingerprint_scheme="sha256", media_type="text/plain"
... ) as session:
...     session.ingest(ChangeSet("work", "t1", "t2", changes))
>>> replica.find_record("work", "a").content
b'Team meeting, moved'
trim_feed(through: int) → None[source]#

Delete every change feed entry numbered at or below through.

Runs in a write transaction of its own.

Raises:

  • TransactionError – If a store transaction is already open in this thread.

  • ValueError – If through is negative or past the last entry.

Examples

>>> replica.trim_feed(replica.feed_head())
>>> replica.read_feed(after=replica.feed_head())
[]
class calendaring_sync.replica.Session(replica: Replica, scope_id: str, *, guarded_by_version: bool, unique_logical_key: bool)[source]#

Bases: object

One sync of one scope, opened with Replica.session().

A session ingests every page first, then the adapter pushes your pending edits: for each record to_push() lists, it claims it, sends it, and settles it with how the push ended.

Examples

The adapter pushes an edit after ingesting what changed on the server, here nothing:

>>> from calendaring_sync import Applied
>>> key = replica.find_record("work", "a").local_key
>>> replica.change(key, b"Team meeting, moved", fingerprint="fp-a2")
>>> with replica.session(
...     "work", fingerprint_scheme="sha256", media_type="text/plain"
... ) as session:
...     session.ingest(ChangeSet("work", "t1", "t2", {}))
...     for record in session.to_push():
...         claimed = session.claim(record.local_key)
...         # The adapter sends claimed.edit to the server here.
...         session.settle(claimed, Applied())
True
>>> replica.record(key).state, replica.record(key).content
(<RecordState.SYNCED: 'synced'>, b'Team meeting, moved')
claim(local_key: int, *, object_id: str | None = None) → Record | None[source]#

Mark a pending edit in flight, so no other process pushes it.

Send the record this returns, then report how the push ended with settle(). Runs in a write transaction of its own; when two processes claim one record, only one gets it.

Parameters:

  • local_key – The record to push.

  • object_id – For a create the server takes ids for, the id it’s about to be sent under. It’s stored with the claim, so the create is recognized by it even if the answer never arrives. An id a removed record holds passes to this one.

Returns:

The claimed record, or None when it no longer holds a pending edit to push: another process claimed it, the edit was discarded or resolved, or it waits for a version.

Raises:

  • TransactionError – If a store transaction is already open in this thread.

  • SessionError – If the session hasn’t ingested its last page, or has ended.

  • DuplicateObjectError – If a record that isn’t removed holds object_id.

  • ValueError – If the record belongs to another scope, object_id is given for a record that has one, or a string holds a lone surrogate.

Examples

A second claim of the same record gets nothing:

>>> key = replica.find_record("work", "a").local_key
>>> replica.change(key, b"Team meeting, moved", fingerprint="fp-a2")
>>> with replica.session(
...     "work", fingerprint_scheme="sha256", media_type="text/plain"
... ) as session:
...     session.ingest(ChangeSet("work", "t1", "t2", {}))
...     claimed = session.claim(key)
...     claimed.state, session.claim(key)
(<RecordState.IN_FLIGHT: 'in_flight'>, None)
ingest(change_set: ChangeSet) → None[source]#

Ingest one change set, or one page of it, in one transaction.

The core applies each object in it: a new object gets a record, a changed one gets the new data, and a removed one is marked removed. An object whose fingerprint, containers and logical key didn’t change only gets its version and the server’s content, and isn’t reported. A truncated page of an incremental change set moves the token to its checkpoint. A full listing moves the token only on its last page, which also removes every record the listing never mentioned. A complete change set, or the last page, also settles every push still in flight from what the server reported, and one it doesn’t mention is pending again. If anything fails, nothing changes.

Raises:

  • TransactionError – If a store transaction is already open in this thread.

  • StaleChangeSetError – If the change set was fetched from a token the scope no longer has, or is a later page of a listing with another token, such as one another session has replaced. The session goes back to idle, so the adapter can fetch again and ingest from the start in the same session.

  • SessionError – If the session has ended, already ingested its last page, has started pushing, or the page is a listing page in an incremental session or the other way round.

  • ValueError – If the change set is for another scope, an incremental change set arrives for a scope with no token, or an Unchanged entry is outside a listing or names a record with no version.

property scope_id: str#

The id of the session’s scope.

settle(claimed: Record, outcome: Applied | Rejected | Gone | LogicalKeyTaken | Unknown) → bool[source]#

Record how the push of a claimed record ended.

Runs in a write transaction of its own, and writes a change feed entry for every outcome but Unknown. If another process already settled this push, for example by ingesting while it was in flight, the outcome only counts while the record still holds the same edit and no one has claimed it again.

Parameters:

  • claimed – The record claim() returned.

  • outcome – How the push ended.

Returns:

True if the store now agrees with the outcome, False if the push had already been settled another way and nothing was written.

Raises:

  • TransactionError – If a store transaction is already open in this thread.

  • SessionError – If the session has ended.

  • DuplicateObjectError – If Applied names an id a record that isn’t removed holds.

  • ValueError – If claimed isn’t a record this session’s claim() returned, or the outcome is one no push of its edit can have: Gone for a create, LogicalKeyTaken for anything but a create, Applied with an id for a delete, or Applied without one for a create that has none.

Examples

The server refused the write, so the edit is pending again:

>>> from calendaring_sync import Rejected
>>> key = replica.find_record("work", "a").local_key
>>> replica.change(key, b"Team meeting, moved", fingerprint="fp-a2")
>>> with replica.session(
...     "work", fingerprint_scheme="sha256", media_type="text/plain"
... ) as session:
...     session.ingest(ChangeSet("work", "t1", "t2", {}))
...     session.settle(session.claim(key), Rejected())
True
>>> replica.record(key).state
<RecordState.PENDING: 'pending'>
property state: SessionState#

Where the session is.

to_push() → list[Record][source]#

Return the pending edits ready to push, by local key.

With guarded_by_version, a change or delete of a record with no version waits for an ingest that gives it one, and isn’t listed. Runs in a read transaction of its own.

Raises:

  • TransactionError – If a store transaction is already open in this thread.

  • SessionError – If the session hasn’t ingested its last page, or has ended.

Examples

>>> key = replica.find_record("work", "a").local_key
>>> replica.change(key, b"Team meeting, moved", fingerprint="fp-a2")
>>> with replica.session(
...     "work", fingerprint_scheme="sha256", media_type="text/plain"
... ) as session:
...     session.ingest(ChangeSet("work", "t1", "t2", {}))
...     [record.local_key for record in session.to_push()] == [key]
True
class calendaring_sync.replica.SessionState(*values)[source]#

Bases: StrEnum

Where a session is.

IDLE = 'idle'#

Nothing ingested yet.

INGESTED = 'ingested'#

The last page was ingested, and the token is final.

INGESTING = 'ingesting'#

A truncated page was ingested, and more pages follow.

PUSHING = 'pushing'#

A push was claimed; no more pages can be ingested.

Last updated on 2026-10-20.