aaronrene / gabriel-muse public
or_set.py python
246 lines 8.9 KB
Raw
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 13 days ago
1 """Observed-Remove Set (OR-Set) — add-wins CRDT for unordered collections.
2
3 An OR-Set is an unordered set where *adds always win over concurrent removes*.
4 Each element is tagged with a unique token set when it is added. Removing an
5 element requires specifying the observed tokens; a concurrent add with a new
6 token survives the remove.
7
8 This gives the "add-wins" property: if agent A adds element X and agent B
9 concurrently removes X, the merged result still contains X. The rationale
10 for Muse: an annotation or a note added by one agent should not be silently
11 deleted by a concurrent tombstone from another.
12
13 **Algorithm (Shapiro et al., 2011 "A Comprehensive Study of CRDTs"):**
14
15 - Each element ``e`` maps to a set of *unique tokens* ``{t₁, t₂, …}``.
16 - ``add(e)`` generates a fresh token and inserts ``(e, token)`` into the
17 payload.
18 - ``remove(e)`` removes every ``(e, token)`` pair currently observed.
19 A concurrent ``add(e)`` with a new token survives because its token was not
20 yet in the "observed" set at remove time.
21 - ``join(a, b)`` is set-union on the ``(element, token)`` pairs, then removes
22 any pair whose token appears in either replica's *tombstone* set (for
23 optimisation — see :attr:`_tombstones`).
24
25 We simplify slightly: tombstones are not compacted in this implementation
26 (correct but not space-optimal). Compaction is a GC concern, not a
27 correctness concern.
28
29 **Lattice laws satisfied by** :meth:`join`:
30 1. Commutativity: ``join(a, b) == join(b, a)``
31 2. Associativity: ``join(join(a, b), c) == join(a, join(b, c))``
32 3. Idempotency: ``join(a, a) == a``
33
34 Public API
35 ----------
36 - :class:`ORSetEntry` — ``TypedDict`` for a single (element, token) pair.
37 - :class:`ORSetDict` — ``TypedDict`` wire format for a complete OR-Set.
38 - :class:`ORSet` — the set itself.
39 """
40
41
42 from __future__ import annotations
43
44 import logging
45 import uuid
46 from typing import TypedDict
47
48 logger = logging.getLogger(__name__)
49
50
51 class ORSetEntry(TypedDict):
52 """A single (element, token) pair in the OR-Set payload.
53
54 ``element`` is the string value being tracked (e.g. a content hash or a
55 label). ``token`` is the unique identifier created when this particular
56 *addition* of ``element`` occurred; it distinguishes concurrent adds of
57 the same element by different agents.
58 """
59
60 element: str
61 token: str
62
63
64 class ORSetDict(TypedDict):
65 """Wire format for a complete :class:`ORSet`.
66
67 ``entries`` holds all live ``(element, token)`` pairs.
68 ``tombstones`` holds all token strings that have been explicitly removed.
69 An entry whose token appears in ``tombstones`` is considered deleted.
70 """
71
72 entries: list[ORSetEntry]
73 tombstones: list[str]
74
75
76 class ORSet:
77 """Observed-Remove Set — an unordered add-wins CRDT set.
78
79 Elements are arbitrary strings (content hashes, labels, identifiers).
80 The set supports concurrent add and remove from multiple agents with the
81 guarantee that adds always win over concurrent removes.
82
83 All mutating methods return new :class:`ORSet` instances; ``self`` is
84 never modified.
85
86 Example::
87
88 s1 = ORSet()
89 s1, tok = s1.add("note-A")
90
91 s2 = ORSet()
92 s2 = s2.remove("note-A", {tok}) # remove the observed token
93
94 # Concurrent add by s1 with a NEW token survives the remove:
95 s1_v2, tok2 = s1.add("note-A") # new token
96
97 merged = s1_v2.join(s2)
98 assert "note-A" in merged.elements() # add-wins
99 """
100
101 def __init__(
102 self,
103 entries: set[tuple[str, str]] | None = None,
104 tombstones: set[str] | None = None,
105 ) -> None:
106 """Initialise an OR-Set, optionally pre-populated.
107
108 Args:
109 entries: Set of ``(element, token)`` pairs (alive entries).
110 tombstones: Set of removed tokens.
111 """
112 self._entries: set[tuple[str, str]] = set(entries) if entries else set()
113 self._tombstones: set[str] = set(tombstones) if tombstones else set()
114
115 # ------------------------------------------------------------------
116 # Mutations (return new ORSet)
117 # ------------------------------------------------------------------
118
119 def add(self, element: str) -> tuple[ORSet, str]:
120 """Add *element* to the set with a fresh unique token.
121
122 Args:
123 element: The string value to add.
124
125 Returns:
126 A ``(new_set, token)`` pair where ``new_set`` contains the
127 added element and ``token`` is the unique identifier of this
128 particular addition (useful for targeted removal later).
129 """
130 token = str(uuid.uuid4())
131 new_entries = self._entries | {(element, token)}
132 return ORSet(new_entries, self._tombstones), token
133
134 def remove(self, element: str, observed_tokens: set[str]) -> ORSet:
135 """Remove *element* by tombstoning all currently observed tokens.
136
137 Only the tokens listed in *observed_tokens* are tombstoned. Any token
138 added *after* this remove (i.e. from a concurrent ``add``) is not
139 affected and the element will survive in the merged result.
140
141 Args:
142 element: The element to remove.
143 observed_tokens: The set of tokens for *element* that were observed
144 at remove time (typically ``self.tokens_for(element)``).
145
146 Returns:
147 A new :class:`ORSet` with *element*'s observed tokens tombstoned.
148 """
149 relevant = {(e, t) for e, t in self._entries if e == element and t in observed_tokens}
150 new_tombstones = self._tombstones | {t for _, t in relevant}
151 new_entries = self._entries - relevant
152 return ORSet(new_entries, new_tombstones)
153
154 # ------------------------------------------------------------------
155 # CRDT join
156 # ------------------------------------------------------------------
157
158 def join(self, other: ORSet) -> ORSet:
159 """Return the lattice join — union of entries minus all tombstones.
160
161 Args:
162 other: The OR-Set to merge with.
163
164 Returns:
165 A new :class:`ORSet` containing all entries from either replica
166 whose tokens have not been tombstoned by either replica.
167 """
168 all_tombstones = self._tombstones | other._tombstones
169 all_entries = (self._entries | other._entries) - {
170 (e, t) for e, t in self._entries | other._entries if t in all_tombstones
171 }
172 return ORSet(all_entries, all_tombstones)
173
174 # ------------------------------------------------------------------
175 # Query
176 # ------------------------------------------------------------------
177
178 def elements(self) -> frozenset[str]:
179 """Return the set of visible elements (those not tombstoned).
180
181 Returns:
182 Frozenset of string elements currently in the set.
183 """
184 return frozenset(e for e, t in self._entries if t not in self._tombstones)
185
186 def tokens_for(self, element: str) -> set[str]:
187 """Return all live tokens for *element*.
188
189 Pass the result to :meth:`remove` to remove *element* without
190 accidentally tombstoning tokens added concurrently.
191
192 Args:
193 element: The element to look up.
194
195 Returns:
196 Set of token strings associated with live copies of *element*.
197 """
198 return {t for e, t in self._entries if e == element and t not in self._tombstones}
199
200 def __contains__(self, element: str) -> bool:
201 return element in self.elements()
202
203 # ------------------------------------------------------------------
204 # Serialisation
205 # ------------------------------------------------------------------
206
207 def to_dict(self) -> ORSetDict:
208 """Return a JSON-serialisable :class:`ORSetDict`.
209
210 Returns:
211 Dict with ``"entries"`` and ``"tombstones"`` lists.
212 """
213 entries: list[ORSetEntry] = [{"element": e, "token": t} for e, t in sorted(self._entries)]
214 return {"entries": entries, "tombstones": sorted(self._tombstones)}
215
216 @classmethod
217 def from_dict(cls, data: ORSetDict) -> ORSet:
218 """Reconstruct an :class:`ORSet` from its wire representation.
219
220 Args:
221 data: Dict as produced by :meth:`to_dict`.
222
223 Returns:
224 A new :class:`ORSet`.
225 """
226 entries = {(entry["element"], entry["token"]) for entry in data["entries"]}
227 tombstones = set(data["tombstones"])
228 return cls(entries, tombstones)
229
230 # ------------------------------------------------------------------
231 # Python dunder helpers
232 # ------------------------------------------------------------------
233
234 def equivalent(self, other: ORSet) -> bool:
235 """Return ``True`` if both OR-Sets have the same visible elements and tombstones.
236
237 Args:
238 other: The OR-Set to compare against.
239
240 Returns:
241 ``True`` when visible elements and tombstone sets are identical.
242 """
243 return self.elements() == other.elements() and self._tombstones == other._tombstones
244
245 def __repr__(self) -> str:
246 return f"ORSet(elements={self.elements()!r})"
File History 1 commit
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 13 days ago