fix(rust_brain): preserve HLC on snapshot restore, bulk_write, and gossip

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:

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:

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"