From 26c19f20f6f0948335bc0adcdfb90fa0ef0138ea Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 7 Jul 2026 11:06:24 +0000 Subject: [PATCH] fix(rust_brain): preserve HLC on snapshot restore and gossip receive After v0.6.0 added HLC monotonic enforcement, restore_from_file() and gossip.receive() still assigned fresh HLC timestamps. Post-restore causal writes were rejected (TimestampRegression) causing silent data loss; gossip without HLC could overwrite fresher data. - Add _parse_hlc() for snapshot/wire normalization with legacy fallback - Restore HLC and advance global clock on snapshot restore - Pass HLC through bulk_write and gossip receive with stale rejection - Hold lock during restore_from_file and gc_expired to prevent races Co-authored-by: Daniel --- hive/gossip.py | 20 ++++++++- hive/rust_brain/__init__.py | 74 +++++++++++++++++++++------------ tests/test_enterprise_backup.py | 28 ++++++++++++- tests/test_gossip.py | 42 +++++++++++++++++++ 4 files changed, 136 insertions(+), 28 deletions(-) diff --git a/hive/gossip.py b/hive/gossip.py index f0e5372..1ab5bd3 100644 --- a/hive/gossip.py +++ b/hive/gossip.py @@ -122,11 +122,29 @@ def receive(self, events: list[dict[str, Any]]) -> int: applied = 0 for ev in events: try: + key = ev["key"] + raw_hlc = ev.get("hlc") + if raw_hlc is None: + # Without an HLC we cannot establish causal order for updates. + if self._brain.get(key) is not None: + _log.debug( + "Skipping gossip update for %r: missing hlc on existing key", + key, + ) + continue + hlc = None + else: + hlc = tuple(raw_hlc) + self._brain.update_hlc(hlc) + self._brain.remember( - ev["key"], + key, ev["value"], trust=ev.get("trust", 1.0), tags=set(ev.get("tags", [])), + ts_ns=ev.get("ts_ns"), + hlc=hlc, + edges=ev.get("edges"), ) applied += 1 except Exception as exc: diff --git a/hive/rust_brain/__init__.py b/hive/rust_brain/__init__.py index a246c1f..f589376 100644 --- a/hive/rust_brain/__init__.py +++ b/hive/rust_brain/__init__.py @@ -125,6 +125,19 @@ def _now_ns() -> int: return time.time_ns() - HIVE_EPOCH_NS +def _parse_hlc(raw: Any, *, ts_ns: int | None = None) -> tuple[int, int, str]: + """Normalise an HLC from wire/snapshot form (list or tuple). + + Falls back to a synthetic HLC from ``ts_ns`` for pre-v0.6.0 snapshots. + """ + if raw is not None: + wall, logical, node_id = raw + return (int(wall), int(logical), str(node_id)) + if ts_ns is not None: + return (ts_ns + HIVE_EPOCH_NS, 0, "legacy") + return _hlc.now() + + # --------------------------------------------------------------------------- # Edge / node model # --------------------------------------------------------------------------- @@ -389,6 +402,9 @@ def bulk_write(self, rows: Iterable[Mapping[str, Any]]) -> int: tags=row.get("tags", ()), edges=row.get("edges"), ts_ns=row.get("ts_ns"), + hlc=_parse_hlc(row.get("hlc"), ts_ns=row.get("ts_ns")) + if row.get("hlc") is not None or row.get("ts_ns") is not None + else None, ) n += 1 return n @@ -432,13 +448,14 @@ def gc_expired(self) -> int: if self._default_ttl_s is None: return 0 removed = 0 - for storage_key in list(self._nodes): - node = self._nodes[storage_key] - age_s = (_now_ns() - node.ts_ns) / 1e9 - if age_s > self._default_ttl_s: - self._nodes.pop(storage_key, None) - self._remove_from_order(storage_key) - removed += 1 + with self._lock: + for storage_key in list(self._nodes): + node = self._nodes[storage_key] + age_s = (_now_ns() - node.ts_ns) / 1e9 + if age_s > self._default_ttl_s: + self._nodes.pop(storage_key, None) + self._remove_from_order(storage_key) + removed += 1 return removed @@ -505,25 +522,30 @@ def restore_from_file(self, path: str) -> int: raise ValueError( "snapshot checksum mismatch: file is corrupt or tampered" ) - self._nodes.clear() - self._order.clear() - self._order_index.clear() - for node_dict in nodes: - node = MemoryNode( - key=node_dict["key"], - value=node_dict["value"], - ts_ns=node_dict["ts_ns"], - trust=node_dict.get("trust", 1.0), - tags=set(node_dict.get("tags", [])), - node_id=node_dict.get("id", uuid.uuid4().hex[:12]), - ) - for kind, neighbours in node_dict.get("edges", {}).items(): - for n in neighbours: - node.attach(kind, n) - storage_key = self._prefix(node.key) - self._nodes[storage_key] = node - self._order_index[storage_key] = len(self._order) - self._order.append(storage_key) + with self._lock: + self._nodes.clear() + self._order.clear() + self._order_index.clear() + for node_dict in nodes: + ts_ns = node_dict["ts_ns"] + node_hlc = _parse_hlc(node_dict.get("hlc"), ts_ns=ts_ns) + node = MemoryNode( + key=node_dict["key"], + value=node_dict["value"], + ts_ns=ts_ns, + hlc=node_hlc, + trust=node_dict.get("trust", 1.0), + tags=set(node_dict.get("tags", [])), + node_id=node_dict.get("id", uuid.uuid4().hex[:12]), + ) + for kind, neighbours in node_dict.get("edges", {}).items(): + for n in neighbours: + node.attach(kind, n) + storage_key = self._prefix(node.key) + self._nodes[storage_key] = node + self._order_index[storage_key] = len(self._order) + self._order.append(storage_key) + _hlc.update(node_hlc) return len(nodes) def __repr__(self) -> str: diff --git a/tests/test_enterprise_backup.py b/tests/test_enterprise_backup.py index de57680..42f38cd 100644 --- a/tests/test_enterprise_backup.py +++ b/tests/test_enterprise_backup.py @@ -9,7 +9,7 @@ import pytest -from hive.rust_brain import RustBrain +from hive.rust_brain import RustBrain, TimestampRegression def test_snapshot_to_file_and_restore(): @@ -149,6 +149,32 @@ def test_version_mismatch(): os.unlink(path) +def test_restore_preserves_hlc_for_replication(): + """HLC must survive snapshot round-trip so post-restore writes stay ordered.""" + brain = RustBrain() + brain.remember("k", "v1", hlc=(5000, 10, "nodeA")) + orig_hlc = brain.get("k").hlc + + with tempfile.NamedTemporaryFile(delete=False, suffix=".gz") as f: + path = f.name + + try: + brain.snapshot_to_file(path) + brain2 = RustBrain() + brain2.restore_from_file(path) + assert brain2.get("k").hlc == orig_hlc + + # Causal successor from the original writer should apply after restore. + brain2.remember("k", "v2", hlc=(5000, 11, "nodeA")) + assert brain2.recall("k") == "v2" + + # Stale replay must still be rejected. + with pytest.raises(TimestampRegression): + brain2.remember("k", "stale", hlc=(5000, 5, "nodeA")) + finally: + os.unlink(path) + + def test_snapshot_with_tenant_isolation(): brain = RustBrain(tenant_id="org_a", tenant_isolation=True) brain.remember("secret", "data") diff --git a/tests/test_gossip.py b/tests/test_gossip.py index f6b3233..bbe60f1 100644 --- a/tests/test_gossip.py +++ b/tests/test_gossip.py @@ -32,3 +32,45 @@ def test_gossip_start_stop(): gossip.start() gossip.stop() # No crash + + +def test_gossip_rejects_stale_hlc_overwrite(): + brain = RustBrain() + gossip = GossipProtocol(brain, peers=[]) + brain.remember("shared", "FRESH", hlc=(2000, 0, "local")) + + applied = gossip.receive( + [{"key": "shared", "value": "STALE", "hlc": [1000, 0, "remote"]}] + ) + assert applied == 0 + assert brain.recall("shared") == "FRESH" + + +def test_gossip_applies_newer_hlc(): + brain = RustBrain() + gossip = GossipProtocol(brain, peers=[]) + brain.remember("shared", "OLD", hlc=(1000, 0, "local")) + + applied = gossip.receive( + [{"key": "shared", "value": "NEW", "hlc": [2000, 0, "remote"]}] + ) + assert applied == 1 + assert brain.recall("shared") == "NEW" + + +def test_gossip_skips_missing_hlc_on_existing_key(): + brain = RustBrain() + gossip = GossipProtocol(brain, peers=[]) + brain.remember("k", "local") + + applied = gossip.receive([{"key": "k", "value": "remote"}]) + assert applied == 0 + assert brain.recall("k") == "local" + + +def test_gossip_allows_missing_hlc_for_new_key(): + brain = RustBrain() + gossip = GossipProtocol(brain, peers=[]) + applied = gossip.receive([{"key": "new_key", "value": "remote"}]) + assert applied == 1 + assert brain.recall("new_key") == "remote"