distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""OutboundCache — per-destination send state caching.
2
3Ported from net.i2p.router.message.OutboundClientMessageOneShotJob
4(internal cache fields).
5
6Caches lease, tunnel, and reply state for destination pairs
7to avoid redundant lookups and rate-limit ACK requests.
8"""
9
10from __future__ import annotations
11
12from dataclasses import dataclass
13
14
15@dataclass(frozen=True)
16class HashPair:
17 """Immutable pair of (from_hash, to_hash) used as cache key."""
18 from_hash: bytes
19 to_hash: bytes
20
21 def __eq__(self, other: object) -> bool:
22 if not isinstance(other, HashPair):
23 return NotImplemented
24 return self.from_hash == other.from_hash and self.to_hash == other.to_hash
25
26 def __hash__(self) -> int:
27 return hash((self.from_hash, self.to_hash))
28
29
30class OutboundCache:
31 """Per-destination send state cache.
32
33 Caches recently used leases, tunnels, and ACK timing
34 to optimize outbound message routing.
35 """
36
37 REPLY_REQUEST_INTERVAL_MS = 60_000 # min 60s between ACK requests
38
39 def __init__(self) -> None:
40 self._lease_cache: dict[HashPair, object] = {}
41 self._tunnel_cache: dict[HashPair, object] = {}
42 self._leaseset_acked: dict[HashPair, int] = {} # pair -> ack timestamp
43 self._last_reply_request: dict[HashPair, int] = {} # pair -> request timestamp
44
45 # --- Lease caching ---
46
47 def get_cached_lease(self, pair: HashPair) -> object | None:
48 return self._lease_cache.get(pair)
49
50 def set_cached_lease(self, pair: HashPair, lease: object) -> None:
51 self._lease_cache[pair] = lease
52
53 def clear_cached_lease(self, pair: HashPair) -> None:
54 self._lease_cache.pop(pair, None)
55
56 # --- Tunnel caching ---
57
58 def get_cached_tunnel(self, pair: HashPair) -> object | None:
59 return self._tunnel_cache.get(pair)
60
61 def set_cached_tunnel(self, pair: HashPair, tunnel: object) -> None:
62 self._tunnel_cache[pair] = tunnel
63
64 def clear_cached_tunnel(self, pair: HashPair) -> None:
65 self._tunnel_cache.pop(pair, None)
66
67 # --- Reply request rate limiting ---
68
69 def should_request_reply(self, pair: HashPair, now_ms: int) -> bool:
70 """True if enough time has passed since last reply request."""
71 last = self._last_reply_request.get(pair)
72 if last is None:
73 return True
74 return (now_ms - last) >= self.REPLY_REQUEST_INTERVAL_MS
75
76 def record_reply_request(self, pair: HashPair, now_ms: int) -> None:
77 self._last_reply_request[pair] = now_ms
78
79 # --- LeaseSet ACK tracking ---
80
81 def is_leaseset_acked(self, pair: HashPair) -> bool:
82 return pair in self._leaseset_acked
83
84 def record_leaseset_ack(self, pair: HashPair, now_ms: int) -> None:
85 self._leaseset_acked[pair] = now_ms
86
87 def clear_leaseset_ack(self, pair: HashPair) -> None:
88 self._leaseset_acked.pop(pair, None)