fix: per-bridge TCP routing, gossip in seed mode, and thread-safety
Multi-bridge routing was broken: all incoming radicle-node connections were routed to remote_bridges[0] regardless of which NID was being reached. Fixed by allocating a dedicated OS-assigned TCP port per discovered remote bridge; auto_seed registers each NID at its specific port so radicle-node connections always reach the correct peer. Gossip relay integrated into 'radicle-rns seed': watches the seed's storage, auto-discovers repos as they are seeded, notifies remote gossip peers of ref changes. bridge_port=None skips the redundant rad-node- connect step (bridge's auto_seed already registered NIDs correctly). Other fixes: - bridge.py: link FAILED state now breaks the wait loop immediately instead of waiting out the full 30 s timeout - bridge.py: get_remote_bridge_nid now reads _bridge_nids under lock - bridge.py: NID length bounds-checked before slice in _handle_announce - gossip.py: add rad_home param so seed's rad calls use correct RAD_HOME - gossip.py: add auto_discover flag to scan storage for new repos each cycle - gossip.py: import os moved to top-level - cli.py: gossip status line only prints when peer count changes (not every 5 s on any stats change) - cli.py: add --poll-interval to seed command Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Maciek "mab122" Bator
parent
e051d82af1
commit
5fa6f890b4
|
|
@ -9,9 +9,9 @@ Architecture:
|
|||
└─────────────┘ localhost └─────────────┘ LoRa/mesh └─────────────┘
|
||||
|
||||
The bridge:
|
||||
1. Listens on localhost TCP (default: 8776, Radicle's default port)
|
||||
2. Accepts connections from local radicle-node
|
||||
3. Tunnels traffic over Reticulum to remote bridges
|
||||
1. Allocates a dedicated TCP port per discovered remote bridge
|
||||
2. Accepts connections from local radicle-node on those ports
|
||||
3. Tunnels traffic over Reticulum to the correct remote bridge
|
||||
4. Remote bridges forward to their local radicle-node
|
||||
"""
|
||||
|
||||
|
|
@ -32,7 +32,7 @@ from radicle_reticulum.identity import RadicleIdentity
|
|||
|
||||
# Default ports
|
||||
RADICLE_DEFAULT_PORT = 8776
|
||||
BRIDGE_DEFAULT_PORT = 8777 # Local listen port for bridge
|
||||
BRIDGE_DEFAULT_PORT = 8777 # Base listen port (first bridge gets this)
|
||||
|
||||
# App name for RNS destinations
|
||||
APP_NAME = "radicle"
|
||||
|
|
@ -64,21 +64,21 @@ class TunnelConnection:
|
|||
if self.tcp_socket:
|
||||
try:
|
||||
self.tcp_socket.close()
|
||||
except:
|
||||
except OSError:
|
||||
pass
|
||||
if self.rns_link:
|
||||
try:
|
||||
self.rns_link.teardown()
|
||||
except:
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
class RadicleBridge:
|
||||
"""Bridges Radicle TCP connections over Reticulum.
|
||||
|
||||
Modes:
|
||||
- Server mode: Listens for TCP from local radicle-node, tunnels to remote bridges
|
||||
- Accepts incoming tunnels from remote bridges, forwards to local radicle-node
|
||||
Each discovered remote bridge gets its own dedicated TCP listen port so
|
||||
that radicle-node connections are always routed to the correct peer.
|
||||
The first bridge gets listen_port; subsequent bridges get OS-assigned ports.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
|
|
@ -97,7 +97,8 @@ class RadicleBridge:
|
|||
|
||||
Args:
|
||||
identity: RNS identity for this bridge
|
||||
listen_port: TCP port to listen on for local radicle-node
|
||||
listen_port: Base TCP port; first discovered bridge gets this port,
|
||||
subsequent ones get OS-assigned ports.
|
||||
radicle_host: Host where radicle-node listens (for incoming tunnels)
|
||||
radicle_port: Port where radicle-node listens
|
||||
config_path: Path to Reticulum config
|
||||
|
|
@ -138,13 +139,16 @@ class RadicleBridge:
|
|||
self._remote_bridges: Dict[bytes, float] = {}
|
||||
self._remote_bridges_lock = threading.Lock()
|
||||
|
||||
# Per-bridge dedicated TCP servers: bridge_hash -> server socket / port
|
||||
self._bridge_servers: Dict[bytes, socket.socket] = {}
|
||||
self._bridge_ports: Dict[bytes, int] = {}
|
||||
self._servers_lock = threading.Lock()
|
||||
|
||||
# Active tunnels
|
||||
self._tunnels: Dict[int, TunnelConnection] = {}
|
||||
self._tunnel_counter = 0
|
||||
self._tunnels_lock = threading.Lock()
|
||||
|
||||
# TCP server
|
||||
self._tcp_server: Optional[socket.socket] = None
|
||||
self._running = False
|
||||
|
||||
# Callbacks
|
||||
|
|
@ -155,7 +159,7 @@ class RadicleBridge:
|
|||
# Local radicle node NID (for announcing to remote bridges)
|
||||
self._local_radicle_nid: Optional[str] = None
|
||||
|
||||
# Remote bridge NIDs: bridge_hash -> radicle_nid
|
||||
# Remote bridge NIDs: bridge_hash -> radicle_nid (guarded by _remote_bridges_lock)
|
||||
self._bridge_nids: Dict[bytes, str] = {}
|
||||
|
||||
def start(self):
|
||||
|
|
@ -164,35 +168,23 @@ class RadicleBridge:
|
|||
|
||||
# Register announce handler to discover other bridges
|
||||
RNS.Transport.register_announce_handler(self._handle_announce)
|
||||
RNS.log(f"Registered announce handler for bridge discovery", RNS.LOG_INFO)
|
||||
|
||||
# Start TCP server
|
||||
self._tcp_server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
self._tcp_server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
self._tcp_server.bind(("127.0.0.1", self.listen_port))
|
||||
self._tcp_server.listen(5)
|
||||
self._tcp_server.setblocking(False)
|
||||
|
||||
# Start accept thread
|
||||
self._accept_thread = threading.Thread(target=self._accept_loop, daemon=True)
|
||||
self._accept_thread.start()
|
||||
RNS.log("Registered announce handler for bridge discovery", RNS.LOG_INFO)
|
||||
|
||||
# Announce presence; repeat a few times so peers that come up shortly
|
||||
# after us don't miss it due to interface initialisation timing
|
||||
self.announce()
|
||||
threading.Thread(target=self._startup_announce_loop, daemon=True).start()
|
||||
|
||||
RNS.log(f"Radicle bridge started", RNS.LOG_INFO)
|
||||
RNS.log(f" TCP listen: 127.0.0.1:{self.listen_port}", RNS.LOG_INFO)
|
||||
RNS.log("Radicle bridge started", RNS.LOG_INFO)
|
||||
RNS.log(f" RNS hash: {self.destination.hexhash}", RNS.LOG_INFO)
|
||||
RNS.log(f" Radicle target: {self.radicle_host}:{self.radicle_port}", RNS.LOG_INFO)
|
||||
RNS.log(
|
||||
f" TCP ports: allocated per remote bridge (base port {self.listen_port})",
|
||||
RNS.LOG_INFO,
|
||||
)
|
||||
|
||||
def _startup_announce_loop(self):
|
||||
"""Re-announce after startup to catch peers that come up slightly later.
|
||||
|
||||
Delays are configurable — use long values on LoRa to respect duty cycle.
|
||||
RNS itself also rate-limits announces per interface.
|
||||
"""
|
||||
"""Re-announce after startup to catch peers that come up slightly later."""
|
||||
for delay in self.announce_retry_delays:
|
||||
time.sleep(delay)
|
||||
if not self._running:
|
||||
|
|
@ -209,43 +201,47 @@ class RadicleBridge:
|
|||
tunnel.close()
|
||||
self._tunnels.clear()
|
||||
|
||||
# Close TCP server
|
||||
if self._tcp_server:
|
||||
self._tcp_server.close()
|
||||
# Close all per-bridge TCP servers
|
||||
with self._servers_lock:
|
||||
for srv in self._bridge_servers.values():
|
||||
try:
|
||||
srv.close()
|
||||
except OSError:
|
||||
pass
|
||||
self._bridge_servers.clear()
|
||||
self._bridge_ports.clear()
|
||||
|
||||
RNS.log("Radicle bridge stopped", RNS.LOG_INFO)
|
||||
|
||||
def set_local_radicle_nid(self, nid: str):
|
||||
"""Set the local radicle node's NID for announcement.
|
||||
|
||||
This allows remote bridges to know which radicle NID is reachable
|
||||
through this bridge.
|
||||
"""
|
||||
"""Set the local radicle node's NID for announcement."""
|
||||
self._local_radicle_nid = nid
|
||||
RNS.log(f"Local radicle NID set: {nid[:32]}...", RNS.LOG_INFO)
|
||||
|
||||
def get_remote_bridge_nid(self, bridge_hash: bytes) -> Optional[str]:
|
||||
"""Get the radicle NID served by a remote bridge."""
|
||||
with self._remote_bridges_lock:
|
||||
return self._bridge_nids.get(bridge_hash)
|
||||
|
||||
def announce(self):
|
||||
"""Announce this bridge on the network."""
|
||||
# Build app_data: magic + optional NID
|
||||
app_data = BRIDGE_APP_DATA_MAGIC
|
||||
if self._local_radicle_nid:
|
||||
nid_bytes = self._local_radicle_nid.encode("utf-8")
|
||||
app_data += struct.pack("!H", len(nid_bytes)) + nid_bytes
|
||||
|
||||
self.destination.announce(app_data=app_data)
|
||||
RNS.log(f"Announced bridge: {self.destination.hexhash} (app_data={len(app_data)} bytes)", RNS.LOG_INFO)
|
||||
RNS.log(
|
||||
f"Announced bridge: {self.destination.hexhash} (app_data={len(app_data)} bytes)",
|
||||
RNS.LOG_INFO,
|
||||
)
|
||||
|
||||
def connect_to_bridge(self, destination_hash: bytes, timeout: float = 30.0) -> bool:
|
||||
"""Connect to a remote bridge.
|
||||
"""Connect to a remote bridge (establish RNS path).
|
||||
|
||||
This establishes the RNS link. Actual tunneling happens when
|
||||
local radicle-node connects to our TCP port.
|
||||
Actual per-connection tunneling is set up lazily when radicle-node
|
||||
connects to the allocated TCP port.
|
||||
"""
|
||||
# Check if we have a path
|
||||
if not RNS.Transport.has_path(destination_hash):
|
||||
RNS.Transport.request_path(destination_hash)
|
||||
deadline = time.time() + timeout
|
||||
|
|
@ -255,7 +251,6 @@ class RadicleBridge:
|
|||
return False
|
||||
time.sleep(0.1)
|
||||
|
||||
# Remember this bridge
|
||||
with self._remote_bridges_lock:
|
||||
self._remote_bridges[destination_hash] = time.time()
|
||||
|
||||
|
|
@ -267,45 +262,78 @@ class RadicleBridge:
|
|||
with self._remote_bridges_lock:
|
||||
return list(self._remote_bridges.keys())
|
||||
|
||||
def _accept_loop(self):
|
||||
"""Accept incoming TCP connections from local radicle-node."""
|
||||
# ── Per-bridge TCP server ────────────────────────────────────────────────
|
||||
|
||||
def _allocate_port_for_bridge(self, bridge_hash: bytes) -> int:
|
||||
"""Allocate a dedicated TCP listen port for a remote bridge.
|
||||
|
||||
The first bridge claims listen_port if available; subsequent bridges
|
||||
get OS-assigned ephemeral ports. Idempotent — returns the same port
|
||||
on repeated calls for the same bridge_hash.
|
||||
"""
|
||||
with self._servers_lock:
|
||||
if bridge_hash in self._bridge_ports:
|
||||
return self._bridge_ports[bridge_hash]
|
||||
|
||||
# Try the configured listen_port first (first bridge keeps the
|
||||
# well-known port); fall back to OS-assigned if already taken.
|
||||
preferred = self.listen_port if not self._bridge_ports else 0
|
||||
srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
try:
|
||||
srv.bind(("127.0.0.1", preferred))
|
||||
except OSError:
|
||||
srv.bind(("127.0.0.1", 0))
|
||||
srv.listen(5)
|
||||
srv.setblocking(False)
|
||||
port = srv.getsockname()[1]
|
||||
|
||||
self._bridge_servers[bridge_hash] = srv
|
||||
self._bridge_ports[bridge_hash] = port
|
||||
|
||||
threading.Thread(
|
||||
target=self._accept_loop_for_bridge,
|
||||
args=(srv, bridge_hash),
|
||||
daemon=True,
|
||||
).start()
|
||||
|
||||
RNS.log(
|
||||
f"Allocated TCP port {port} for bridge {bridge_hash.hex()[:16]}",
|
||||
RNS.LOG_INFO,
|
||||
)
|
||||
return port
|
||||
|
||||
def _accept_loop_for_bridge(self, server: socket.socket, bridge_hash: bytes):
|
||||
"""Accept TCP connections destined for a specific remote bridge."""
|
||||
while self._running:
|
||||
try:
|
||||
readable, _, _ = select.select([self._tcp_server], [], [], 1.0)
|
||||
readable, _, _ = select.select([server], [], [], 1.0)
|
||||
if readable:
|
||||
client_socket, addr = self._tcp_server.accept()
|
||||
RNS.log(f"TCP connection from {addr}", RNS.LOG_DEBUG)
|
||||
|
||||
# Handle in new thread
|
||||
thread = threading.Thread(
|
||||
target=self._handle_local_connection,
|
||||
args=(client_socket,),
|
||||
daemon=True,
|
||||
client_socket, addr = server.accept()
|
||||
RNS.log(
|
||||
f"TCP from {addr} → bridge {bridge_hash.hex()[:16]}",
|
||||
RNS.LOG_DEBUG,
|
||||
)
|
||||
thread.start()
|
||||
threading.Thread(
|
||||
target=self._handle_local_connection,
|
||||
args=(client_socket, bridge_hash),
|
||||
daemon=True,
|
||||
).start()
|
||||
except Exception as e:
|
||||
if self._running:
|
||||
RNS.log(f"Accept error: {e}", RNS.LOG_ERROR)
|
||||
RNS.log(
|
||||
f"Accept error for bridge {bridge_hash.hex()[:16]}: {e}",
|
||||
RNS.LOG_ERROR,
|
||||
)
|
||||
|
||||
def _handle_local_connection(self, tcp_socket: socket.socket):
|
||||
"""Handle a connection from local radicle-node.
|
||||
|
||||
Creates tunnel to remote bridge and forwards traffic.
|
||||
"""
|
||||
# Get a remote bridge to tunnel to
|
||||
remote_bridges = self.get_remote_bridges()
|
||||
if not remote_bridges:
|
||||
RNS.log("No remote bridges available", RNS.LOG_WARNING)
|
||||
tcp_socket.close()
|
||||
return
|
||||
|
||||
# Use first available bridge (could add load balancing later)
|
||||
remote_hash = remote_bridges[0]
|
||||
|
||||
# Create RNS link to remote bridge
|
||||
remote_identity = RNS.Identity.recall(remote_hash)
|
||||
def _handle_local_connection(self, tcp_socket: socket.socket, bridge_hash: bytes):
|
||||
"""Handle a connection from local radicle-node, tunneled to bridge_hash."""
|
||||
remote_identity = RNS.Identity.recall(bridge_hash)
|
||||
if not remote_identity:
|
||||
RNS.log(f"Cannot recall identity for {remote_hash.hex()}", RNS.LOG_WARNING)
|
||||
RNS.log(
|
||||
f"Cannot recall identity for {bridge_hash.hex()[:16]}",
|
||||
RNS.LOG_WARNING,
|
||||
)
|
||||
tcp_socket.close()
|
||||
return
|
||||
|
||||
|
|
@ -319,11 +347,14 @@ class RadicleBridge:
|
|||
|
||||
rns_link = RNS.Link(remote_dest)
|
||||
|
||||
# Wait for link establishment
|
||||
# Wait for link establishment (Noise XK handshake)
|
||||
deadline = time.time() + 30.0
|
||||
while rns_link.status != RNS.Link.ACTIVE:
|
||||
if rns_link.status == RNS.Link.CLOSED:
|
||||
RNS.log("Link closed before becoming active (remote refused or unreachable)", RNS.LOG_WARNING)
|
||||
if rns_link.status in (RNS.Link.CLOSED, RNS.Link.FAILED):
|
||||
RNS.log(
|
||||
"Link closed/failed before becoming active",
|
||||
RNS.LOG_WARNING,
|
||||
)
|
||||
tcp_socket.close()
|
||||
return
|
||||
if time.time() > deadline:
|
||||
|
|
@ -332,7 +363,6 @@ class RadicleBridge:
|
|||
return
|
||||
time.sleep(0.1)
|
||||
|
||||
# Create tunnel
|
||||
with self._tunnels_lock:
|
||||
self._tunnel_counter += 1
|
||||
tunnel_id = self._tunnel_counter
|
||||
|
|
@ -341,18 +371,17 @@ class RadicleBridge:
|
|||
tunnel_id=tunnel_id,
|
||||
tcp_socket=tcp_socket,
|
||||
rns_link=rns_link,
|
||||
remote_destination=remote_hash,
|
||||
remote_destination=bridge_hash,
|
||||
)
|
||||
|
||||
with self._tunnels_lock:
|
||||
self._tunnels[tunnel_id] = tunnel
|
||||
|
||||
RNS.log(f"Tunnel {tunnel_id} opened to {remote_hash.hex()[:16]}", RNS.LOG_INFO)
|
||||
RNS.log(f"Tunnel {tunnel_id} opened to {bridge_hash.hex()[:16]}", RNS.LOG_INFO)
|
||||
|
||||
if self._tunnel_opened_cb:
|
||||
self._tunnel_opened_cb(tunnel)
|
||||
|
||||
# Set up bidirectional forwarding
|
||||
rns_link.set_packet_callback(
|
||||
lambda data, pkt: self._on_rns_data(tunnel_id, data)
|
||||
)
|
||||
|
|
@ -360,7 +389,6 @@ class RadicleBridge:
|
|||
lambda link: self._on_tunnel_closed(tunnel_id)
|
||||
)
|
||||
|
||||
# Forward TCP to RNS
|
||||
self._forward_tcp_to_rns(tunnel)
|
||||
|
||||
def _forward_tcp_to_rns(self, tunnel: TunnelConnection):
|
||||
|
|
@ -379,9 +407,8 @@ class RadicleBridge:
|
|||
if readable:
|
||||
data = tcp_socket.recv(RNS_BUFFER_SIZE)
|
||||
if not data:
|
||||
break # Connection closed
|
||||
break
|
||||
|
||||
# Send over RNS
|
||||
if rns_link.status == RNS.Link.ACTIVE:
|
||||
packet = RNS.Packet(rns_link, data)
|
||||
packet.send()
|
||||
|
|
@ -418,17 +445,17 @@ class RadicleBridge:
|
|||
if tunnel:
|
||||
tunnel.close()
|
||||
RNS.log(
|
||||
f"Tunnel {tunnel_id} closed (sent: {tunnel.bytes_sent}, recv: {tunnel.bytes_received})",
|
||||
RNS.LOG_INFO
|
||||
f"Tunnel {tunnel_id} closed "
|
||||
f"(sent: {tunnel.bytes_sent}, recv: {tunnel.bytes_received})",
|
||||
RNS.LOG_INFO,
|
||||
)
|
||||
if self._tunnel_closed_cb:
|
||||
self._tunnel_closed_cb(tunnel)
|
||||
|
||||
def _on_incoming_link(self, link: RNS.Link):
|
||||
"""Handle incoming RNS link from remote bridge."""
|
||||
RNS.log(f"Incoming link from remote bridge", RNS.LOG_DEBUG)
|
||||
RNS.log("Incoming link from remote bridge", RNS.LOG_DEBUG)
|
||||
|
||||
# Connect to local radicle-node
|
||||
try:
|
||||
tcp_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
tcp_socket.connect((self.radicle_host, self.radicle_port))
|
||||
|
|
@ -437,7 +464,6 @@ class RadicleBridge:
|
|||
link.teardown()
|
||||
return
|
||||
|
||||
# Create tunnel
|
||||
with self._tunnels_lock:
|
||||
self._tunnel_counter += 1
|
||||
tunnel_id = self._tunnel_counter
|
||||
|
|
@ -454,7 +480,6 @@ class RadicleBridge:
|
|||
|
||||
RNS.log(f"Incoming tunnel {tunnel_id} opened", RNS.LOG_INFO)
|
||||
|
||||
# Set up bidirectional forwarding
|
||||
link.set_packet_callback(
|
||||
lambda data, pkt: self._on_rns_data(tunnel_id, data)
|
||||
)
|
||||
|
|
@ -462,19 +487,20 @@ class RadicleBridge:
|
|||
lambda l: self._on_tunnel_closed(tunnel_id)
|
||||
)
|
||||
|
||||
# Forward TCP to RNS in background
|
||||
thread = threading.Thread(
|
||||
threading.Thread(
|
||||
target=self._forward_tcp_to_rns,
|
||||
args=(tunnel,),
|
||||
daemon=True,
|
||||
)
|
||||
thread.start()
|
||||
).start()
|
||||
|
||||
def set_on_bridge_discovered(self, callback: Optional[Callable[[bytes, Optional[str]], None]]):
|
||||
# ── Bridge discovery ─────────────────────────────────────────────────────
|
||||
|
||||
def set_on_bridge_discovered(
|
||||
self, callback: Optional[Callable[[bytes, Optional[str]], None]]
|
||||
):
|
||||
"""Set callback for when a new bridge is discovered.
|
||||
|
||||
Callback receives (destination_hash, radicle_nid) where radicle_nid
|
||||
may be None if the remote bridge hasn't set one.
|
||||
Callback: (destination_hash: bytes, radicle_nid: Optional[str])
|
||||
"""
|
||||
self._on_bridge_discovered = callback
|
||||
|
||||
|
|
@ -484,28 +510,30 @@ class RadicleBridge:
|
|||
announced_identity: RNS.Identity,
|
||||
app_data: Optional[bytes],
|
||||
):
|
||||
"""Handle announce - only process if it's a bridge announce."""
|
||||
RNS.log(f"Received announce: {destination_hash.hex()[:16]}... app_data={app_data[:20] if app_data else None}", RNS.LOG_VERBOSE)
|
||||
"""Handle RNS announce — process only bridge announces."""
|
||||
RNS.log(
|
||||
f"Received announce: {destination_hash.hex()[:16]}... "
|
||||
f"app_data={app_data[:20] if app_data else None}",
|
||||
RNS.LOG_VERBOSE,
|
||||
)
|
||||
|
||||
# Ignore our own announcements
|
||||
if destination_hash == self.destination.hash:
|
||||
RNS.log("Ignoring own announcement", RNS.LOG_VERBOSE)
|
||||
return
|
||||
|
||||
# Filter: only accept announces with bridge magic in app_data
|
||||
if app_data is None or not app_data.startswith(BRIDGE_APP_DATA_MAGIC):
|
||||
# Not a bridge announce, ignore
|
||||
RNS.log(f"Ignoring non-bridge announce (no magic)", RNS.LOG_VERBOSE)
|
||||
RNS.log("Ignoring non-bridge announce (no magic)", RNS.LOG_VERBOSE)
|
||||
return
|
||||
|
||||
# Extract radicle NID if present
|
||||
radicle_nid = None
|
||||
if len(app_data) > len(BRIDGE_APP_DATA_MAGIC):
|
||||
try:
|
||||
offset = len(BRIDGE_APP_DATA_MAGIC)
|
||||
nid_len = struct.unpack("!H", app_data[offset:offset+2])[0]
|
||||
nid_len = struct.unpack("!H", app_data[offset:offset + 2])[0]
|
||||
offset += 2
|
||||
nid_str = app_data[offset:offset+nid_len].decode("utf-8")
|
||||
available = len(app_data) - offset
|
||||
nid_len = min(nid_len, available)
|
||||
nid_str = app_data[offset:offset + nid_len].decode("utf-8")
|
||||
radicle_nid = nid_str if nid_str else None
|
||||
except Exception as e:
|
||||
RNS.log(f"Failed to parse bridge app_data: {e}", RNS.LOG_DEBUG)
|
||||
|
|
@ -513,20 +541,22 @@ class RadicleBridge:
|
|||
with self._remote_bridges_lock:
|
||||
is_new = destination_hash not in self._remote_bridges
|
||||
self._remote_bridges[destination_hash] = time.time()
|
||||
# Check and update NID mapping under the same lock to avoid double-registering
|
||||
nid_is_new = bool(radicle_nid and destination_hash not in self._bridge_nids)
|
||||
nid_is_new = bool(
|
||||
radicle_nid and destination_hash not in self._bridge_nids
|
||||
)
|
||||
if radicle_nid:
|
||||
self._bridge_nids[destination_hash] = radicle_nid
|
||||
|
||||
if is_new:
|
||||
nid_info = f" (NID: {radicle_nid[:32]}...)" if radicle_nid else ""
|
||||
RNS.log(f"Discovered bridge: {destination_hash.hex()}{nid_info}", RNS.LOG_INFO)
|
||||
RNS.log(
|
||||
f"Discovered bridge: {destination_hash.hex()}{nid_info}",
|
||||
RNS.LOG_INFO,
|
||||
)
|
||||
|
||||
# Notify callback
|
||||
if self._on_bridge_discovered:
|
||||
self._on_bridge_discovered(destination_hash, radicle_nid)
|
||||
|
||||
# Auto-connect if enabled
|
||||
if self.auto_connect:
|
||||
threading.Thread(
|
||||
target=self._auto_connect_to_bridge,
|
||||
|
|
@ -534,12 +564,10 @@ class RadicleBridge:
|
|||
daemon=True,
|
||||
).start()
|
||||
|
||||
# Auto-register seed whenever we first learn the NID (covers --connect path
|
||||
# where the bridge is pre-registered before its announce arrives)
|
||||
if self.auto_seed and nid_is_new:
|
||||
threading.Thread(
|
||||
target=self._auto_register_seed,
|
||||
args=(radicle_nid,),
|
||||
args=(radicle_nid, destination_hash),
|
||||
daemon=True,
|
||||
).start()
|
||||
|
||||
|
|
@ -549,17 +577,23 @@ class RadicleBridge:
|
|||
if self.connect_to_bridge(destination_hash, timeout=30.0):
|
||||
RNS.log(f"Auto-connected to bridge: {destination_hash.hex()}", RNS.LOG_INFO)
|
||||
else:
|
||||
RNS.log(f"Auto-connect failed for bridge: {destination_hash.hex()}", RNS.LOG_WARNING)
|
||||
RNS.log(
|
||||
f"Auto-connect failed for bridge: {destination_hash.hex()}",
|
||||
RNS.LOG_WARNING,
|
||||
)
|
||||
|
||||
def register_seed(self, radicle_nid: str) -> bool:
|
||||
"""Register a remote radicle NID as a seed through this bridge.
|
||||
def register_seed(self, radicle_nid: str, port: Optional[int] = None) -> bool:
|
||||
"""Register a remote radicle NID as reachable through this bridge.
|
||||
|
||||
Calls 'rad node connect <NID>@127.0.0.1:<listen_port>' to tell
|
||||
radicle-node that the given NID is reachable through our bridge.
|
||||
Calls 'rad node connect <NID>@127.0.0.1:<port>' to tell radicle-node
|
||||
that the given NID is reachable through our bridge TCP server.
|
||||
|
||||
Returns True if successful.
|
||||
port: the bridge's per-NID listen port. Falls back to self.listen_port
|
||||
if not specified (backward-compatible for single-bridge setups).
|
||||
"""
|
||||
addr = f"{radicle_nid}@127.0.0.1:{self.listen_port}"
|
||||
if port is None:
|
||||
port = self.listen_port
|
||||
addr = f"{radicle_nid}@127.0.0.1:{port}"
|
||||
RNS.log(f"Registering seed: {addr}", RNS.LOG_INFO)
|
||||
|
||||
env = None
|
||||
|
|
@ -582,7 +616,7 @@ class RadicleBridge:
|
|||
RNS.log(f"Failed to register seed: {result.stderr}", RNS.LOG_WARNING)
|
||||
return False
|
||||
except FileNotFoundError:
|
||||
RNS.log("'rad' command not found - cannot auto-register seed", RNS.LOG_WARNING)
|
||||
RNS.log("'rad' command not found — cannot auto-register seed", RNS.LOG_WARNING)
|
||||
return False
|
||||
except subprocess.TimeoutExpired:
|
||||
RNS.log("Timeout registering seed", RNS.LOG_WARNING)
|
||||
|
|
@ -591,11 +625,11 @@ class RadicleBridge:
|
|||
RNS.log(f"Error registering seed: {e}", RNS.LOG_WARNING)
|
||||
return False
|
||||
|
||||
def _auto_register_seed(self, radicle_nid: str):
|
||||
"""Auto-register a seed in background after discovery."""
|
||||
# Small delay to allow bridge connection to establish first
|
||||
def _auto_register_seed(self, radicle_nid: str, bridge_hash: bytes):
|
||||
"""Allocate a dedicated TCP port for this bridge and register its NID."""
|
||||
time.sleep(2.0)
|
||||
self.register_seed(radicle_nid)
|
||||
port = self._allocate_port_for_bridge(bridge_hash)
|
||||
self.register_seed(radicle_nid, port)
|
||||
|
||||
def get_stats(self) -> dict:
|
||||
"""Get bridge statistics."""
|
||||
|
|
|
|||
|
|
@ -299,15 +299,15 @@ def cmd_gossip(args):
|
|||
signal.signal(signal.SIGINT, signal_handler)
|
||||
signal.signal(signal.SIGTERM, signal_handler)
|
||||
|
||||
last_stats = None
|
||||
last_known_peers = -1
|
||||
try:
|
||||
while running:
|
||||
stats = relay.get_stats()
|
||||
if stats != last_stats:
|
||||
if stats["known_peers"] != last_known_peers:
|
||||
last_known_peers = stats["known_peers"]
|
||||
print(f"[Status] Peers: {stats['known_peers']}, "
|
||||
f"Repos: {stats['watched_repos']}, "
|
||||
f"Refs: {stats['refs_per_repo']}")
|
||||
last_stats = dict(stats)
|
||||
time.sleep(5)
|
||||
finally:
|
||||
relay.stop()
|
||||
|
|
@ -315,7 +315,7 @@ def cmd_gossip(args):
|
|||
|
||||
|
||||
def cmd_seed(args):
|
||||
"""Start a dedicated seed radicle-node and bridge it to the mesh."""
|
||||
"""Start a dedicated seed radicle-node, bridge, and gossip relay."""
|
||||
seed_home = Path(args.seed_home)
|
||||
|
||||
seed = SeedNode(seed_home=seed_home, port=args.seed_port)
|
||||
|
|
@ -376,6 +376,20 @@ def cmd_seed(args):
|
|||
|
||||
bridge.set_on_bridge_discovered(on_peer_seed_discovered)
|
||||
|
||||
# Gossip relay: watches seed's storage, notifies remote seeds of ref changes.
|
||||
# bridge_port=None: bridge's auto_seed already registered NIDs on correct ports.
|
||||
gossip = GossipRelay(
|
||||
identity=identity,
|
||||
rids=[],
|
||||
storage=seed_home / "storage",
|
||||
radicle_nid=nid,
|
||||
bridge_port=None,
|
||||
poll_interval=args.poll_interval,
|
||||
announce_retry_delays=announce_retry_delays,
|
||||
auto_discover=True,
|
||||
rad_home=str(seed_home),
|
||||
)
|
||||
|
||||
running = True
|
||||
|
||||
def signal_handler(sig, frame):
|
||||
|
|
@ -388,14 +402,16 @@ def cmd_seed(args):
|
|||
|
||||
try:
|
||||
bridge.start()
|
||||
gossip.start()
|
||||
|
||||
print()
|
||||
print("Seed node running.")
|
||||
print(f" Seed NID: {nid}")
|
||||
print(f" Seed port: {args.seed_port}")
|
||||
print(f" Bridge hash: {bridge.destination.hexhash}")
|
||||
print(f" Gossip hash: {gossip.destination.hexhash}")
|
||||
print()
|
||||
print("Add this seed to your radicle node:")
|
||||
print("Add this seed to your radicle node (one-time setup):")
|
||||
print(f" rad node connect {nid}@127.0.0.1:{args.seed_port}")
|
||||
print()
|
||||
print("Other machines running 'radicle-rns seed' will discover this")
|
||||
|
|
@ -414,10 +430,12 @@ def cmd_seed(args):
|
|||
if stats["known_bridges"] != last_known_bridges:
|
||||
last_known_bridges = stats["known_bridges"]
|
||||
print(f"[Status] Remote seeds: {stats['known_bridges']}, "
|
||||
f"Tunnels: {stats['active_tunnels']}")
|
||||
f"Tunnels: {stats['active_tunnels']}, "
|
||||
f"Gossip peers: {gossip.get_stats()['known_peers']}")
|
||||
|
||||
time.sleep(5)
|
||||
finally:
|
||||
gossip.stop()
|
||||
bridge.stop()
|
||||
seed.stop()
|
||||
print("Seed stopped.")
|
||||
|
|
@ -675,6 +693,13 @@ def main():
|
|||
metavar="PORT",
|
||||
help="TCP listen port for the seed bridge (default: 8778)",
|
||||
)
|
||||
seed_parser.add_argument(
|
||||
"--poll-interval",
|
||||
type=int,
|
||||
default=30,
|
||||
metavar="SECONDS",
|
||||
help="Seconds between gossip ref polls (default: 30)",
|
||||
)
|
||||
seed_parser.add_argument(
|
||||
"--announce-retry-delays",
|
||||
default="5,15,30",
|
||||
|
|
|
|||
|
|
@ -15,6 +15,7 @@ Flow:
|
|||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import struct
|
||||
import subprocess
|
||||
import threading
|
||||
|
|
@ -87,10 +88,12 @@ class GossipRelay:
|
|||
rids: List[str],
|
||||
storage: Optional[Path] = None,
|
||||
radicle_nid: Optional[str] = None,
|
||||
bridge_port: int = 8777,
|
||||
bridge_port: Optional[int] = 8777,
|
||||
poll_interval: int = DEFAULT_POLL_INTERVAL,
|
||||
announce_retry_delays: Tuple[int, ...] = (5, 15, 30),
|
||||
config_path: Optional[str] = None,
|
||||
auto_discover: bool = False,
|
||||
rad_home: Optional[str] = None,
|
||||
):
|
||||
"""
|
||||
Args:
|
||||
|
|
@ -98,10 +101,15 @@ class GossipRelay:
|
|||
rids: List of Radicle repository IDs to watch (e.g. 'rad:z3abc...').
|
||||
storage: Path to radicle storage dir. Auto-detected if None.
|
||||
radicle_nid: Local radicle NID to advertise to peers.
|
||||
bridge_port: TCP port the bridge listens on (for rad node connect).
|
||||
bridge_port: TCP port for 'rad node connect' in _trigger_sync.
|
||||
Pass None to skip that step (seed mode: bridge's auto_seed
|
||||
already registered NIDs on correct per-bridge ports).
|
||||
poll_interval: Seconds between ref polls.
|
||||
announce_retry_delays: Startup re-announce delays (seconds).
|
||||
config_path: Reticulum config path (None = default).
|
||||
auto_discover: Scan storage each poll cycle and add new repo dirs
|
||||
to rids automatically. Useful in seed mode.
|
||||
rad_home: RAD_HOME override for rad CLI calls. None = system default.
|
||||
"""
|
||||
self.identity = identity
|
||||
self.rids = list(rids)
|
||||
|
|
@ -110,6 +118,8 @@ class GossipRelay:
|
|||
self.bridge_port = bridge_port
|
||||
self.poll_interval = poll_interval
|
||||
self.announce_retry_delays = announce_retry_delays
|
||||
self.auto_discover = auto_discover
|
||||
self.rad_home = rad_home
|
||||
|
||||
self.reticulum = RNS.Reticulum(config_path)
|
||||
|
||||
|
|
@ -196,8 +206,25 @@ class GossipRelay:
|
|||
return
|
||||
self.announce()
|
||||
|
||||
def _discover_rids(self):
|
||||
"""Scan storage for new repo dirs and add them to self.rids."""
|
||||
if not self.storage.exists():
|
||||
return
|
||||
try:
|
||||
for repo_dir in self.storage.iterdir():
|
||||
if repo_dir.is_dir():
|
||||
rid = f"rad:{repo_dir.name}"
|
||||
if rid not in self.rids:
|
||||
self.rids.append(rid)
|
||||
RNS.log(f"Auto-discovered repo: {rid[:40]}", RNS.LOG_INFO)
|
||||
except Exception as e:
|
||||
RNS.log(f"Error scanning storage: {e}", RNS.LOG_DEBUG)
|
||||
|
||||
def _poll_loop(self):
|
||||
while self._running:
|
||||
if self.auto_discover:
|
||||
self._discover_rids()
|
||||
|
||||
for rid in self.rids:
|
||||
try:
|
||||
refs = _read_refs(self.storage, rid)
|
||||
|
|
@ -304,15 +331,20 @@ class GossipRelay:
|
|||
|
||||
def _trigger_sync(self, rid: str, nid: str):
|
||||
"""Run rad node connect (if needed) then rad sync --fetch."""
|
||||
if nid:
|
||||
env = None
|
||||
if self.rad_home:
|
||||
env = os.environ.copy()
|
||||
env["RAD_HOME"] = self.rad_home
|
||||
|
||||
if nid and self.bridge_port is not None:
|
||||
subprocess.run(
|
||||
["rad", "node", "connect", f"{nid}@127.0.0.1:{self.bridge_port}"],
|
||||
capture_output=True, timeout=15,
|
||||
capture_output=True, timeout=15, env=env,
|
||||
)
|
||||
|
||||
result = subprocess.run(
|
||||
["rad", "sync", "--fetch", "--rid", rid],
|
||||
capture_output=True, text=True, timeout=120,
|
||||
capture_output=True, text=True, timeout=120, env=env,
|
||||
)
|
||||
|
||||
if result.returncode == 0:
|
||||
|
|
|
|||
Loading…
Reference in New Issue