124 lines
4.3 KiB
Python
124 lines
4.3 KiB
Python
"""Small op-based CRDT and PlanDelta model for deterministic delta-sync.
|
|
|
|
This is intentionally compact: op-based PlanDeltas carry a list of ops and a
|
|
version vector. The DeltaStore applies and merges deltas deterministically
|
|
using version vectors and a last-writer-wins tie-breaker (timestamp + author).
|
|
"""
|
|
from dataclasses import dataclass, field, asdict
|
|
from typing import List, Dict, Any, Tuple
|
|
import time
|
|
import json
|
|
|
|
|
|
@dataclass
|
|
class VersionVector:
|
|
vv: Dict[str, int] = field(default_factory=dict)
|
|
|
|
def bump(self, node: str) -> None:
|
|
self.vv[node] = self.vv.get(node, 0) + 1
|
|
|
|
def update(self, other: "VersionVector") -> None:
|
|
for k, v in other.vv.items():
|
|
self.vv[k] = max(self.vv.get(k, 0), v)
|
|
|
|
def dominates(self, other: "VersionVector") -> bool:
|
|
# True if self >= other componentwise and at least one >
|
|
ge = True
|
|
strictly_greater = False
|
|
for k in set(self.vv.keys()).union(other.vv.keys()):
|
|
a = self.vv.get(k, 0)
|
|
b = other.vv.get(k, 0)
|
|
if a < b:
|
|
ge = False
|
|
break
|
|
if a > b:
|
|
strictly_greater = True
|
|
return ge and strictly_greater
|
|
|
|
def to_dict(self) -> Dict[str, int]:
|
|
return dict(self.vv)
|
|
|
|
@classmethod
|
|
def from_dict(cls, d: Dict[str, int]) -> "VersionVector":
|
|
return cls(dict(d))
|
|
|
|
|
|
@dataclass
|
|
class PlanDelta:
|
|
delta_id: str
|
|
author: str
|
|
contract_id: str
|
|
timestamp: float
|
|
ops: List[Dict[str, Any]]
|
|
version_vector: VersionVector
|
|
signature: str = None
|
|
|
|
def to_json(self) -> str:
|
|
payload = asdict(self)
|
|
payload["version_vector"] = self.version_vector.to_dict()
|
|
return json.dumps(payload, sort_keys=True)
|
|
|
|
@classmethod
|
|
def create(cls, delta_id: str, author: str, contract_id: str, ops: List[Dict[str, Any]], vv: VersionVector) -> "PlanDelta":
|
|
return cls(delta_id=delta_id, author=author, contract_id=contract_id, timestamp=time.time(), ops=ops, version_vector=vv)
|
|
|
|
|
|
class DeltaStore:
|
|
"""A simple in-memory store that applies PlanDeltas to a shared map.
|
|
|
|
The underlying state is a mapping of dotted paths to values. Ops are
|
|
simple: {op: 'set'|'delete', path: 'a.b.c', value: ...}.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.state: Dict[str, Any] = {}
|
|
self.applied: List[Tuple[str, float, str]] = [] # (delta_id, timestamp, author)
|
|
self.vv = VersionVector()
|
|
|
|
def apply(self, delta: PlanDelta) -> None:
|
|
# Skip applying if delta is already dominated by local vv
|
|
if self.vv.dominates(delta.version_vector):
|
|
return
|
|
|
|
# deterministic ordering of ops: sort by (timestamp, author) if present inside op, else keep provided order
|
|
for op in delta.ops:
|
|
self._apply_op(op, delta)
|
|
|
|
# update version vector
|
|
self.vv.update(delta.version_vector)
|
|
self.applied.append((delta.delta_id, delta.timestamp, delta.author))
|
|
|
|
def _apply_op(self, op: Dict[str, Any], delta: PlanDelta) -> None:
|
|
typ = op.get("op")
|
|
path = op.get("path")
|
|
if typ == "set":
|
|
self._set(path, op.get("value"), delta)
|
|
elif typ == "delete":
|
|
self._delete(path, delta)
|
|
else:
|
|
raise ValueError(f"unknown op: {typ}")
|
|
|
|
def _set(self, path: str, value: Any, delta: PlanDelta) -> None:
|
|
# LWW semantics: if existing metadata exists, compare (timestamp, author) to decide
|
|
meta_key = f"__meta__:{path}"
|
|
existing_meta = self.state.get(meta_key)
|
|
incoming_meta = (delta.timestamp, delta.author)
|
|
if existing_meta is None or incoming_meta >= existing_meta:
|
|
self.state[path] = value
|
|
self.state[meta_key] = incoming_meta
|
|
|
|
def _delete(self, path: str, delta: PlanDelta) -> None:
|
|
meta_key = f"__meta__:{path}"
|
|
existing_meta = self.state.get(meta_key)
|
|
incoming_meta = (delta.timestamp, delta.author)
|
|
if existing_meta is None or incoming_meta >= existing_meta:
|
|
if path in self.state:
|
|
del self.state[path]
|
|
self.state[meta_key] = incoming_meta
|
|
|
|
def get(self, path: str, default=None):
|
|
return self.state.get(path, default)
|
|
|
|
def merge_remote_vv(self, remote_vv: VersionVector) -> None:
|
|
self.vv.update(remote_vv)
|