distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Built-in router jobs — standard periodic and one-shot tasks.
2
3Ported from various net.i2p.router.* job classes.
4
5Each job has a minimal ``run()`` / ``execute()`` that logs and delegates
6to the subsystem it wraps. The actual tunnel-building, NetDB searching,
7and peer-profiling logic lives in the respective subsystem modules.
8"""
9
10from __future__ import annotations
11
12import logging
13from typing import Any
14
15from i2p_router.job import Job, TimedJob
16
17log = logging.getLogger(__name__)
18
19
20class TunnelBuildJob(TimedJob):
21 """Build new tunnels when tunnel pools are low.
22
23 Runs every 60 seconds. Delegates to the TunnelManager on the context.
24 """
25
26 def __init__(self, priority: int = 5) -> None:
27 super().__init__(
28 name="tunnel-build",
29 priority=priority,
30 interval_seconds=60.0,
31 )
32
33 def run(self, context: Any) -> None:
34 log.debug("TunnelBuildJob: checking tunnel pools")
35 if hasattr(context, "tunnel_manager"):
36 context.tunnel_manager.build_if_needed()
37
38
39class TunnelRefreshJob(TimedJob):
40 """Remove expired tunnels and request rebuilds.
41
42 Runs every 30 seconds. Delegates to the TunnelManager on the context.
43 """
44
45 def __init__(self, priority: int = 4) -> None:
46 super().__init__(
47 name="tunnel-refresh",
48 priority=priority,
49 interval_seconds=30.0,
50 )
51
52 def run(self, context: Any) -> None:
53 log.debug("TunnelRefreshJob: expiring stale tunnels")
54 if hasattr(context, "tunnel_manager"):
55 context.tunnel_manager.expire_tunnels()
56
57
58class DatabaseSearchJob(Job):
59 """One-shot iterative Kademlia search in the network database.
60
61 This is a one-shot job (not a TimedJob) — it executes once and is
62 removed from the queue.
63 """
64
65 def __init__(self, priority: int = 3) -> None:
66 super().__init__(name="database-search", priority=priority)
67
68 def execute(self, context: Any) -> None:
69 log.debug("DatabaseSearchJob: running iterative lookup")
70 if hasattr(context, "netdb_handler"):
71 context.netdb_handler.run_search()
72
73
74class PeerProfileUpdateJob(TimedJob):
75 """Update peer statistics and prune dead peers.
76
77 Runs every 120 seconds.
78 """
79
80 def __init__(self, priority: int = 7) -> None:
81 super().__init__(
82 name="peer-profile-update",
83 priority=priority,
84 interval_seconds=120.0,
85 )
86
87 def run(self, context: Any) -> None:
88 log.debug("PeerProfileUpdateJob: updating peer profiles")
89 if hasattr(context, "peer_manager"):
90 context.peer_manager.update_profiles()