From 324f078ab4c2dfed8582dd04dd845e27c05373c9 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 14 Jun 2026 11:03:43 +0000 Subject: [PATCH] fix(rust_brain): preserve HLC on restore and in gossip receive v0.6.0 switched monotonic enforcement from ts_ns to HLC but two call paths were not updated: - restore_from_file() dropped HLC from snapshots, assigning fresh clocks that blocked legitimate post-restore replication and broke causal order. - gossip.receive() omitted HLC, so late-arriving events got a fresh local timestamp and could overwrite newer data on the same key. Restore now round-trips HLC (with ts_ns fallback for legacy snapshots) and advances the process clock. Gossip applies remote HLCs via update_hlc(), rejects stale writes, and skips unordered updates to existing keys when no HLC is present. Regression tests added for both paths. Co-authored-by: Daniel --- hive/gossip.py | 20 ++++++++++++++- hive/rust_brain/__init__.py | 22 ++++++++++++++++- tests/test_enterprise_backup.py | 28 ++++++++++++++++++++- tests/test_gossip.py | 44 +++++++++++++++++++++++++++++++++ 4 files changed, 111 insertions(+), 3 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..84e19b5 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 @@ -509,10 +525,13 @@ def restore_from_file(self, path: str) -> int: 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=node_dict["ts_ns"], + 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]), @@ -524,6 +543,7 @@ def restore_from_file(self, path: str) -> int: 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..153b20d 100644 --- a/tests/test_gossip.py +++ b/tests/test_gossip.py @@ -2,6 +2,8 @@ from __future__ import annotations +import pytest + from hive.gossip import GossipProtocol from hive.rust_brain import RustBrain @@ -32,3 +34,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"