silk-graph 0.2.4

Merkle-CRDT graph engine for distributed, conflict-free knowledge graphs
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115

"""R-05: Gossip Peer Selection — logarithmic fan-out for scalable sync.

Tests verifying peer registry and sync target selection.
"""

import json
import math
from silk import GraphStore

ONTOLOGY = json.dumps({
    "node_types": {"entity": {"properties": {}}},
    "edge_types": {}
})


def _store():
    return GraphStore("test", ONTOLOGY)


def test_register_and_list():
    store = _store()
    store.register_peer("a", "tcp://a:7701")
    store.register_peer("b", "tcp://b:7701")
    peers = store.list_peers()
    assert len(peers) == 2
    ids = {p["peer_id"] for p in peers}
    assert ids == {"a", "b"}


def test_unregister():
    store = _store()
    store.register_peer("a", "tcp://a:7701")
    assert store.unregister_peer("a") is True
    assert store.unregister_peer("a") is False
    assert len(store.list_peers()) == 0


def test_select_empty():
    store = _store()
    assert store.select_sync_targets() == []


def test_select_one_peer():
    store = _store()
    store.register_peer("only", "tcp://only:7701")
    targets = store.select_sync_targets()
    assert targets == ["only"]


def test_select_two_peers():
    store = _store()
    store.register_peer("a", "tcp://a:7701")
    store.register_peer("b", "tcp://b:7701")
    targets = store.select_sync_targets()
    assert len(targets) == 2
    assert set(targets) == {"a", "b"}


def test_select_logarithmic_100():
    """100 peers → ceil(ln(100) + 1) = 6 targets."""
    store = _store()
    for i in range(100):
        store.register_peer(f"p{i}", f"tcp://p{i}:7701")
    targets = store.select_sync_targets()
    expected = math.ceil(math.log(100) + 1)
    assert len(targets) == expected


def test_select_logarithmic_1000():
    """1000 peers → ceil(ln(1000) + 1) = 8 targets."""
    store = _store()
    for i in range(1000):
        store.register_peer(f"p{i}", f"tcp://p{i}:7701")
    targets = store.select_sync_targets()
    expected = math.ceil(math.log(1000) + 1)
    assert len(targets) == expected


def test_select_no_duplicates():
    store = _store()
    for i in range(50):
        store.register_peer(f"p{i}", f"tcp://p{i}:7701")
    targets = store.select_sync_targets()
    assert len(targets) == len(set(targets))


def test_select_all_valid():
    store = _store()
    ids = {f"p{i}" for i in range(50)}
    for pid in ids:
        store.register_peer(pid, f"tcp://{pid}:7701")
    targets = store.select_sync_targets()
    for t in targets:
        assert t in ids


def test_record_sync():
    store = _store()
    store.register_peer("a", "tcp://a:7701")
    peers = store.list_peers()
    assert peers[0]["last_seen_ms"] == 0
    store.record_sync("a")
    peers = store.list_peers()
    assert peers[0]["last_seen_ms"] > 0


def test_peer_info_structure():
    store = _store()
    store.register_peer("test-peer", "tcp://10.0.0.1:7701")
    peers = store.list_peers()
    assert len(peers) == 1
    p = peers[0]
    assert p["peer_id"] == "test-peer"
    assert p["address"] == "tcp://10.0.0.1:7701"
    assert isinstance(p["last_seen_ms"], int)