gabriel / muse public
rga.py python
327 lines 12.5 KB
Raw
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 28 days ago
1 """Replicated Growable Array (RGA) — CRDT for ordered sequences.
2
3 The RGA (Roh et al., 2011 "Replicated abstract data types") provides
4 Google-Docs-style collaborative editing semantics for any ordered sequence
5 domain. Every element carries a globally unique, immutable identifier
6 ``f"{timestamp}@{author}"``; this identifier determines insertion order when
7 two agents concurrently insert at the same position.
8
9 **Core invariant**: the visible sequence is the list of elements whose
10 ``deleted`` flag is ``False``, in the order determined by their identifiers.
11 Deletions are *tombstoned* (``deleted=True``) rather than physically removed
12 so that identifiers remain stable across replicas.
13
14 **Insertion semantics**: ``insert(after_id, element)`` inserts *element*
15 immediately after the element with ``id == after_id`` (``None`` means
16 prepend). Concurrent inserts at the same position are resolved by sorting
17 the new element's ID lexicographically (descending) — the "bigger" ID wins
18 and is placed first, giving a deterministic outcome independent of delivery
19 order.
20
21 **Lattice laws satisfied by** :meth:`join`:
22 1. Commutativity: ``join(a, b) == join(b, a)``
23 2. Associativity: ``join(join(a, b), c) == join(a, join(b, c))``
24 3. Idempotency: ``join(a, a) == a``
25
26 Public API
27 ----------
28 - :class:`RGAElement` — ``TypedDict`` for one array element.
29 - :class:`RGA` — the array itself.
30 """
31
32
33 from __future__ import annotations
34
35 import logging
36 from typing import TypedDict
37
38
39 type _ElemMap = dict[str, "RGAElement"]
40 logger = logging.getLogger(__name__)
41
42
43 class RGAElement(TypedDict):
44 """A single element in an :class:`RGA`.
45
46 ``id`` is the stable unique identifier ``"{timestamp}@{author}"`` assigned
47 at insertion time. ``value`` is the content hash of the element (it
48 references the object store — all binary content lives there).
49 ``deleted`` is ``True`` for tombstoned elements that no longer appear in
50 the visible sequence.
51
52 ``parent_id`` is the ``id`` of the element this one was inserted after
53 (``None`` means it was prepended — inserted at the head). This is
54 required for the commutative join algorithm to correctly place concurrent
55 inserts regardless of which replica initiates the join.
56 """
57
58 id: str
59 value: str
60 deleted: bool
61 parent_id: str | None
62
63
64 class RGA:
65 """Replicated Growable Array — CRDT for ordered sequences.
66
67 Provides ``insert``, ``delete``, ``join``, and ``to_sequence`` operations.
68 All mutating methods return new :class:`RGA` instances; ``self`` is
69 never modified.
70
71 The internal representation is a list of :class:`RGAElement` dicts in
72 insertion order (not visible order — tombstones are kept inline).
73
74 Example::
75
76 rga = RGA()
77 rga, id_a = rga.insert(None, "note-hash-A") # prepend
78 rga, id_b = rga.insert(id_a, "note-hash-B") # insert after A
79 rga = rga.delete(id_a) # tombstone A
80 assert rga.to_sequence() == ["note-hash-B"]
81 """
82
83 def __init__(self, elements: list[RGAElement] | None = None) -> None:
84 """Construct an RGA, optionally pre-populated.
85
86 Args:
87 elements: Ordered list of :class:`RGAElement` dicts (may contain
88 tombstones). Copied defensively.
89 """
90 self._elements: list[RGAElement] = list(elements) if elements else []
91
92 # ------------------------------------------------------------------
93 # Mutations (return new RGA)
94 # ------------------------------------------------------------------
95
96 def insert(self, after_id: str | None, value: str, *, element_id: str) -> RGA:
97 """Return a new RGA with *value* inserted after *after_id*.
98
99 Concurrent inserts at the same position are resolved by placing the
100 element with the lexicographically *larger* ``element_id`` first.
101
102 Args:
103 after_id: The ``id`` of the element to insert after, or ``None``
104 to prepend (insert before all existing elements).
105 value: The content hash of the new element.
106 element_id: The stable unique ID for the new element; callers
107 should use ``f"{timestamp}@{author}"`` to ensure global
108 uniqueness across agents.
109
110 Returns:
111 A new :class:`RGA` with the element inserted at the correct position.
112 """
113 new_elem: RGAElement = {
114 "id": element_id,
115 "value": value,
116 "deleted": False,
117 "parent_id": after_id,
118 }
119 elems = list(self._elements)
120
121 if after_id is None:
122 # Prepend: among concurrent prepends (same parent_id=None), larger ID goes first.
123 insert_pos = 0
124 while (
125 insert_pos < len(elems)
126 and elems[insert_pos]["parent_id"] is None
127 and elems[insert_pos]["id"] > element_id
128 ):
129 insert_pos += 1
130 elems.insert(insert_pos, new_elem)
131 else:
132 # Find the anchor element.
133 anchor_idx = next(
134 (i for i, e in enumerate(elems) if e["id"] == after_id), None
135 )
136 if anchor_idx is None:
137 # Unknown anchor — append at end (safe degradation).
138 logger.warning("RGA.insert: unknown after_id=%r, appending at end", after_id)
139 elems.append(new_elem)
140 else:
141 # Insert after anchor. Skip any existing elements that also
142 # have the same parent_id AND a larger element ID (concurrent
143 # inserts at the same position; larger ID wins leftmost slot).
144 insert_pos = anchor_idx + 1
145 while (
146 insert_pos < len(elems)
147 and elems[insert_pos]["parent_id"] == after_id
148 and elems[insert_pos]["id"] > element_id
149 ):
150 insert_pos += 1
151 elems.insert(insert_pos, new_elem)
152
153 return RGA(elems)
154
155 def delete(self, element_id: str) -> RGA:
156 """Return a new RGA with *element_id* tombstoned.
157
158 Tombstoning is idempotent — deleting an already-deleted or unknown
159 element is a no-op.
160
161 Args:
162 element_id: The ``id`` of the element to tombstone.
163
164 Returns:
165 A new :class:`RGA` with the element marked ``deleted=True``.
166 """
167 new_elems: list[RGAElement] = []
168 for elem in self._elements:
169 if elem["id"] == element_id:
170 new_elems.append({
171 "id": elem["id"],
172 "value": elem["value"],
173 "deleted": True,
174 "parent_id": elem["parent_id"],
175 })
176 else:
177 new_elems.append({
178 "id": elem["id"],
179 "value": elem["value"],
180 "deleted": elem["deleted"],
181 "parent_id": elem["parent_id"],
182 })
183 return RGA(new_elems)
184
185 # ------------------------------------------------------------------
186 # CRDT join
187 # ------------------------------------------------------------------
188
189 def join(self, other: RGA) -> RGA:
190 """Return the lattice join — the union of both arrays.
191
192 Elements are keyed by ``id``. The join:
193 1. Takes the union of all element IDs from both replicas.
194 2. For each ID, marks the element ``deleted`` if *either* replica has
195 it tombstoned (once deleted, always deleted — monotone).
196 3. Preserves the insertion-order sequence from ``self``; appends any
197 elements from ``other`` not yet seen in ``self``.
198
199 Args:
200 other: The RGA to merge with.
201
202 Returns:
203 A new :class:`RGA` that is the join of ``self`` and *other*.
204 """
205 # Build ID → element maps from both replicas.
206 self_map: _ElemMap = {e["id"]: e for e in self._elements}
207 other_map: _ElemMap = {e["id"]: e for e in other._elements}
208
209 # Merge deletions monotonically: once deleted in either, always deleted.
210 merged_map: _ElemMap = {}
211 all_ids = set(self_map) | set(other_map)
212 for eid in all_ids:
213 if eid in self_map and eid in other_map:
214 s = self_map[eid]
215 o = other_map[eid]
216 # In practice the same element_id always carries the same value
217 # (because element_id = "{timestamp}@{author}" uniquely identifies
218 # a write). If values differ (only possible in crafted test scenarios),
219 # pick the lexicographically larger value for commutativity.
220 winning_value = s["value"] if s["value"] >= o["value"] else o["value"]
221 merged_map[eid] = {
222 "id": eid,
223 "value": winning_value,
224 "deleted": s["deleted"] or o["deleted"],
225 "parent_id": s["parent_id"],
226 }
227 elif eid in self_map:
228 src = self_map[eid]
229 merged_map[eid] = {
230 "id": src["id"],
231 "value": src["value"],
232 "deleted": src["deleted"],
233 "parent_id": src["parent_id"],
234 }
235 else:
236 src = other_map[eid]
237 merged_map[eid] = {
238 "id": src["id"],
239 "value": src["value"],
240 "deleted": src["deleted"],
241 "parent_id": src["parent_id"],
242 }
243
244 # Rebuild a canonical ordered sequence using parent_id links.
245 # Group elements by parent_id. Within each group, sort by ID
246 # descending (larger ID → leftmost, per concurrent-insert tiebreak rule).
247 # Traverse recursively: start with children of None (prepended), then
248 # recurse on each child's children.
249 from collections import defaultdict
250 children: dict[str | None, list[str]] = defaultdict(list)
251 for eid, elem in merged_map.items():
252 children[elem["parent_id"]].append(eid)
253 for group in children.values():
254 group.sort(reverse=True) # larger ID first
255
256 ordered: list[RGAElement] = []
257
258 def _traverse(parent: str | None) -> None:
259 for eid in children.get(parent, []):
260 ordered.append(merged_map[eid])
261 _traverse(eid)
262
263 _traverse(None)
264 return RGA(ordered)
265
266 # ------------------------------------------------------------------
267 # Query
268 # ------------------------------------------------------------------
269
270 def to_sequence(self) -> list[str]:
271 """Return the visible element values (excluding tombstones).
272
273 Returns:
274 List of ``value`` strings in document order, tombstones excluded.
275 """
276 return [e["value"] for e in self._elements if not e["deleted"]]
277
278 def __len__(self) -> int:
279 return len([e for e in self._elements if not e["deleted"]])
280
281 # ------------------------------------------------------------------
282 # Serialisation
283 # ------------------------------------------------------------------
284
285 def to_dict(self) -> list[RGAElement]:
286 """Return a JSON-serialisable list of :class:`RGAElement` dicts.
287
288 Returns:
289 Ordered list of all elements (including tombstones).
290 """
291 return [
292 {"id": e["id"], "value": e["value"], "deleted": e["deleted"], "parent_id": e["parent_id"]}
293 for e in self._elements
294 ]
295
296 @classmethod
297 def from_dict(cls, data: list[RGAElement]) -> RGA:
298 """Reconstruct an :class:`RGA` from its wire representation.
299
300 Args:
301 data: List of :class:`RGAElement` dicts as produced by
302 :meth:`to_dict`.
303
304 Returns:
305 A new :class:`RGA`.
306 """
307 return cls(list(data))
308
309 # ------------------------------------------------------------------
310 # Python dunder helpers
311 # ------------------------------------------------------------------
312
313 def equivalent(self, other: RGA) -> bool:
314 """Return ``True`` if both RGAs have identical element lists (including tombstones).
315
316 Note: use :meth:`to_sequence` comparison when only visible content matters.
317
318 Args:
319 other: The RGA to compare against.
320
321 Returns:
322 ``True`` when the full internal element lists are equal.
323 """
324 return self._elements == other._elements
325
326 def __repr__(self) -> str:
327 return f"RGA(len={len(self)}, elements={self._elements!r})"
File History 4 commits
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 28 days ago
sha256:50d413a635f57761c3fce1f70251b957b6423821525bf330747ef4007339222e Merge branch 'dev' into main Human 31 days ago
sha256:fb67fed5a4d3e40de84bdd163de94ef1386570bef1dd1a020a732c8a038962ce Merge branch 'dev' into main Human 50 days ago
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 102 days ago