beacon v2 discovery: meshdisco.py on 239.255.99.1:9099, core located by metamesh.core cap, /api/neighbors ?cap=, nav lists services only
This commit is contained in:
@@ -19,7 +19,7 @@ External port in the dev stack: **8182** (auth-gated, via Caddy + nginx-hash-loc
|
||||
│ ▼ ▼ ▼ │
|
||||
│ ┌─────────────────────────────────────────────────────┐ │
|
||||
│ │ meta-core (owns Redis; /urls + /meta APIs) │ │
|
||||
│ │ located over UDP multicast (meta-discovery v1) │ │
|
||||
│ │ located over UDP multicast (beacon v2) │ │
|
||||
│ └─────────────────────────────────────────────────────┘ │
|
||||
│ │ │
|
||||
│ ▼ │
|
||||
@@ -34,7 +34,7 @@ External port in the dev stack: **8182** (auth-gated, via Caddy + nginx-hash-loc
|
||||
- **HLS transcoding** — adaptive preset+CRF targeting a 60–80% transcode ratio, prefetch of N segments ahead, persistent segment cache.
|
||||
- **Stremio protocol** — manifest, catalog, meta, stream; one HLS stream per audio track; subtitles extracted to VTT.
|
||||
- **Direct file serving** — range-request capable, for clients that can play the source directly.
|
||||
- **meta-core over HTTP** — `LeaderStorage` locates meta-core via **meta-discovery v1** (UDP multicast `239.255.77.1:9399`; see [`service-discovery.md`](../../docs/project-architecture/service-discovery.md)) or a pinned `META_CORE_URL`, reads every record through meta-core's `/meta/*` API, and keeps its cache fresh from the SSE meta-stream. The WebDAV URL is taken from meta-core's announce/`/urls` — nothing is configured by hand and no `/meta-core` volume is mounted. (`RedisStorage` remains in `src/storage/` as a legacy, unused backend; meta-core no longer publishes a Redis URL.)
|
||||
- **meta-core over HTTP** — `LeaderStorage` locates meta-core via **beacon v2** (UDP multicast `239.255.99.1:9099`; see [`beacon-v2.md`](../../docs/project-architecture/beacon-v2.md)) or a pinned `META_CORE_URL`, reads every record through meta-core's `/meta/*` API, and keeps its cache fresh from the SSE meta-stream. The WebDAV URL is taken from meta-core's announce/`/urls` — nothing is configured by hand and no `/meta-core` volume is mounted. (`RedisStorage` remains in `src/storage/` as a legacy, unused backend; meta-core no longer publishes a Redis URL.)
|
||||
- **Neighbour nav** — this service announces itself on the discovery group and serves its own neighbour map at `/api/neighbors` for the dashboard's shared `<meta-service-menu>`.
|
||||
- **Path-token gate for addon URLs** — when `HASH_API_SEED` is set, addon paths must be prefixed with `/s/<token>`. Dashboard/browser paths are protected separately by the upstream hash-lock + OIDC sidecar (Stremio clients can't carry cookies, so they get the path-token instead).
|
||||
- **Language-configurable manifest** — `displayLanguage` in the per-install config (base64-in-path) selects which `titles` translation meta-sort's TMDB plugin populated.
|
||||
@@ -89,7 +89,7 @@ python server.py # finds meta-core over UDP, or set META_CORE_URL=<meta-core A
|
||||
| `PUBLIC_URL` | — | Browser-facing URL announced to neighbours (nav menu); wins over `BASE_URL` for the announce. |
|
||||
| `STORAGE_MODE` | `leader` (image) | Read only by `docker/start.sh`: `leader` starts the bundled meta-core sidecar in-container, anything else (e.g. `direct`) skips it. The Python server always uses `LeaderStorage`. |
|
||||
| `META_CORE_URL` | — | Pins meta-core's HTTP API (e.g. `http://metacore-app:9000`); when set, UDP discovery never overrides it. Unset → meta-core is located over UDP. |
|
||||
| `ENABLE_UDP_DISCOVERY` | `true` | meta-discovery v1 announce/listen. |
|
||||
| `ENABLE_UDP_DISCOVERY` | `true` | beacon v2 announce/listen. |
|
||||
| `LEADER_WAIT_TIMEOUT` | `0` | Seconds to wait for meta-core at startup (`0` = forever); on timeout, keeps retrying in the background. |
|
||||
| `LEADER_RETRY_INTERVAL` | `5` | Seconds between meta-core lookup attempts. |
|
||||
| `FILES_PATH` | `/files` | Files volume (local fallback when WebDAV is not configured). |
|
||||
@@ -113,7 +113,7 @@ All routes below are matched in `src/server.py`. When `HASH_API_SEED` is set, **
|
||||
| `/health` | GET | Liveness + storage status |
|
||||
| `/api/stats` | GET | Library statistics |
|
||||
| `/api/library` | GET | Full library list (videos + count) |
|
||||
| `/api/neighbors` | GET | This peer's UDP neighbour map (meta-discovery v1) for the nav menu |
|
||||
| `/api/neighbors` | GET | This peer's UDP neighbour map (beacon v2) for the nav menu |
|
||||
| `/api/services` | GET | Alias of `/api/neighbors` (kept for older dashboards) |
|
||||
| `/api/languages` | GET | Languages selectable from the `/configure` page |
|
||||
| `/meta-service-menu.js` | GET | Shared `<meta-service-menu>` nav element loaded by the dashboard |
|
||||
@@ -180,7 +180,7 @@ meta-stremio/
|
||||
│ ├── provider.py # StorageProvider + VideoMetadata abstract base
|
||||
│ ├── leader_storage.py # Storage backend in use: meta-core HTTP API + SSE cache invalidation
|
||||
│ ├── leader_client.py # Locates meta-core (UDP or META_CORE_URL pin), exposes its /urls set
|
||||
│ ├── meshdisco.py # meta-discovery v1 (Python port of the Go reference in meta-core)
|
||||
│ ├── meshdisco.py # beacon v2 (Python port of the Go reference in meta-core)
|
||||
│ ├── meta_consumer.py # SSE consumer from meta-core /meta stream
|
||||
│ ├── meta_core_api_client.py # HTTP client for meta-core REST API
|
||||
│ └── redis_storage.py # Legacy direct-Redis backend (unused)
|
||||
|
||||
+12
-5
@@ -66,7 +66,7 @@ DASHBOARD_PATHS = frozenset({
|
||||
# Initialize storage
|
||||
storage = stremio.init_storage()
|
||||
|
||||
# Service discovery is UDP now (meta-discovery v1), so there is no
|
||||
# Service discovery is UDP now (beacon v2), so there is no
|
||||
# META_CORE_PATH gate: the leader client's node announces this service on the
|
||||
# same multicast group it listens on. Being discoverable and discovering are
|
||||
# one thing. See docs/project-architecture/service-discovery.md.
|
||||
@@ -654,18 +654,25 @@ class Handler(BaseHTTPRequestHandler):
|
||||
self.wfile.write(content.encode('utf-8'))
|
||||
|
||||
def handle_services_api(self):
|
||||
"""Neighbours for the dashboard nav, from this peer's own UDP map.
|
||||
"""Neighbours for the dashboard nav, from this peer's own beacon v2 view.
|
||||
|
||||
No hop through meta-core any more: every service keeps its own
|
||||
neighbour table, so the nav still renders when meta-core is down.
|
||||
`services` is kept as an alias of `neighbors` so the existing
|
||||
dashboards keep working during the UI migration.
|
||||
`?all=1` returns every instance, `?cap=<pattern>` filters by
|
||||
capability (docs/project-architecture/beacon-v2.md). `services` is kept
|
||||
as an alias of `neighbors` so the existing dashboards keep working.
|
||||
"""
|
||||
from urllib.parse import parse_qs, urlparse
|
||||
|
||||
from storage.leader_client import get_leader_client
|
||||
|
||||
try:
|
||||
query = parse_qs(urlparse(self.path).query)
|
||||
client = get_leader_client()
|
||||
neighbors = client.neighbors() if client else []
|
||||
neighbors = client.neighbors(
|
||||
all=bool(query.get('all')),
|
||||
cap=(query.get('cap') or [None])[0],
|
||||
) if client else []
|
||||
return self.send_json({
|
||||
'current': 'meta-stremio',
|
||||
'enabled': True,
|
||||
|
||||
@@ -1,9 +1,9 @@
|
||||
"""LeaderClient — locates meta-core and exposes its URLs.
|
||||
|
||||
Since meta-discovery v1 this is a thin adapter over ``MetaCoreLocator``
|
||||
(``meshdisco.py``): meta-core is found by UDP announce instead of by reading
|
||||
``/meta-core/locks/kv-leader.info``, so meta-stremio no longer needs the
|
||||
``/meta-core`` volume mounted at all.
|
||||
Since UDP discovery (meta-discovery v1, now beacon v2) this is a thin adapter
|
||||
over ``MetaCoreLocator`` (``meshdisco.py``): meta-core is found by UDP advertise
|
||||
instead of by reading ``/meta-core/locks/kv-leader.info``, so meta-stremio no
|
||||
longer needs the ``/meta-core`` volume mounted at all.
|
||||
|
||||
The public surface is deliberately unchanged so ``LeaderStorage`` did not have
|
||||
to move: ``get_leader_info``, ``get_urls``, ``get_api_url``,
|
||||
@@ -263,10 +263,11 @@ class LeaderClient:
|
||||
|
||||
# -- nav menu ----------------------------------------------------------
|
||||
|
||||
def neighbors(self) -> List[dict]:
|
||||
"""Neighbours for the nav menu (one row per service name)."""
|
||||
def neighbors(self, all: bool = False, cap: Optional[str] = None) -> List[dict]:
|
||||
"""Neighbours for /api/neighbors: one row per name (every instance with
|
||||
``all``), optionally only nodes with a capability matching ``cap``."""
|
||||
self._ensure_started()
|
||||
return self._locator.neighbors()
|
||||
return self._locator.neighbors(all=all, cap=cap)
|
||||
|
||||
def self_announce(self) -> dict:
|
||||
"""This service's own announce, so the menu can show itself."""
|
||||
|
||||
+213
-119
@@ -1,13 +1,13 @@
|
||||
"""meta-discovery v1 — UDP multicast service discovery.
|
||||
"""Beacon v2 — local resource advertise / discover over UDP multicast
|
||||
(239.255.99.1:9099).
|
||||
|
||||
Replaces the two file-based mechanisms that required a shared /meta-core
|
||||
volume: ``locks/kv-leader.info`` (locating meta-core at boot) and
|
||||
``services/<name>-<host>.json`` (the dashboard nav registry). Both answered the
|
||||
same question — where is meta-core, and who else is on this network — so one
|
||||
announce packet answers both, and discovery scope becomes the docker network
|
||||
rather than the mounted volume.
|
||||
Every node advertises resources tagged with capability strings
|
||||
(``metamesh.core``, ``metamesh.service/meta-stremio``,
|
||||
``metamesh.transport/nzb@1``) and keeps a live view of everyone else's. This
|
||||
service uses it to locate meta-core (the ``metamesh.core`` resource carries its
|
||||
/urls block) and to feed the nav menu, with no shared volume.
|
||||
|
||||
The normative spec is docs/project-architecture/service-discovery.md.
|
||||
The normative spec is docs/project-architecture/beacon-v2.md.
|
||||
|
||||
Two details this module must get right, both of which fail silently:
|
||||
|
||||
@@ -30,18 +30,98 @@ import threading
|
||||
import time
|
||||
from typing import Callable, Dict, List, Optional
|
||||
|
||||
PROTOCOL_VERSION = 1
|
||||
DEFAULT_GROUP = "239.255.77.1"
|
||||
DEFAULT_PORT = 9399
|
||||
PROTO = "beacon"
|
||||
VERSION = 2
|
||||
DEFAULT_GROUP = "239.255.99.1"
|
||||
DEFAULT_PORT = 9099
|
||||
DEFAULT_INTERVAL = 10.0
|
||||
# A neighbour unseen for interval * this is dropped. Replaces the reaper.
|
||||
# A node unheard for interval * this is dropped (evaluated at read time).
|
||||
LIVENESS_FACTOR = 3
|
||||
READ_BUFFER = 8192
|
||||
# Senders stay under this so nothing fragments.
|
||||
MAX_DATAGRAM = 1400
|
||||
|
||||
TYPE_DISCOVERY = "discovery"
|
||||
TYPE_ANNOUNCE = "announce"
|
||||
ROLE_CORE = "core"
|
||||
ROLE_SERVICE = "service"
|
||||
TYPE_PROBE = "probe"
|
||||
TYPE_ADVERTISE = "advertise"
|
||||
TYPE_BYE = "bye"
|
||||
|
||||
CAP_CORE = "metamesh.core"
|
||||
CAP_SERVICE_PREFIX = "metamesh.service/"
|
||||
CAP_ANY_SERVICE = "metamesh.service/*"
|
||||
|
||||
|
||||
def cap_matches(pattern: str, cap: str) -> bool:
|
||||
"""Does capability ``cap`` satisfy ``pattern``?
|
||||
|
||||
``*`` matches everything; an ``@N`` on the pattern must equal the cap's
|
||||
contract (none ignores it); ``x/*`` matches any variant of ``x``; otherwise
|
||||
the bases must be equal.
|
||||
"""
|
||||
if pattern == "*":
|
||||
return True
|
||||
pbase, _, pc = pattern.rpartition("@") if "@" in pattern else (pattern, "", None)
|
||||
cbase, _, cc = cap.rpartition("@") if "@" in cap else (cap, "", None)
|
||||
if pc is not None and cc != pc:
|
||||
return False
|
||||
if pbase.endswith("/*"):
|
||||
prefix = pbase[:-2]
|
||||
if not cbase.startswith(prefix):
|
||||
return False
|
||||
rest = cbase[len(prefix):]
|
||||
return rest.startswith("/") and len(rest) > 1
|
||||
return pbase == cbase
|
||||
|
||||
|
||||
def resource_matches(resource: dict, pattern: str) -> bool:
|
||||
return any(cap_matches(pattern, c) for c in resource.get("caps") or [])
|
||||
|
||||
|
||||
def resource_endpoint(resource: dict, name: str) -> Optional[str]:
|
||||
"""The named endpoint as an absolute URL ('/…' joins onto endpoints.http)."""
|
||||
endpoints = resource.get("endpoints") or {}
|
||||
v = endpoints.get(name)
|
||||
if not v:
|
||||
return None
|
||||
if not v.startswith("/"):
|
||||
return v
|
||||
base = endpoints.get("http")
|
||||
return base.rstrip("/") + v if base else None
|
||||
|
||||
|
||||
def parse_message(data: bytes) -> Optional[dict]:
|
||||
"""Decode a datagram; None for anything that is not well-formed beacon v2
|
||||
(beacon v1, meta-discovery v1, other versions, junk)."""
|
||||
try:
|
||||
m = json.loads(data.decode("utf-8"))
|
||||
except (ValueError, UnicodeDecodeError):
|
||||
return None
|
||||
if not isinstance(m, dict) or m.get("proto") != PROTO or m.get("v") != VERSION:
|
||||
return None
|
||||
t = m.get("type")
|
||||
if t == TYPE_PROBE:
|
||||
return m
|
||||
if t in (TYPE_ADVERTISE, TYPE_BYE):
|
||||
node = m.get("node")
|
||||
if not isinstance(node, dict) or not node.get("name") or not node.get("instance"):
|
||||
return None
|
||||
return m
|
||||
return None
|
||||
|
||||
|
||||
def _neighbor_row(node: dict, resources: List[dict], addr: str, last_seen: float) -> dict:
|
||||
caps: List[str] = []
|
||||
base_url = None
|
||||
for r in resources:
|
||||
for c in r.get("caps") or []:
|
||||
if c not in caps:
|
||||
caps.append(c)
|
||||
if base_url is None and resource_matches(r, CAP_ANY_SERVICE):
|
||||
base_url = resource_endpoint(r, "ui")
|
||||
row = dict(node)
|
||||
row.update({"resources": resources, "caps": caps, "addr": addr, "lastSeen": last_seen})
|
||||
if base_url:
|
||||
row["baseUrl"] = base_url
|
||||
return row
|
||||
|
||||
|
||||
def _env_flag(name: str, default: bool = True) -> bool:
|
||||
@@ -97,33 +177,33 @@ def local_ipv4() -> str:
|
||||
|
||||
|
||||
class MeshNode:
|
||||
"""Announces this service, answers probes, tracks neighbours."""
|
||||
"""One beacon v2 participant: advertises its resources, answers probes,
|
||||
and keeps a TTL map of every other node."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
name: str,
|
||||
role: str = ROLE_SERVICE,
|
||||
instance: Optional[str] = None,
|
||||
version: str = "",
|
||||
group: str = DEFAULT_GROUP,
|
||||
port: int = DEFAULT_PORT,
|
||||
interval: float = DEFAULT_INTERVAL,
|
||||
group: Optional[str] = None,
|
||||
port: Optional[int] = None,
|
||||
interval: Optional[float] = None,
|
||||
enabled: bool = True,
|
||||
payload: Optional[Callable[[], dict]] = None,
|
||||
) -> None:
|
||||
self.name = name
|
||||
self.role = role
|
||||
self.instance = instance or socket.gethostname()
|
||||
self.version = version
|
||||
self.group = group
|
||||
self.port = port
|
||||
self.interval = interval
|
||||
self.group = group or os.environ.get("BEACON_GROUP") or DEFAULT_GROUP
|
||||
self.port = port or int(os.environ.get("BEACON_PORT") or DEFAULT_PORT)
|
||||
self.interval = interval or (float(os.environ.get("BEACON_INTERVAL_MS") or 0) / 1000.0) or DEFAULT_INTERVAL
|
||||
self.enabled = enabled
|
||||
# payload() -> {"status": str?, "resources": [resource, ...]}
|
||||
self._payload = payload
|
||||
|
||||
self._sock: Optional[socket.socket] = None
|
||||
self._ifaces: List[tuple] = []
|
||||
self._neighbors: Dict[str, dict] = {}
|
||||
self._nodes: Dict[str, dict] = {}
|
||||
self._lock = threading.Lock()
|
||||
self._send_lock = threading.Lock()
|
||||
self._running = False
|
||||
@@ -135,9 +215,8 @@ class MeshNode:
|
||||
def start(self) -> None:
|
||||
if self._running or not self.enabled:
|
||||
if not self.enabled:
|
||||
print("[meshdisco] Disabled by configuration")
|
||||
print("[beacon] Disabled by configuration")
|
||||
return
|
||||
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
try:
|
||||
@@ -160,16 +239,16 @@ class MeshNode:
|
||||
sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
|
||||
joined += 1
|
||||
except OSError as e:
|
||||
print(f"[meshdisco] Warning: join {self.group} on {ifname} failed: {e}")
|
||||
print(f"[beacon] Warning: join {self.group} on {ifname} failed: {e}")
|
||||
|
||||
self._sock = sock
|
||||
self._running = True
|
||||
print(
|
||||
f"[meshdisco] Listening on {self.group}:{self.port} as "
|
||||
f"[beacon] Listening on {self.group}:{self.port} as "
|
||||
f"{self.name}/{self.instance} (joined {joined} of {len(self._ifaces)} interfaces)"
|
||||
)
|
||||
|
||||
for target in (self._read_loop, self._announce_loop):
|
||||
for target in (self._read_loop, self._advertise_loop):
|
||||
t = threading.Thread(target=target, daemon=True)
|
||||
t.start()
|
||||
self._threads.append(t)
|
||||
@@ -178,10 +257,10 @@ class MeshNode:
|
||||
if not self._running:
|
||||
return
|
||||
self._running = False
|
||||
# Final announce so neighbours drop us immediately rather than waiting
|
||||
# out the staleness window.
|
||||
# Bye, so neighbours drop us immediately rather than waiting out the
|
||||
# staleness window.
|
||||
try:
|
||||
self._send(self._build_announce("stopping"))
|
||||
self._send({"proto": PROTO, "v": VERSION, "type": TYPE_BYE, "node": self._node_info()})
|
||||
except OSError:
|
||||
pass
|
||||
if self._sock:
|
||||
@@ -193,19 +272,25 @@ class MeshNode:
|
||||
|
||||
# -- api ---------------------------------------------------------------
|
||||
|
||||
def probe(self) -> None:
|
||||
"""Multicast a probe; every listener replies immediately."""
|
||||
self._send({"v": PROTOCOL_VERSION, "type": TYPE_DISCOVERY})
|
||||
def probe(self, want: Optional[List[str]] = None) -> None:
|
||||
"""Multicast a probe. Every node owning a resource matching one of
|
||||
``want`` (every node, when empty) replies immediately."""
|
||||
msg = {"proto": PROTO, "v": VERSION, "type": TYPE_PROBE, "from": self.instance}
|
||||
if want:
|
||||
msg["want"] = list(want)
|
||||
self._send(msg)
|
||||
|
||||
def neighbors(self) -> List[dict]:
|
||||
"""Every live node, sorted by (name, instance)."""
|
||||
cutoff = time.time() - self.interval * LIVENESS_FACTOR
|
||||
with self._lock:
|
||||
for key in [k for k, v in self._neighbors.items() if v["lastSeen"] < cutoff]:
|
||||
del self._neighbors[key]
|
||||
out = list(self._neighbors.values())
|
||||
return sorted(out, key=lambda n: n.get("name") or "")
|
||||
for key in [k for k, v in self._nodes.items() if v["lastSeen"] < cutoff]:
|
||||
del self._nodes[key]
|
||||
out = [_neighbor_row(v["node"], v["resources"], v["addr"], v["lastSeen"]) for v in self._nodes.values()]
|
||||
return sorted(out, key=lambda n: (n.get("name") or "", n.get("instance") or ""))
|
||||
|
||||
def neighbors_by_name(self) -> List[dict]:
|
||||
"""One row per node name, most recently seen instance wins."""
|
||||
best: Dict[str, dict] = {}
|
||||
for nb in self.neighbors():
|
||||
cur = best.get(nb.get("name") or "")
|
||||
@@ -213,55 +298,53 @@ class MeshNode:
|
||||
best[nb.get("name") or ""] = nb
|
||||
return sorted(best.values(), key=lambda n: n.get("name") or "")
|
||||
|
||||
def cores(self) -> List[dict]:
|
||||
return [n for n in self.neighbors() if n.get("role") == ROLE_CORE and n.get("urls")]
|
||||
|
||||
def self_announce(self) -> dict:
|
||||
msg = self._build_announce("running")
|
||||
msg["addr"] = ""
|
||||
msg["lastSeen"] = time.time()
|
||||
return msg
|
||||
"""This node as a row, so a UI can render itself without waiting."""
|
||||
status, resources = self._get_payload()
|
||||
return _neighbor_row(self._node_info(status), resources, "", time.time())
|
||||
|
||||
def on_announce(self, cb: Callable[[dict], None]) -> None:
|
||||
def on_advertise(self, cb: Callable[[dict], None]) -> None:
|
||||
self._listeners.append(cb)
|
||||
|
||||
# -- internals ---------------------------------------------------------
|
||||
|
||||
def _build_announce(self, status: str) -> dict:
|
||||
msg = {
|
||||
"v": PROTOCOL_VERSION,
|
||||
"type": TYPE_ANNOUNCE,
|
||||
"name": self.name,
|
||||
"instance": self.instance,
|
||||
"role": self.role,
|
||||
"version": self.version,
|
||||
"status": status,
|
||||
def _get_payload(self) -> tuple:
|
||||
if not self._payload:
|
||||
return None, []
|
||||
try:
|
||||
p = self._payload() or {}
|
||||
except Exception:
|
||||
p = {}
|
||||
return p.get("status"), list(p.get("resources") or [])
|
||||
|
||||
def _node_info(self, status: Optional[str] = None) -> dict:
|
||||
n = {"name": self.name, "instance": self.instance}
|
||||
if self.version:
|
||||
n["version"] = self.version
|
||||
if status:
|
||||
n["status"] = status
|
||||
return n
|
||||
|
||||
def _advertise_msg(self) -> dict:
|
||||
status, resources = self._get_payload()
|
||||
return {
|
||||
"proto": PROTO, "v": VERSION, "type": TYPE_ADVERTISE,
|
||||
"node": self._node_info(status), "resources": resources,
|
||||
}
|
||||
if self._payload:
|
||||
try:
|
||||
p = self._payload() or {}
|
||||
except Exception:
|
||||
p = {}
|
||||
if p.get("baseUrl"):
|
||||
msg["baseUrl"] = p["baseUrl"]
|
||||
if p.get("status") and status == "running":
|
||||
msg["status"] = p["status"]
|
||||
# Only a core may carry a URLs block.
|
||||
if self.role == ROLE_CORE and p.get("urls"):
|
||||
msg["urls"] = p["urls"]
|
||||
return msg
|
||||
|
||||
def _send(self, msg: dict, dest: Optional[tuple] = None) -> None:
|
||||
sock = self._sock
|
||||
if sock is None:
|
||||
return
|
||||
body = json.dumps(msg).encode("utf-8")
|
||||
if len(body) > MAX_DATAGRAM:
|
||||
print(f"[beacon] Warning: {len(body)}-byte datagram exceeds the fragmentation-safe {MAX_DATAGRAM}")
|
||||
with self._send_lock:
|
||||
if dest is not None:
|
||||
try:
|
||||
sock.sendto(body, dest)
|
||||
except OSError as e:
|
||||
print(f"[meshdisco] Reply to {dest} failed: {e}")
|
||||
print(f"[beacon] Reply to {dest} failed: {e}")
|
||||
return
|
||||
# One write per interface — the default route alone reaches a
|
||||
# single network.
|
||||
@@ -274,7 +357,7 @@ class MeshNode:
|
||||
)
|
||||
sock.sendto(body, (self.group, self.port))
|
||||
except OSError as e:
|
||||
print(f"[meshdisco] Warning: send on {ifname} failed: {e}")
|
||||
print(f"[beacon] Warning: send on {ifname} failed: {e}")
|
||||
|
||||
def _read_loop(self) -> None:
|
||||
while self._running:
|
||||
@@ -287,54 +370,55 @@ class MeshNode:
|
||||
if not self._running:
|
||||
return
|
||||
continue
|
||||
try:
|
||||
msg = json.loads(data.decode("utf-8"))
|
||||
except (ValueError, UnicodeDecodeError):
|
||||
continue # foreign traffic on the group
|
||||
if msg.get("v") != PROTOCOL_VERSION:
|
||||
continue
|
||||
# Drop our own multicast echo.
|
||||
if msg.get("instance") == self.instance and msg.get("name") == self.name:
|
||||
msg = parse_message(data)
|
||||
if msg is None:
|
||||
continue # v1 or foreign traffic on the shared group
|
||||
|
||||
if msg["type"] == TYPE_PROBE:
|
||||
if msg.get("from") == self.instance:
|
||||
continue
|
||||
ad = self._advertise_msg()
|
||||
want = msg.get("want") or []
|
||||
if not want or any(resource_matches(r, p) for r in ad["resources"] for p in want):
|
||||
self._send(ad, dest=addr)
|
||||
continue
|
||||
|
||||
if msg.get("type") == TYPE_DISCOVERY:
|
||||
self._send(self._build_announce("running"), dest=addr)
|
||||
continue
|
||||
if msg.get("type") != TYPE_ANNOUNCE or not msg.get("name"):
|
||||
continue
|
||||
|
||||
key = f"{msg['name']}|{msg.get('instance', '')}"
|
||||
if msg.get("status") == "stopping":
|
||||
node = msg["node"]
|
||||
if node["instance"] == self.instance:
|
||||
continue # our own multicast echo
|
||||
if msg["type"] == TYPE_BYE:
|
||||
with self._lock:
|
||||
self._neighbors.pop(key, None)
|
||||
self._nodes.pop(node["instance"], None)
|
||||
continue
|
||||
|
||||
# Source address comes from the packet, never the payload.
|
||||
msg["addr"] = addr[0]
|
||||
msg["lastSeen"] = time.time()
|
||||
entry = {"node": node, "resources": msg.get("resources") or [], "addr": addr[0], "lastSeen": time.time()}
|
||||
with self._lock:
|
||||
self._neighbors[key] = msg
|
||||
self._nodes[node["instance"]] = entry
|
||||
row = _neighbor_row(entry["node"], entry["resources"], entry["addr"], entry["lastSeen"])
|
||||
for cb in list(self._listeners):
|
||||
try:
|
||||
cb(msg)
|
||||
cb(row)
|
||||
except Exception as e:
|
||||
print(f"[meshdisco] Listener raised: {e}")
|
||||
print(f"[beacon] Listener raised: {e}")
|
||||
|
||||
def _announce_loop(self) -> None:
|
||||
self._send(self._build_announce("running"))
|
||||
def _advertise_loop(self) -> None:
|
||||
self._send(self._advertise_msg())
|
||||
self.probe()
|
||||
while self._running:
|
||||
time.sleep(self.interval)
|
||||
if not self._running:
|
||||
return
|
||||
try:
|
||||
self._send(self._build_announce("running"))
|
||||
self._send(self._advertise_msg())
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
class MetaCoreLocator:
|
||||
"""Locates meta-core over UDP, replacing the kv-leader.info read."""
|
||||
"""Locates meta-core over beacon v2 (the first ``metamesh.core`` resource
|
||||
carrying ``data.urls``) and advertises this service as
|
||||
``metamesh.service/<service_name>``."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
@@ -350,44 +434,49 @@ class MetaCoreLocator:
|
||||
|
||||
self._node = MeshNode(
|
||||
name=service_name,
|
||||
role=ROLE_SERVICE,
|
||||
version=version,
|
||||
enabled=_env_flag("ENABLE_UDP_DISCOVERY", True) if enabled is None else enabled,
|
||||
payload=lambda: {"baseUrl": base_url or f"http://{local_ipv4()}"},
|
||||
payload=lambda: {"resources": [{
|
||||
"id": "service",
|
||||
"caps": [CAP_SERVICE_PREFIX + service_name],
|
||||
"endpoints": {"ui": base_url or f"http://{local_ipv4()}"},
|
||||
}]},
|
||||
)
|
||||
self._node.on_announce(self._on_announce)
|
||||
self._node.on_advertise(self._on_advertise)
|
||||
|
||||
def _on_announce(self, nb: dict) -> None:
|
||||
if nb.get("role") != ROLE_CORE or not nb.get("urls"):
|
||||
return
|
||||
def _on_advertise(self, nb: dict) -> None:
|
||||
if self._pin:
|
||||
return # pinned: the wire cannot move us
|
||||
core = next((r for r in nb.get("resources") or [] if resource_matches(r, CAP_CORE)), None)
|
||||
urls = ((core or {}).get("data") or {}).get("urls")
|
||||
if not isinstance(urls, dict) or not urls.get("apiUrl"):
|
||||
return
|
||||
|
||||
prev = self._current
|
||||
if prev is None:
|
||||
self._current = nb["urls"]
|
||||
print(f"[meshdisco] meta-core discovered at {nb['urls']['apiUrl']} ({nb.get('addr')})")
|
||||
self._current = urls
|
||||
print(f"[beacon] meta-core discovered at {urls['apiUrl']} ({nb.get('addr')})")
|
||||
for cb in list(self._change_callbacks):
|
||||
try:
|
||||
cb()
|
||||
except Exception as e:
|
||||
print(f"[meshdisco] Change callback raised: {e}")
|
||||
print(f"[beacon] Change callback raised: {e}")
|
||||
return
|
||||
if prev.get("apiUrl") == nb["urls"].get("apiUrl"):
|
||||
self._current = nb["urls"]
|
||||
if prev.get("apiUrl") == urls.get("apiUrl"):
|
||||
self._current = urls
|
||||
return
|
||||
# A second, different core. Do not flap — keep the first and say so,
|
||||
# because on a PCS box this means a client container can see a core it
|
||||
# must not use.
|
||||
print(
|
||||
f"[meshdisco] Ignoring a second meta-core at {nb['urls']['apiUrl']} "
|
||||
f"[beacon] Ignoring a second meta-core at {urls['apiUrl']} "
|
||||
f"({nb.get('addr')}); staying with {prev.get('apiUrl')}. "
|
||||
f"Set META_CORE_URL to pin this explicitly."
|
||||
)
|
||||
|
||||
def start(self) -> None:
|
||||
if self._pin:
|
||||
print(f"[meshdisco] meta-core pinned to {self._pin}; discovery is advisory")
|
||||
print(f"[beacon] meta-core pinned to {self._pin}; discovery is advisory")
|
||||
self._node.start()
|
||||
|
||||
def stop(self) -> None:
|
||||
@@ -409,7 +498,7 @@ class MetaCoreLocator:
|
||||
|
||||
``timeout=None`` waits forever, which is meta-stremio's default
|
||||
(LEADER_WAIT_TIMEOUT=0). Re-probes every 500ms rather than waiting out
|
||||
an announce interval.
|
||||
an advertise interval.
|
||||
"""
|
||||
deadline = None if timeout is None else time.time() + timeout
|
||||
logged = False
|
||||
@@ -418,20 +507,25 @@ class MetaCoreLocator:
|
||||
if url:
|
||||
return url
|
||||
if not logged:
|
||||
print("[meshdisco] Waiting for meta-core...")
|
||||
print("[beacon] Waiting for meta-core...")
|
||||
logged = True
|
||||
self._node.probe()
|
||||
self._node.probe([CAP_CORE])
|
||||
time.sleep(0.5)
|
||||
return None
|
||||
|
||||
def on_change(self, cb: Callable[[], None]) -> None:
|
||||
self._change_callbacks.append(cb)
|
||||
|
||||
def neighbors(self) -> List[dict]:
|
||||
return self._node.neighbors_by_name()
|
||||
def neighbors(self, all: bool = False, cap: Optional[str] = None) -> List[dict]:
|
||||
"""One row per name (every instance with ``all``), optionally only
|
||||
nodes with a resource matching ``cap``."""
|
||||
rows = self._node.neighbors() if all else self._node.neighbors_by_name()
|
||||
if cap:
|
||||
rows = [r for r in rows if any(resource_matches(res, cap) for res in r.get("resources") or [])]
|
||||
return rows
|
||||
|
||||
def self_announce(self) -> dict:
|
||||
return self._node.self_announce()
|
||||
|
||||
def probe(self) -> None:
|
||||
self._node.probe()
|
||||
def probe(self, want: Optional[List[str]] = None) -> None:
|
||||
self._node.probe(want)
|
||||
|
||||
@@ -238,6 +238,10 @@ class MetaServiceMenu extends HTMLElement {
|
||||
for (const n of this._neighbors) {
|
||||
const name = n.name || "";
|
||||
if (!name || seen.has(name)) continue;
|
||||
// Beacon v2 neighbours include plugins; the nav lists services only.
|
||||
// (A row without `caps` comes from a pre-v2 backend — keep it.)
|
||||
const isService = (c) => String(c).startsWith("metamesh.service/");
|
||||
if (Array.isArray(n.caps) && !n.caps.some(isService)) continue;
|
||||
seen.add(name);
|
||||
rows.push(n);
|
||||
}
|
||||
@@ -275,7 +279,7 @@ class MetaServiceMenu extends HTMLElement {
|
||||
el.href = url;
|
||||
el.setAttribute("role", "menuitem");
|
||||
}
|
||||
if (!isActive && !url) el.title = "No reachable URL announced";
|
||||
if (!isActive && !url) el.title = "No reachable URL advertised";
|
||||
|
||||
const icon = document.createElement("span");
|
||||
icon.className = "icon";
|
||||
@@ -285,7 +289,8 @@ class MetaServiceMenu extends HTMLElement {
|
||||
label.textContent = titleCase(n.name);
|
||||
|
||||
el.append(icon, label);
|
||||
if (n.status === "running" || isActive) {
|
||||
// Beacon v2 omits `status` when running.
|
||||
if ((n.status || "running") === "running" || isActive) {
|
||||
const dot = document.createElement("span");
|
||||
dot.className = "dot";
|
||||
dot.title = n.status || "running";
|
||||
|
||||
Reference in New Issue
Block a user