CRDT：Notion・Figma・Linear のリアルタイム同時編集を支える分散データ構造

class GCounter:    '''各ノードのカウンタを別々に保持し、合計を value() とする'''    def __init__(self, node_id):        self.node = node_id        self.counts = {}  # {node_id: count}
    def inc(self, n=1):        self.counts[self.node] = self.counts.get(self.node, 0) + n
    def value(self):        return sum(self.counts.values())
    def merge(self, other):        '''各ノードの最大値を取る (べき等・可換・結合)'''        for k, v in other.counts.items():            self.counts[k] = max(self.counts.get(k, 0), v)
# 3 ノードでバラバラに +1, +2, +3a = GCounter('A'); a.inc(3)b = GCounter('B'); b.inc(2)c = GCounter('C'); c.inc(1)
# 任意の順序でマージa.merge(b); a.merge(c)b.merge(a)c.merge(a)
print(f'A={a.value()}, B={b.value()}, C={c.value()}')# 全て 6 (= 3+2+1)

import time
class LWWSet:    '''Last-Write-Wins Element Set'''    def __init__(self):        self.adds = {}     # {element: timestamp}        self.removes = {}  # {element: timestamp}
    def add(self, e, ts=None):        ts = ts or time.time()        self.adds[e] = max(self.adds.get(e, 0), ts)
    def remove(self, e, ts=None):        ts = ts or time.time()        self.removes[e] = max(self.removes.get(e, 0), ts)
    def __contains__(self, e):        a = self.adds.get(e, 0)        r = self.removes.get(e, 0)        return a > r  # add が新しければ含まれる
    def merge(self, other):        for e, t in other.adds.items():            self.adds[e] = max(self.adds.get(e, 0), t)        for e, t in other.removes.items():            self.removes[e] = max(self.removes.get(e, 0), t)
# シナリオ: 2 人が同じ要素を異なるタイミングで add/removes1 = LWWSet(); s2 = LWWSet()s1.add('apple', ts=100)s2.add('apple', ts=100)s2.remove('apple', ts=200)  # s2 だけ削除
s1.merge(s2)print(f'apple in s1: {"apple" in s1}')  # False (削除が新しい)

# pip install y-py# ※ 本番では JS の Yjs を WebSocket で接続するのが一般的
import y_py as Y
doc = Y.YDoc()text = doc.get_text('content')with doc.begin_transaction() as txn:    text.insert(txn, 0, 'Hello, ')    text.insert(txn, 7, 'World!')
print(str(text))  # 'Hello, World!'# Yjs の update バイナリは差分同期に使える (WebRTC / WebSocket で配信)update = Y.encode_state_as_update(doc)