distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Peer history tracking.
2
3Ported from net.i2p.router.peermanager.DBHistory / TunnelHistory.
4"""
5
6import time
7from dataclasses import dataclass, field
8
9
10@dataclass
11class HistoryEntry:
12 timestamp: float
13 success: bool
14 latency_ms: float = 0.0
15 detail: str = ""
16
17
18class DBHistory:
19 """Historical record of NetDB operations with a peer."""
20
21 def __init__(self, max_entries: int = 100):
22 self._entries: list[HistoryEntry] = []
23 self._max_entries = max_entries
24
25 def _add_entry(self, entry: HistoryEntry) -> None:
26 self._entries.append(entry)
27 if len(self._entries) > self._max_entries:
28 self._entries = self._entries[-self._max_entries:]
29
30 def record_store(self, success: bool, latency_ms: float = 0.0) -> None:
31 self._add_entry(HistoryEntry(
32 timestamp=time.time(),
33 success=success,
34 latency_ms=latency_ms,
35 detail="store",
36 ))
37
38 def record_lookup(self, success: bool, latency_ms: float = 0.0) -> None:
39 self._add_entry(HistoryEntry(
40 timestamp=time.time(),
41 success=success,
42 latency_ms=latency_ms,
43 detail="lookup",
44 ))
45
46 @property
47 def success_rate(self) -> float:
48 if not self._entries:
49 return 0.0
50 successes = sum(1 for e in self._entries if e.success)
51 return successes / len(self._entries)
52
53 @property
54 def average_latency(self) -> float:
55 if not self._entries:
56 return 0.0
57 return sum(e.latency_ms for e in self._entries) / len(self._entries)
58
59 @property
60 def recent_entries(self) -> list[HistoryEntry]:
61 return list(self._entries)
62
63
64class TunnelHistory:
65 """Historical record of tunnel participation."""
66
67 def __init__(self, max_entries: int = 100):
68 self._entries: list[HistoryEntry] = []
69 self._max_entries = max_entries
70
71 def _add_entry(self, entry: HistoryEntry) -> None:
72 self._entries.append(entry)
73 if len(self._entries) > self._max_entries:
74 self._entries = self._entries[-self._max_entries:]
75
76 def record_build(self, success: bool) -> None:
77 self._add_entry(HistoryEntry(
78 timestamp=time.time(),
79 success=success,
80 detail="build",
81 ))
82
83 def record_participation(self, success: bool, data_transferred: int = 0) -> None:
84 self._add_entry(HistoryEntry(
85 timestamp=time.time(),
86 success=success,
87 detail=f"participate:{data_transferred}",
88 ))
89
90 @property
91 def build_success_rate(self) -> float:
92 builds = [e for e in self._entries if e.detail == "build"]
93 if not builds:
94 return 0.0
95 return sum(1 for e in builds if e.success) / len(builds)
96
97 @property
98 def participation_rate(self) -> float:
99 participations = [e for e in self._entries if e.detail.startswith("participate")]
100 if not participations:
101 return 0.0
102 return sum(1 for e in participations if e.success) / len(participations)