From 73a5b48a89623d36077ea4de339ef62ed6ea18c8 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 30 Aug 2026 11:01:35 +0000 Subject: [PATCH] fix(rust_brain): preserve HLC timestamps on snapshot restore and gossip MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit restore_from_file() assigned fresh HLC values instead of restoring the snapshotted timestamps. After disaster recovery, gossip replays and causal successor writes with pre-crash HLC were rejected with TimestampRegression — silent data loss in distributed deployments. - Add _parse_hlc() for wire/snapshot normalisation - Restore HLC and advance global clock on snapshot restore - Pass HLC through bulk_write() row payloads - Apply HLC in gossip.receive() with stale-update rejection - Hold lock during restore to prevent concurrent mutation Regression tests in test_hlc_snapshot_gossip.py (29 targeted tests pass). Co-authored-by: Daniel --- hive/gossip.py | 19 ++++- hive/rust_brain/__init__.py | 59 +++++++++----- tests/test_hlc_snapshot_gossip.py | 125 ++++++++++++++++++++++++++++++ 3 files changed, 183 insertions(+), 20 deletions(-) create mode 100644 tests/test_hlc_snapshot_gossip.py diff --git a/hive/gossip.py b/hive/gossip.py index f0e5372..d8c3cdd 100644 --- a/hive/gossip.py +++ b/hive/gossip.py @@ -122,11 +122,28 @@ 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, ) applied += 1 except Exception as exc: diff --git a/hive/rust_brain/__init__.py b/hive/rust_brain/__init__.py index a246c1f..ce521cc 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 @@ -505,25 +521,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_hlc_snapshot_gossip.py b/tests/test_hlc_snapshot_gossip.py new file mode 100644 index 0000000..f7eea38 --- /dev/null +++ b/tests/test_hlc_snapshot_gossip.py @@ -0,0 +1,125 @@ +"""Regression tests for HLC preservation across snapshot restore and gossip.""" + +from __future__ import annotations + +import os +import tempfile + +import pytest + +from hive.gossip import GossipProtocol +from hive.rust_brain import RustBrain, TimestampRegression + + +def test_restore_preserves_hlc_and_allows_causal_successor(): + """After restore, writes with causally later HLC must not be rejected.""" + brain = RustBrain(tenant_id="test") + original_hlc = (5000, 10, "nodeA") + brain.remember("k1", "v1", hlc=original_hlc) + + with tempfile.NamedTemporaryFile(delete=False, suffix=".gz") as f: + path = f.name + + try: + brain.snapshot_to_file(path) + + brain2 = RustBrain(tenant_id="test") + brain2.restore_from_file(path) + node = brain2.get("k1") + assert node is not None + assert node.hlc == original_hlc + + successor_hlc = (5000, 11, "nodeA") + brain2.remember("k1", "v2", hlc=successor_hlc) + updated = brain2.get("k1") + assert updated is not None + assert updated.value == "v2" + assert updated.hlc == successor_hlc + finally: + os.unlink(path) + + +def test_restore_rejects_stale_hlc_after_restore(): + """Writes with HLC earlier than restored state must still raise.""" + brain = RustBrain(tenant_id="test") + brain.remember("k1", "v1", hlc=(5000, 10, "nodeA")) + + with tempfile.NamedTemporaryFile(delete=False, suffix=".gz") as f: + path = f.name + + try: + brain.snapshot_to_file(path) + brain2 = RustBrain(tenant_id="test") + brain2.restore_from_file(path) + + with pytest.raises(TimestampRegression): + brain2.remember("k1", "stale", hlc=(5000, 9, "nodeA")) + finally: + os.unlink(path) + + +def test_bulk_write_preserves_hlc(): + """bulk_write must round-trip HLC from row payloads.""" + brain = RustBrain() + rows = [ + {"key": "a", "value": "1", "hlc": [1000, 0, "n1"]}, + {"key": "b", "value": "2", "hlc": [1001, 1, "n2"]}, + ] + brain.bulk_write(rows) + + node_a = brain.get("a") + node_b = brain.get("b") + assert node_a is not None + assert node_b is not None + assert node_a.hlc == (1000, 0, "n1") + assert node_b.hlc == (1001, 1, "n2") + + +def test_gossip_receive_applies_hlc(): + """Gossip receive must apply HLC for causal ordering.""" + brain = RustBrain() + gossip = GossipProtocol(brain, peers=[]) + + applied = gossip.receive( + [ + { + "key": "remote", + "value": "from_peer", + "hlc": [2000, 0, "peer1"], + } + ] + ) + assert applied == 1 + node = brain.get("remote") + assert node is not None + assert node.hlc == (2000, 0, "peer1") + + +def test_gossip_receive_rejects_stale_update(): + """Gossip with stale HLC must not overwrite newer local state.""" + brain = RustBrain() + brain.remember("k", "local", hlc=(3000, 5, "local")) + + gossip = GossipProtocol(brain, peers=[]) + applied = gossip.receive( + [ + { + "key": "k", + "value": "stale_peer", + "hlc": [3000, 3, "peer"], + } + ] + ) + assert applied == 0 + assert brain.recall("k") == "local" + + +def test_gossip_receive_skips_missing_hlc_on_existing_key(): + """Updates without HLC must not overwrite existing keys.""" + brain = RustBrain() + brain.remember("k", "original", hlc=(4000, 0, "local")) + + gossip = GossipProtocol(brain, peers=[]) + applied = gossip.receive([{"key": "k", "value": "no_hlc_update"}]) + assert applied == 0 + assert brain.recall("k") == "original"