From 27ed45999b6c691c29c80bb3293251131fbf8d06 Mon Sep 17 00:00:00 2001 From: Worph Date: Sun, 4 Oct 2026 11:51:03 +0000 Subject: [PATCH] beacon v2 discovery: advertise metamesh.core + service resources on 239.255.99.1:9099, /api/neighbors ?cap= filter --- README.md | 28 +-- cmd/meta-core/main.go | 31 ++- dashboard/src/meta-service-menu.js | 9 +- go.mod | 2 +- internal/api/handlers.go | 19 +- internal/api/server.go | 4 +- internal/config/config.go | 22 +- internal/meshdisco/node.go | 261 ++++++++++++----------- internal/meshdisco/proto.go | 327 +++++++++++++++++++---------- internal/meshdisco/proto_test.go | 140 ++++++++++++ 10 files changed, 559 insertions(+), 284 deletions(-) create mode 100644 internal/meshdisco/proto_test.go diff --git a/README.md b/README.md index b57f89c..16a5f7e 100644 --- a/README.md +++ b/README.md @@ -8,8 +8,8 @@ MetaMesh dev/prod stack. meta-core runs as its own container (`metacore-app`). Other MetaMesh services (meta-sort, meta-fuse, meta-stremio, meta-dup, meta-search, meta-share, -meta-gateway, meta-watch, …) locate it over UDP multicast (meta-discovery v1 — -its `role=core` announce carries the full `/urls` payload) and then call its +meta-gateway, meta-watch, …) locate it over UDP multicast (beacon v2 — +its `metamesh.core` advertise carries the full `/urls` payload) and then call its HTTP API. `META_CORE_URL` on a client pins it instead. Redis is not published outside the container — all metadata I/O is mediated by the HTTP/SSE API (see [`docs/api-mediated-access.md`](docs/api-mediated-access.md)). @@ -24,7 +24,7 @@ outside the container — all metadata I/O is mediated by the HTTP/SSE API | **In-memory search index** | `POST /api/metadata/search` matches against an in-memory snapshot of every record, kept current by incremental refreshes (full reconcile every 30 min) — no Redis round-trip per record on the request path (`internal/api/search_index.go`). | | **SSE event streams** | `/api/events/files` and `/api/events/meta` mirror the underlying Redis Streams so external services never speak Redis. | | **WebDAV server** | `/webdav/*` exposes the `/files` volume (read+write) for cross-service file access. nginx in front of meta-core handles caching. | -| **Service discovery** | meta-discovery v1: UDP multicast announce/listen on `239.255.77.1:9399`, neighbours served at `/neighbors`. The file-based registry under `/meta-core/services/*.json` it replaced is gone. | +| **Service discovery** | beacon v2: UDP multicast advertise/listen on `239.255.99.1:9099`, neighbours served at `/neighbors`. The file-based registry under `/meta-core/services/*.json` it replaced is gone. | | **File watcher / scanner** | Recursive scan + MidHash256 computation; emits events on `file:events`. Watched roots are POSTed watcher configs (`/meta-core/watchers.json`), not env vars. | | **Mount management** | rclone-only mount manager (SMB rendered into `:smb:`, plus pre-defined remotes). Read-only by construction. | | **UUID-rooted storage** | Roots are UUIDv7 (Crockford Base32, ULID layout); CIDs are reverse-index aliases. Schema-version sentinel refuses to boot against stale data. | @@ -58,7 +58,7 @@ outside the container — all metadata I/O is mediated by the HTTP/SSE API │ ├─ rclone rcd (mount daemon, RC API on :5572) │ │ ├─ mount-watcher.sh (rclone mount lifecycle) │ │ └─ meta-core (Go) ─► HTTP API + SSE + WebDAV on :9000 │ -│ ─► UDP announce/listen 239.255.77.1:9399 │ +│ ─► UDP advertise/listen 239.255.99.1:9099 │ └──────────────────────────────────────────────────────────────────────────┘ │ ┌──────────────────────┴──────────────────────┐ @@ -72,7 +72,7 @@ outside the container — all metadata I/O is mediated by the HTTP/SSE API └─────────────────────────────────────────────┘ meta-sort / meta-fuse / meta-stremio / meta-search / meta-share / … - │ 1. hear role=core announce on UDP multicast (carries /urls) + │ 1. hear the metamesh.core advertise on UDP multicast (carries /urls) └─ 2. HTTP + SSE (+ WebDAV) to metacore-app — never Redis, never the volume ``` @@ -140,7 +140,7 @@ Environment variables (from `internal/config/config.go` unless noted): |---|---|---| | `META_CORE_PATH` | `/meta-core` | meta-core's own volume root (lock, Redis data, identity, mounts, watchers, cache). No other service mounts it. | | `FILES_PATH` | `/files` | Files volume root. | -| `SERVICE_NAME` | `meta-core` | Name in the meta-discovery announce. | +| `SERVICE_NAME` | `meta-core` | Name in the beacon v2 advertise. | | `SERVICE_VERSION` | `1.0.0` | Reported via `/status`. | | `API_PORT` | `8180` | External port baked into the constructed `baseUrl` when `BASE_URL` is unset. | | `BASE_URL` | _empty_ | Overrides the constructed `baseUrl` (the Caddy/HTTPS perimeter URL). | @@ -149,9 +149,9 @@ Environment variables (from `internal/config/config.go` unless noted): | `META_CORE_HTTP_PORT` | `9000` | Go HTTP+SSE+WebDAV port. | | `META_CORE_HTTP_HOST` | `0.0.0.0` | HTTP bind. | | `ENABLE_FILE_WATCHER` | `true` | Disable to suppress the in-process watcher entirely (`POST /api/files/register` stays available). | -| `ENABLE_UDP_DISCOVERY` | `true` | meta-discovery v1 announce + listen. | -| `DISCOVERY_GROUP` / `DISCOVERY_PORT` | `239.255.77.1` / `9399` | Multicast group and UDP port. | -| `DISCOVERY_INTERVAL_MS` | `10000` | Unsolicited announce interval. | +| `ENABLE_UDP_DISCOVERY` | `true` | beacon v2 advertise + listen. | +| `BEACON_GROUP` / `BEACON_PORT` | `239.255.99.1` / `9099` | Multicast group and UDP port. | +| `BEACON_INTERVAL_MS` | `10000` | Unsolicited advertise interval. | | `META_CORE_INDEX_EXCLUDE_PREFIXES` | `categories/newznab/` | Comma-separated field prefixes left out of the in-memory search index; set empty to index everything (`internal/storage/client.go`). | | `ELECTION_RETRY_SECS` | `5` | flock retry interval (read by `docker/leader-election.sh`, not the Go binary). | @@ -178,14 +178,14 @@ GET /health # storage + role GET /status # version, uptime, etc. GET /leader # current leader info (hostname, PID, timestamps) GET /urls # baseUrl, apiUrl, webdavUrl, webdavUrlInternal (redisUrl is intentionally empty) -GET /neighbors # neighbours heard over UDP (meta-discovery v1) +GET /neighbors # neighbours heard over UDP (beacon v2) GET /api/neighbors # same, under /api GET /api/stats # dashboard counters (see below) ``` `/services`, `/api/services`, `/services/{name}` and `/services/cleanup/stats` are **gone** — the registration-and-heartbeat registry they served was replaced -by meta-discovery v1 (`/neighbors`). nginx answers an unknown non-`/api` path +by beacon v2 (`/neighbors`). nginx answers an unknown non-`/api` path with the dashboard's `index.html`, so a stale caller gets `200 text/html` rather than a 404; that is what made the dashboard fail with `SyntaxError: Unexpected token '<'` until 1.0.21. @@ -395,7 +395,7 @@ binary. The lock is a **mutex, not service discovery** — it only stops two Redis writers landing on one RDB/AOF. Nothing reads it, and the `kv-leader.info` file it used to publish is gone: siblings locate meta-core over UDP -(see [service-discovery.md](../../docs/project-architecture/service-discovery.md)). +(see [beacon-v2.md](../../docs/project-architecture/beacon-v2.md)). `internal/leader` keeps its name for history, but only builds the `/urls` payload (`baseUrl`, `apiUrl`, `webdavUrl`, `webdavUrlInternal`) that `/urls` and the UDP announce share. @@ -418,7 +418,7 @@ the repo-root [`METADATA_KEYS.md`](../../METADATA_KEYS.md). | `cmd/meta-core` | Entry point; Redis connect + schema sentinel + UDP discovery + service wiring. | | `internal/config` | Env-driven configuration, path helpers. | | `internal/leader` | `LeaderInfoProvider` — builds the `/urls` payload from hostname/IP/config; no election logic. | -| `internal/meshdisco` | meta-discovery v1 announcer + listener (the Go reference implementation). | +| `internal/meshdisco` | beacon v2 node (advertise, probe, neighbour map) — the Go port of the spec. | | `internal/storage` | Redis wrapper (`client.go`), UUIDv7 minting, CID reverse index (`cid_resolution.go`), schema sentinel, field indexes + search-index bulk reads, UDL store (`udl.go`, `udl_admin.go`), stats, one-shot migrations/sweeps. | | `internal/cid` | CID parsing + rank ladder for canonical-CID selection (`cid-rank-vectors.json`, guarded by the meta-root `scripts/check-cid-vectors.sh`). | | `internal/identity` | Multi-account secp256k1 keystore, challenge/proof-of-possession, signature verification. | @@ -438,7 +438,7 @@ the repo-root [`METADATA_KEYS.md`](../../METADATA_KEYS.md). 2. Build LeaderInfoProvider (no election — the bash gate already won) 3. Connect to local Redis (retry up to 30× / 1s) 4. EnsureSchemaVersion — abort if the existing Redis layout is stale -5. Start meta-discovery v1 node (unless ENABLE_UDP_DISCOVERY=false) +5. Start beacon v2 node (unless ENABLE_UDP_DISCOVERY=false) 6. Construct API server (identity keystore migration, watcher, watchers, mounts, WebDAV) 7. Start API server + pollers + search-index refresh loop; start diff --git a/cmd/meta-core/main.go b/cmd/meta-core/main.go index d9c6bf0..391d1b0 100644 --- a/cmd/meta-core/main.go +++ b/cmd/meta-core/main.go @@ -1,6 +1,7 @@ package main import ( + "encoding/json" "fmt" "log" "os" @@ -60,37 +61,47 @@ func main() { } log.Printf("[meta-core] Redis schema version OK (v%d)", storage.SchemaVersion) - // UDP service discovery (meta-discovery v1). Runs alongside the file - // registry above for now — nothing consumes the wire yet, so this cannot - // regress the existing path. The announce payload is rebuilt from - // leaderProvider on every tick, which is what keeps it and GET /urls from - // drifting apart. + // Beacon v2 (docs/project-architecture/beacon-v2.md): advertise meta-core + // as metamesh.core — carrying the /urls block services locate us by — plus + // its own metamesh.service/ nav entry. The payload is rebuilt from + // leaderProvider on every tick, which keeps it and GET /urls from drifting. var mesh *meshdisco.Node if cfg.EnableUDPDiscovery { hostname, _ := os.Hostname() mesh = meshdisco.New(meshdisco.Config{ Name: cfg.ServiceName, Instance: hostname, - Role: meshdisco.RoleCore, Version: Version, Group: cfg.DiscoveryGroup, Port: cfg.DiscoveryPort, Interval: time.Duration(cfg.DiscoveryIntervalMS) * time.Millisecond, - Payload: func() (string, string, *meshdisco.URLs) { + Payload: func() (string, []meshdisco.Resource) { + service := meshdisco.Resource{ + ID: "service", + Caps: []string{meshdisco.CapServicePrefix + cfg.ServiceName}, + Endpoints: map[string]string{"ui": coreBaseURL(cfg)}, + } info := leaderProvider.LeaderInfo() if info == nil { - return "", "starting", nil + return "starting", []meshdisco.Resource{service} } - return coreBaseURL(cfg), "running", &meshdisco.URLs{ + data, _ := json.Marshal(map[string]meshdisco.URLs{"urls": { Hostname: info.Hostname, BaseUrl: info.BaseUrl, ApiUrl: info.ApiUrl, WebdavUrl: info.WebdavUrl, WebdavUrlInternal: info.WebdavUrlInternal, + }}) + core := meshdisco.Resource{ + ID: "core", + Caps: []string{meshdisco.CapCore}, + Endpoints: map[string]string{"http": info.ApiUrl}, + Data: data, } + return "running", []meshdisco.Resource{core, service} }, }) - if err := mesh.Start(); err != nil { + if err := mesh.Start(); err != nil { log.Printf("[meta-core] Warning: UDP discovery disabled: %v", err) mesh = nil } diff --git a/dashboard/src/meta-service-menu.js b/dashboard/src/meta-service-menu.js index 17fa6b2..a3b59a1 100644 --- a/dashboard/src/meta-service-menu.js +++ b/dashboard/src/meta-service-menu.js @@ -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"; diff --git a/go.mod b/go.mod index 63fd171..8a3c37e 100644 --- a/go.mod +++ b/go.mod @@ -18,5 +18,5 @@ require ( github.com/cespare/xxhash/v2 v2.2.0 // indirect github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect github.com/yuin/gopher-lua v1.1.0 // indirect - golang.org/x/sys v0.28.0 // indirect + golang.org/x/sys v0.28.0 ) diff --git a/internal/api/handlers.go b/internal/api/handlers.go index fbb0f9b..a602a10 100644 --- a/internal/api/handlers.go +++ b/internal/api/handlers.go @@ -550,11 +550,11 @@ func (s *Server) handleGetFileByCID(w http.ResponseWriter, r *http.Request) { // handleListNeighbors handles GET /neighbors and GET /api/neighbors. // -// meta-discovery v1. The list is built from UDP announces held in memory, not -// from files under /meta-core/services, so it works across stacks that share a -// network but no volume. Instances of the same service are collapsed to the -// most recent, which preserves the one-row-per-service shape the old -// /api/services had. `?all=1` returns every instance, for debugging. +// Beacon v2 (docs/project-architecture/beacon-v2.md). The list is this node's +// in-memory view of the UDP group, so it works across stacks that share a +// network but no volume. Instances of one name collapse to the most recent +// (one row per service, the nav menu's shape); `?all=1` returns every +// instance, `?cap=` keeps only nodes with a matching capability. func (s *Server) handleListNeighbors(w http.ResponseWriter, r *http.Request) { response := map[string]interface{}{ "current": s.config.ServiceName, @@ -576,6 +576,15 @@ func (s *Server) handleListNeighbors(w http.ResponseWriter, r *http.Request) { } else { list = s.mesh.NeighborsByName() } + if pattern := r.URL.Query().Get("cap"); pattern != "" { + kept := list[:0] + for _, nb := range list { + if nb.Matches(pattern) { + kept = append(kept, nb) + } + } + list = kept + } response["enabled"] = true response["neighbors"] = list diff --git a/internal/api/server.go b/internal/api/server.go index 6b6a354..e3661b9 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -43,7 +43,7 @@ type Server struct { // challenges backs proof-of-possession on reveal/delete. In-process, so a // restart invalidates the ones in flight — see identity/challenge.go. challenges *identity.ChallengeStore - // mesh is the UDP discovery node (meta-discovery v1). nil when discovery + // mesh is the UDP discovery node (beacon v2). nil when discovery // is disabled or failed to bind, in which case /api/neighbors reports an // empty list rather than erroring — a dashboard nav is decoration and // must not take the API down with it. @@ -291,7 +291,7 @@ func (s *Server) setupRoutes() { s.router.HandleFunc("/api/events/files", s.handleEventsFiles).Methods("GET") s.router.HandleFunc("/api/events/meta", s.handleEventsMeta).Methods("GET") - // meta-discovery v1: neighbours heard over UDP. Supersedes /api/services, + // beacon v2: neighbours heard over UDP. Supersedes /api/services, // which is still served above while consumers migrate. s.router.HandleFunc("/neighbors", s.handleListNeighbors).Methods("GET") s.router.HandleFunc("/api/neighbors", s.handleListNeighbors).Methods("GET") diff --git a/internal/config/config.go b/internal/config/config.go index 43921b1..de72e56 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -41,13 +41,13 @@ type Config struct { // Mount configuration MountsDir string // Path to mounts configuration (default: /meta-core/mounts) - // UDP service discovery (meta-discovery v1). Replaces the file-based - // registry that used to live under /meta-core/services; see - // docs/project-architecture/service-discovery.md. - EnableUDPDiscovery bool // Announce + listen on the multicast group (default: true) - DiscoveryGroup string // IPv4 multicast group (default: 239.255.77.1) - DiscoveryPort int // UDP port (default: 9399) - DiscoveryIntervalMS int // Unsolicited announce interval in ms (default: 10000) + // UDP discovery (beacon v2). Replaces the file-based registry that used + // to live under /meta-core/services; see + // docs/project-architecture/beacon-v2.md. + EnableUDPDiscovery bool // Advertise + listen on the multicast group (default: true) + DiscoveryGroup string // BEACON_GROUP: IPv4 multicast group (default: 239.255.99.1) + DiscoveryPort int // BEACON_PORT: UDP port (default: 9099) + DiscoveryIntervalMS int // BEACON_INTERVAL_MS: unsolicited advertise interval (default: 10000) } // Load creates a Config from environment variables @@ -72,9 +72,9 @@ func Load() *Config { DebounceMS: getEnvInt("DEBOUNCE_MS", 30000), EnableFileWatcher: getEnvBool("ENABLE_FILE_WATCHER", true), EnableUDPDiscovery: getEnvBool("ENABLE_UDP_DISCOVERY", true), - DiscoveryGroup: getEnv("DISCOVERY_GROUP", "239.255.77.1"), - DiscoveryPort: getEnvInt("DISCOVERY_PORT", 9399), - DiscoveryIntervalMS: getEnvInt("DISCOVERY_INTERVAL_MS", 10000), + DiscoveryGroup: getEnv("BEACON_GROUP", "239.255.99.1"), + DiscoveryPort: getEnvInt("BEACON_PORT", 9099), + DiscoveryIntervalMS: getEnvInt("BEACON_INTERVAL_MS", 10000), } // Set mounts directory @@ -85,7 +85,7 @@ func Load() *Config { // LockFilePath returns the path to the Redis-volume mutex. Held by // docker/leader-election.sh; nothing reads it, and it is NOT service -// discovery — see docs/project-architecture/service-discovery.md. +// discovery — see docs/project-architecture/beacon-v2.md. func (c *Config) LockFilePath() string { return c.MetaCorePath + "/locks/kv-leader.lock" } diff --git a/internal/meshdisco/node.go b/internal/meshdisco/node.go index 0deecc7..e0d2a62 100644 --- a/internal/meshdisco/node.go +++ b/internal/meshdisco/node.go @@ -13,24 +13,24 @@ import ( "time" "golang.org/x/net/ipv4" + "golang.org/x/sys/unix" ) // Config configures a Node. Payload is a function rather than a value so the -// announce is rebuilt from live state on every tick — that is what keeps the -// announced URLs and GET /urls from drifting apart. +// advertise is rebuilt from live state on every tick — that is what keeps the +// advertised core URLs and GET /urls from drifting apart. type Config struct { Name string Instance string - Role string Version string Group string Port int Interval time.Duration - // Payload returns the identity/URL fields for this announce. Name, - // Instance, Role, Version and the envelope are filled in by the Node. - Payload func() (baseURL string, status string, urls *URLs) + // Payload returns this node's status ("starting" / "running") and its + // resources. The envelope and NodeInfo are filled in by the Node. + Payload func() (status string, resources []Resource) } func (c *Config) applyDefaults() { @@ -43,13 +43,17 @@ func (c *Config) applyDefaults() { if c.Interval <= 0 { c.Interval = DefaultInterval } - if c.Role == "" { - c.Role = RoleService - } } -// Node is one participant in the mesh: it announces itself, answers probes, -// and keeps a TTL map of everyone else it has heard from. +type seen struct { + node NodeInfo + resources []Resource + addr string + lastSeen time.Time +} + +// Node is one beacon v2 participant: it advertises its resources, answers +// probes, and keeps a TTL map of every other node it has heard from. type Node struct { cfg Config group net.IP @@ -60,7 +64,7 @@ type Node struct { // ifaces are the multicast-capable interfaces we joined and send on. // A container attached to two docker networks (metashare-app on `pcs` // and `metamesh-mesh`, metawatch-* on `metawatch` and `metamesh-mesh`) - // has two entries here, and MUST be announced on both — sending via the + // has two entries here, and MUST be advertised on both — sending via the // default route alone silently covers only one network. ifaces []net.Interface @@ -68,8 +72,8 @@ type Node struct { // two-step operation on one shared socket. sendMu sync.Mutex - mu sync.RWMutex - neighbors map[string]*Neighbor + mu sync.RWMutex + nodes map[string]*seen stopChan chan struct{} stopOnce sync.Once @@ -80,27 +84,26 @@ type Node struct { func New(cfg Config) *Node { cfg.applyDefaults() return &Node{ - cfg: cfg, - neighbors: make(map[string]*Neighbor), - stopChan: make(chan struct{}), + cfg: cfg, + nodes: make(map[string]*seen), + stopChan: make(chan struct{}), } } // Start opens the multicast socket, joins the group on every eligible -// interface, and begins announcing. A join failure on one interface is logged -// and skipped rather than fatal — matching the ENABLE_MDNS precedent in the -// Rust services, where a host that blocks multicast degrades instead of -// failing to boot. +// interface, and begins advertising. A join failure on one interface is logged +// and skipped rather than fatal — a host that blocks multicast degrades +// instead of failing to boot. func (n *Node) Start() error { n.group = net.ParseIP(n.cfg.Group) if n.group == nil || n.group.To4() == nil { - return fmt.Errorf("meshdisco: invalid multicast group %q", n.cfg.Group) + return fmt.Errorf("beacon: invalid multicast group %q", n.cfg.Group) } lc := net.ListenConfig{Control: reusePort} conn, err := lc.ListenPacket(context.Background(), "udp4", fmt.Sprintf(":%d", n.cfg.Port)) if err != nil { - return fmt.Errorf("meshdisco: listen on :%d: %w", n.cfg.Port, err) + return fmt.Errorf("beacon: listen on :%d: %w", n.cfg.Port, err) } n.closer = conn @@ -108,41 +111,41 @@ func (n *Node) Start() error { // TTL 1: link-local only. Discovery never routes off the local segment; // the two-host topology keeps two separate meshes by design. if err := pc.SetMulticastTTL(1); err != nil { - log.Printf("[meshdisco] Warning: could not set multicast TTL: %v", err) + log.Printf("[beacon] Warning: could not set multicast TTL: %v", err) } n.pc = pc n.ifaces = eligibleInterfaces() if len(n.ifaces) == 0 { - log.Println("[meshdisco] Warning: no multicast-capable interfaces found; discovery will be inert") + log.Println("[beacon] Warning: no multicast-capable interfaces found; discovery will be inert") } group := &net.UDPAddr{IP: n.group} joined := 0 for i := range n.ifaces { if err := pc.JoinGroup(&n.ifaces[i], group); err != nil { - log.Printf("[meshdisco] Warning: join %s on %s failed: %v", n.cfg.Group, n.ifaces[i].Name, err) + log.Printf("[beacon] Warning: join %s on %s failed: %v", n.cfg.Group, n.ifaces[i].Name, err) continue } joined++ } - log.Printf("[meshdisco] Listening on %s:%d as %s/%s (joined %d of %d interfaces)", + log.Printf("[beacon] Listening on %s:%d as %s/%s (joined %d of %d interfaces)", n.cfg.Group, n.cfg.Port, n.cfg.Name, n.cfg.Instance, joined, len(n.ifaces)) n.wg.Add(2) go n.readLoop() - go n.announceLoop() + go n.advertiseLoop() return nil } -// Stop closes the socket and waits for the loops to exit. It sends a final -// announce with status "stopping" so neighbours drop us immediately instead of -// waiting out the staleness window. +// Stop multicasts a bye — so neighbours drop us immediately instead of +// waiting out the staleness window — then closes the socket and waits for the +// loops to exit. func (n *Node) Stop() error { var err error n.stopOnce.Do(func() { - n.sendAnnounce(nil, "stopping") + n.broadcast(Message{Proto: Proto, V: Version, Type: TypeBye, Node: n.nodeInfo("")}) close(n.stopChan) if n.closer != nil { err = n.closer.Close() @@ -152,28 +155,27 @@ func (n *Node) Stop() error { return err } -// Probe multicasts a discovery message on every interface. Every listener -// replies immediately, so a caller does not have to wait out an announce -// interval. This is what makes the protocol usable on the boot path. -func (n *Node) Probe() { - n.broadcast(Message{V: ProtocolVersion, Type: TypeDiscovery}) +// Probe multicasts a probe on every interface. Every node owning a resource +// matching one of want (every node, when want is empty) replies immediately, +// so a caller does not have to wait out an interval. +func (n *Node) Probe(want ...string) { + n.broadcast(Message{Proto: Proto, V: Version, Type: TypeProbe, From: n.cfg.Instance, Want: want}) } -// Neighbors returns everyone heard from within the staleness window, sorted by -// name. Entries older than that are dropped as a side effect — expiry is a -// read-time concern, so there is no reaper goroutine and no equivalent of the -// old cleaner.go. +// Neighbors returns every node heard from within the staleness window, +// sorted by (name, instance). Expired entries are dropped as a side effect — +// expiry is a read-time concern, so there is no reaper goroutine. func (n *Node) Neighbors() []Neighbor { cutoff := time.Now().Add(-n.cfg.Interval * LivenessFactor) n.mu.Lock() - out := make([]Neighbor, 0, len(n.neighbors)) - for k, nb := range n.neighbors { - if nb.LastSeen.Before(cutoff) { - delete(n.neighbors, k) + out := make([]Neighbor, 0, len(n.nodes)) + for k, s := range n.nodes { + if s.lastSeen.Before(cutoff) { + delete(n.nodes, k) continue } - out = append(out, *nb) + out = append(out, newNeighbor(s.node, s.resources, s.addr, s.lastSeen)) } n.mu.Unlock() @@ -186,13 +188,12 @@ func (n *Node) Neighbors() []Neighbor { return out } -// NeighborsByName collapses multiple instances of one service down to the most -// recently seen, which is the shape the nav menu wants (and what the existing -// /api/services dedup contract asserts). +// NeighborsByName collapses multiple instances of one name down to the most +// recently seen — the shape the nav menu wants. func (n *Node) NeighborsByName() []Neighbor { best := make(map[string]Neighbor) for _, nb := range n.Neighbors() { - if cur, ok := best[nb.Name]; !ok || nb.LastSeen.After(cur.LastSeen) { + if cur, ok := best[nb.Name]; !ok || nb.LastSeen > cur.LastSeen { best[nb.Name] = nb } } @@ -204,10 +205,27 @@ func (n *Node) NeighborsByName() []Neighbor { return out } -// Self returns this node's own announce, so callers can render themselves in -// the menu without waiting to hear their own multicast echo back. +// Self returns this node as a row, so callers can render themselves without +// waiting to hear their own multicast echo back. func (n *Node) Self() Neighbor { - return Neighbor{Message: n.buildAnnounce("running"), LastSeen: time.Now()} + status, resources := n.payload() + return newNeighbor(*n.nodeInfo(status), resources, "", time.Now()) +} + +func (n *Node) payload() (string, []Resource) { + if n.cfg.Payload == nil { + return "running", nil + } + return n.cfg.Payload() +} + +func (n *Node) nodeInfo(status string) *NodeInfo { + return &NodeInfo{Name: n.cfg.Name, Instance: n.cfg.Instance, Version: n.cfg.Version, Status: status} +} + +func (n *Node) advertiseMsg() Message { + status, resources := n.payload() + return Message{Proto: Proto, V: Version, Type: TypeAdvertise, Node: n.nodeInfo(status), Resources: resources} } func (n *Node) readLoop() { @@ -228,32 +246,39 @@ func (n *Node) readLoop() { return default: } - log.Printf("[meshdisco] Read error: %v", err) + log.Printf("[beacon] Read error: %v", err) continue } - var msg Message - if err := json.Unmarshal(buf[:nRead], &msg); err != nil { - // Malformed or foreign traffic on the group — ignore quietly. - continue - } - if msg.V != ProtocolVersion { - continue - } - // Drop our own multicast echo. Loopback is left enabled so that a - // single-container test still sees traffic; we filter by identity - // instead of relying on the socket option. - if msg.Instance != "" && msg.Instance == n.cfg.Instance && msg.Name == n.cfg.Name { - continue + msg, ok := Parse(buf[:nRead]) + if !ok { + continue // v1 or foreign traffic on the shared group } switch msg.Type { - case TypeDiscovery: - // Unicast the reply straight back to the prober's source - // address. The kernel picks the right interface by route, and - // the prober's ephemeral source port is where it is listening. - n.sendAnnounce(src, "running") - case TypeAnnounce: + case TypeProbe: + if msg.From == n.cfg.Instance { + continue + } + ad := n.advertiseMsg() + wanted := len(msg.Want) == 0 + for _, r := range ad.Resources { + if r.MatchesAny(msg.Want) { + wanted = true + break + } + } + if wanted { + // Unicast straight back to the prober's source address; its + // ephemeral source port is where it is listening. + n.send(ad, src) + } + case TypeAdvertise, TypeBye: + // Drop our own multicast echo. Loopback stays enabled so a + // single-container test still sees traffic; we filter by identity. + if msg.Node.Instance == n.cfg.Instance { + continue + } n.record(msg, src) } } @@ -269,25 +294,21 @@ func (n *Node) record(msg Message, src net.Addr) { } } - if msg.Status == "stopping" { - n.mu.Lock() - delete(n.neighbors, msg.key()) - n.mu.Unlock() + n.mu.Lock() + defer n.mu.Unlock() + if msg.Type == TypeBye { + delete(n.nodes, msg.Node.Instance) return } - - nb := &Neighbor{Message: msg, Addr: addr, LastSeen: time.Now()} - n.mu.Lock() - n.neighbors[msg.key()] = nb - n.mu.Unlock() + n.nodes[msg.Node.Instance] = &seen{node: *msg.Node, resources: msg.Resources, addr: addr, lastSeen: time.Now()} } -func (n *Node) announceLoop() { +func (n *Node) advertiseLoop() { defer n.wg.Done() - // Announce immediately, then probe once so the map fills without waiting - // a full interval for everyone else's ticker. - n.sendAnnounce(nil, "running") + // Advertise immediately, then probe once so the map fills without + // waiting a full interval for everyone else's ticker. + n.broadcast(n.advertiseMsg()) n.Probe() ticker := time.NewTicker(n.cfg.Interval) @@ -298,51 +319,32 @@ func (n *Node) announceLoop() { case <-n.stopChan: return case <-ticker.C: - n.sendAnnounce(nil, "running") + n.broadcast(n.advertiseMsg()) } } } -func (n *Node) buildAnnounce(status string) Message { - msg := Message{ - V: ProtocolVersion, - Type: TypeAnnounce, - Name: n.cfg.Name, - Instance: n.cfg.Instance, - Role: n.cfg.Role, - Version: n.cfg.Version, - Status: status, - } - if n.cfg.Payload != nil { - baseURL, st, urls := n.cfg.Payload() - msg.BaseUrl = baseURL - if st != "" && status == "running" { - msg.Status = st - } - // Only meta-core carries a URLs block; a service announcing one - // would let any container on the network impersonate the core. - if n.cfg.Role == RoleCore { - msg.URLs = urls - } - } - return msg -} - -// sendAnnounce multicasts when dst is nil, or unicasts to dst (a probe reply). -func (n *Node) sendAnnounce(dst net.Addr, status string) { - msg := n.buildAnnounce(status) - if dst == nil { - n.broadcast(msg) - return - } +func encode(msg Message) ([]byte, bool) { body, err := json.Marshal(msg) if err != nil { + return nil, false + } + if len(body) > MaxDatagram { + log.Printf("[beacon] Warning: %d-byte datagram exceeds the fragmentation-safe %d", len(body), MaxDatagram) + } + return body, true +} + +// send unicasts msg to dst (a probe reply). +func (n *Node) send(msg Message, dst net.Addr) { + body, ok := encode(msg) + if !ok { return } n.sendMu.Lock() defer n.sendMu.Unlock() if _, err := n.pc.WriteTo(body, nil, dst); err != nil { - log.Printf("[meshdisco] Reply to %s failed: %v", dst, err) + log.Printf("[beacon] Reply to %s failed: %v", dst, err) } } @@ -353,8 +355,8 @@ func (n *Node) broadcast(msg Message) { if n.pc == nil { return } - body, err := json.Marshal(msg) - if err != nil { + body, ok := encode(msg) + if !ok { return } dst := &net.UDPAddr{IP: n.group, Port: n.cfg.Port} @@ -363,11 +365,11 @@ func (n *Node) broadcast(msg Message) { defer n.sendMu.Unlock() for i := range n.ifaces { if err := n.pc.SetMulticastInterface(&n.ifaces[i]); err != nil { - log.Printf("[meshdisco] Warning: select interface %s failed: %v", n.ifaces[i].Name, err) + log.Printf("[beacon] Warning: select interface %s failed: %v", n.ifaces[i].Name, err) continue } if _, err := n.pc.WriteTo(body, nil, dst); err != nil { - log.Printf("[meshdisco] Warning: send on %s failed: %v", n.ifaces[i].Name, err) + log.Printf("[beacon] Warning: send on %s failed: %v", n.ifaces[i].Name, err) } } } @@ -377,7 +379,7 @@ func (n *Node) broadcast(msg Message) { func eligibleInterfaces() []net.Interface { all, err := net.Interfaces() if err != nil { - log.Printf("[meshdisco] Warning: could not enumerate interfaces: %v", err) + log.Printf("[beacon] Warning: could not enumerate interfaces: %v", err) return nil } @@ -402,12 +404,15 @@ func eligibleInterfaces() []net.Interface { return out } -// reusePort sets SO_REUSEADDR so a second listener (a test, or a sidecar) can -// bind the same multicast port on the same host. +// reusePort sets SO_REUSEADDR + SO_REUSEPORT so several listeners (a test, a +// sidecar, a beacon v1 responder in the same netns) can bind 9099 on one host. func reusePort(_, _ string, c syscall.RawConn) error { var serr error err := c.Control(func(fd uintptr) { - serr = syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, syscall.SO_REUSEADDR, 1) + serr = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_REUSEADDR, 1) + if serr == nil { + serr = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_REUSEPORT, 1) + } }) if err != nil { return err diff --git a/internal/meshdisco/proto.go b/internal/meshdisco/proto.go index 5ea943f..a8bcfab 100644 --- a/internal/meshdisco/proto.go +++ b/internal/meshdisco/proto.go @@ -1,75 +1,62 @@ -// Package meshdisco implements meta-discovery v1: UDP multicast service -// discovery for the meta-* services. +// Package meshdisco implements beacon v2: local resource advertise / discover +// over UDP multicast (239.255.99.1:9099), the protocol every meta-* service +// and SDK-hosted plugin uses to find its neighbours. // -// It replaces two file-based mechanisms that both required a shared -// /meta-core volume: +// A node advertises Resources, each tagged with capability strings +// ("metamesh.core", "metamesh.service/meta-sort", "metamesh.transport/nzb@1"); +// consumers keep a live view of everyone else's and filter by capability. +// meta-core advertises "metamesh.core" carrying its /urls block, which is how +// services locate it without a shared volume. // -// - /meta-core/locks/kv-leader.info — a one-line file holding meta-core's -// API URL, read at boot to locate meta-core. -// - /meta-core/services/-.json — per-service registration files -// aggregated into the dashboard's service nav. -// -// Both answer the same question — where is meta-core, and who else is on this -// network — so one announce packet answers both. Discovery scope becomes the -// docker network the container is attached to rather than the volume it -// mounts, which is what lets per-app CasaOS stacks and the meta-watch island -// see their neighbours without sharing storage. -// -// The wire protocol is documented in -// docs/project-architecture/service-discovery.md. Keep this file and that doc -// in step — every other language implementation is written against the doc. +// The wire protocol is documented in docs/project-architecture/beacon-v2.md. +// Keep this package and that doc in step — the Rust (meta-feeder-sdk), +// TypeScript and Python implementations are written against the doc. package meshdisco -import "time" - -const ( - // ProtocolVersion is the `v` field on every message. Receivers ignore - // messages carrying a version they don't know. - ProtocolVersion = 1 - - // DefaultGroup is the IPv4 multicast group. Deliberately NOT beacon's - // 239.255.99.1 (sandbox/beacon), so the two protocols cannot cross-talk - // if they ever share a network. - DefaultGroup = "239.255.77.1" - - // DefaultPort is the UDP port. Deliberately NOT beacon's 9099, for the - // same reason. - DefaultPort = 9399 - - // DefaultInterval is how often an unsolicited announce is emitted. - DefaultInterval = 10 * time.Second - - // LivenessFactor multiplies the announce interval to get the staleness - // window: a neighbour last seen longer ago than interval*factor is - // dropped. This replaces the old heartbeat file + cleaner.go reaper. - LivenessFactor = 3 - - // ReadBufferSize is the datagram receive buffer. Beacon's reference - // implementation uses 1 KiB, which truncates a realistic payload; an - // announce carrying the full URLs block is comfortably under 8 KiB. - ReadBufferSize = 8192 - - // TypeDiscovery is a probe: "who is out there?". Sent at startup and on - // a manual refresh, answered immediately by every listener. - TypeDiscovery = "discovery" - - // TypeAnnounce is a service describing itself, sent unsolicited on a - // ticker and unicast in reply to a probe. - TypeAnnounce = "announce" - - // RoleCore marks meta-core itself — the only role that carries a URLs - // block. Consumers locating meta-core filter on this. - RoleCore = "core" - - // RoleService marks every other meta-* service. - RoleService = "service" +import ( + "encoding/json" + "strings" + "time" ) -// URLs mirrors meta-core's existing GET /urls response (api.URLsResponse) -// minus redisUrl, which the api-mediated-access lockdown retired. Embedding it -// in the announce makes the announce a drop-in for the /urls hop that -// LeaderClient performs today; /urls itself stays, since meta-search and -// meta-share still call it. +const ( + // Proto and Version form the envelope every v2 message carries. Anything + // else on the group (beacon v1, junk) is dropped silently. + Proto = "beacon" + Version = 2 + + // DefaultGroup / DefaultPort are beacon's endpoint, on purpose: one + // well-known place for every kind of local resource. v1 and v2 coexist + // because v2 never sends v1's "discovery" / "announce" types. + DefaultGroup = "239.255.99.1" + DefaultPort = 9099 + + // DefaultInterval is how often an unsolicited advertise is emitted. + DefaultInterval = 10 * time.Second + + // LivenessFactor multiplies the interval to get the staleness window. + // Expiry is evaluated at read time; there is no reaper. + LivenessFactor = 3 + + // ReadBufferSize is the receive buffer; MaxDatagram is what a sender + // stays under so nothing fragments. + ReadBufferSize = 8192 + MaxDatagram = 1400 + + TypeProbe = "probe" + TypeAdvertise = "advertise" + TypeBye = "bye" + + // CapCore is meta-core's capability; its resource carries Data.urls. + CapCore = "metamesh.core" + // CapServicePrefix + name is every service's capability. + CapServicePrefix = "metamesh.service/" + CapAnyService = "metamesh.service/*" +) + +// URLs mirrors meta-core's GET /urls response (api.URLsResponse) minus +// redisUrl. It travels as the metamesh.core resource's Data.urls, so a service +// locates meta-core from the advertise alone. type URLs struct { Hostname string `json:"hostname"` BaseUrl string `json:"baseUrl"` @@ -78,55 +65,173 @@ type URLs struct { WebdavUrlInternal string `json:"webdavUrlInternal"` } -// Message is the single wire type: both probe and announce are one JSON -// datagram. Everything but `v` and `type` is omitted on a probe. -type Message struct { - V int `json:"v"` - Type string `json:"type"` - - // Name is the service name, e.g. "meta-core", "meta-sort". - Name string `json:"name,omitempty"` - // Instance distinguishes two instances of the same Name — the container - // hostname. Also used to recognise and drop our own multicast echo. - Instance string `json:"instance,omitempty"` - // Role is RoleCore or RoleService. - Role string `json:"role,omitempty"` - // Version is the service's build version, for display only. - Version string `json:"version,omitempty"` - // Status is "running" | "starting" | "stopping". +// NodeInfo says who is advertising. +type NodeInfo struct { + Name string `json:"name"` + Instance string `json:"instance"` + Version string `json:"version,omitempty"` + // Status is "starting" | "running"; absent reads as running. Status string `json:"status,omitempty"` - - // BaseUrl is the browser-facing URL the nav menu links to. Built with - // the same three-way fallback the file registry used: - // META_CORE_PUBLIC_URL -> BASE_URL -> http://:. - BaseUrl string `json:"baseUrl,omitempty"` - - // URLs is populated only when Role == RoleCore. - URLs *URLs `json:"urls,omitempty"` - - // Token is reserved for a future HMAC over the payload. v1 senders omit - // it and v1 receivers ignore it; see the doc's Security section. - Token string `json:"token,omitempty"` } -// Neighbor is a Message plus the receive-side facts the sender cannot state -// about itself. -type Neighbor struct { - Message - // Addr is the source address the datagram arrived from. Taken from the - // packet, never from the payload — a sender does not get to claim its - // own IP. - Addr string `json:"addr"` - // LastSeen is when we last received an announce from this instance. - LastSeen time.Time `json:"lastSeen"` +// Resource is one advertised thing. +type Resource struct { + ID string `json:"id"` + Caps []string `json:"caps"` + // Endpoints: well-known names http, ui, manifest, mcp. A value starting + // with "/" is relative to endpoints.http. + Endpoints map[string]string `json:"endpoints,omitempty"` + Rev string `json:"rev,omitempty"` + // Binds names the one consumer instance this resource belongs to. + Binds string `json:"binds,omitempty"` + Data json.RawMessage `json:"data,omitempty"` } -// key identifies a neighbour entry. Two instances of the same service are -// distinct neighbours; callers that want one row per service (the nav menu) -// collapse by Name themselves. -func (m Message) key() string { - if m.Instance == "" { - return m.Name +// Matches reports whether any cap matches pattern. +func (r Resource) Matches(pattern string) bool { + for _, c := range r.Caps { + if CapMatches(pattern, c) { + return true + } } - return m.Name + "|" + m.Instance + return false +} + +// MatchesAny reports whether any cap matches any pattern; an empty list matches. +func (r Resource) MatchesAny(patterns []string) bool { + if len(patterns) == 0 { + return true + } + for _, p := range patterns { + if r.Matches(p) { + return true + } + } + return false +} + +// Endpoint returns the named endpoint as an absolute URL. +func (r Resource) Endpoint(name string) string { + v := r.Endpoints[name] + if strings.HasPrefix(v, "/") { + base := r.Endpoints["http"] + if base == "" { + return "" + } + return strings.TrimRight(base, "/") + v + } + return v +} + +// Message is every v2 datagram. +type Message struct { + Proto string `json:"proto"` + V int `json:"v"` + Type string `json:"type"` + Node *NodeInfo `json:"node,omitempty"` + Resources []Resource `json:"resources,omitempty"` + // From and Want are probe-only. + From string `json:"from,omitempty"` + Want []string `json:"want,omitempty"` +} + +// Parse decodes a datagram, returning ok=false for anything that is not a +// well-formed beacon v2 message (v1 traffic, other versions, unknown types, +// an advertise/bye without a node). +func Parse(b []byte) (Message, bool) { + var m Message + if err := json.Unmarshal(b, &m); err != nil { + return m, false + } + if m.Proto != Proto || m.V != Version { + return m, false + } + switch m.Type { + case TypeProbe: + return m, true + case TypeAdvertise, TypeBye: + if m.Node == nil || m.Node.Name == "" || m.Node.Instance == "" { + return m, false + } + return m, true + } + return m, false +} + +func splitContract(s string) (string, string, bool) { + if i := strings.LastIndex(s, "@"); i >= 0 { + return s[:i], s[i+1:], true + } + return s, "", false +} + +// CapMatches reports whether capability cap satisfies pattern: +// "*" matches everything; an "@N" on the pattern must equal the cap's +// contract (none on the pattern ignores it); "x/*" matches any variant of x; +// otherwise the bases must be equal. +func CapMatches(pattern, cap string) bool { + if pattern == "*" { + return true + } + pbase, pc, phas := splitContract(pattern) + cbase, cc, chas := splitContract(cap) + if phas && (!chas || pc != cc) { + return false + } + if prefix, ok := strings.CutSuffix(pbase, "/*"); ok { + rest, ok := strings.CutPrefix(cbase, prefix) + return ok && strings.HasPrefix(rest, "/") && len(rest) > 1 + } + return pbase == cbase +} + +// Neighbor is a node heard on the wire, as served by /api/neighbors. +type Neighbor struct { + NodeInfo + Resources []Resource `json:"resources"` + // Caps flattens every resource's caps, for consumers that just filter. + Caps []string `json:"caps"` + // BaseUrl is the metamesh.service/* resource's endpoints.ui — kept so + // v1-era readers (the nav menu) keep working. + BaseUrl string `json:"baseUrl,omitempty"` + // Addr is the packet's source address, never a payload claim. + Addr string `json:"addr"` + // LastSeen is unix seconds. + LastSeen float64 `json:"lastSeen"` +} + +// Matches reports whether any resource matches pattern. +func (n Neighbor) Matches(pattern string) bool { + for _, r := range n.Resources { + if r.Matches(pattern) { + return true + } + } + return false +} + +func newNeighbor(node NodeInfo, resources []Resource, addr string, seen time.Time) Neighbor { + nb := Neighbor{ + NodeInfo: node, + Resources: resources, + Caps: []string{}, + Addr: addr, + LastSeen: float64(seen.UnixNano()) / 1e9, + } + if nb.Resources == nil { + nb.Resources = []Resource{} + } + seenCap := map[string]bool{} + for _, r := range resources { + for _, c := range r.Caps { + if !seenCap[c] { + seenCap[c] = true + nb.Caps = append(nb.Caps, c) + } + } + if nb.BaseUrl == "" && r.Matches(CapAnyService) { + nb.BaseUrl = r.Endpoint("ui") + } + } + return nb } diff --git a/internal/meshdisco/proto_test.go b/internal/meshdisco/proto_test.go new file mode 100644 index 0000000..8799369 --- /dev/null +++ b/internal/meshdisco/proto_test.go @@ -0,0 +1,140 @@ +package meshdisco + +import ( + "encoding/json" + "net" + "testing" + "time" +) + +func TestCapMatches(t *testing.T) { + cases := []struct { + pattern, cap string + want bool + }{ + {"metamesh.transport/*", "metamesh.transport/nzb@1", true}, + {"metamesh.transport/*", "metamesh.transport/ipfs", true}, + {"metamesh.transport/*", "metamesh.transport", false}, + {"metamesh.transport/*", "metamesh.transportx/a", false}, + {"metamesh.transport/*", "metamesh.transport/", false}, + {"metamesh.transport/nzb", "metamesh.transport/nzb@2", true}, + {"metamesh.transport/nzb@2", "metamesh.transport/nzb@2", true}, + {"metamesh.transport/nzb@2", "metamesh.transport/nzb@1", false}, + {"metamesh.transport/nzb@2", "metamesh.transport/nzb", false}, + {"metamesh.transport/*@1", "metamesh.transport/nzb@1", true}, + {"metamesh.core", "metamesh.core", true}, + {"metamesh.core", "metamesh.core2", false}, + {"mcp", "mcp/x", false}, + {"*", "anything@3", true}, + } + for _, c := range cases { + if got := CapMatches(c.pattern, c.cap); got != c.want { + t.Errorf("CapMatches(%q, %q) = %v, want %v", c.pattern, c.cap, got, c.want) + } + } +} + +func TestParseRejectsV1AndForeign(t *testing.T) { + for _, raw := range []string{ + `{"type":"discovery"}`, + `{"type":"announce","name":"x","tools":[]}`, + `{"v":1,"type":"announce","name":"meta-sort"}`, + `{"proto":"beacon","v":3,"type":"probe"}`, + `{"proto":"beacon","v":2,"type":"announce"}`, + `{"proto":"beacon","v":2,"type":"advertise"}`, + "\x00garbage", + } { + if _, ok := Parse([]byte(raw)); ok { + t.Errorf("Parse(%q) accepted", raw) + } + } + if _, ok := Parse([]byte(`{"proto":"beacon","v":2,"type":"probe","want":["metamesh.core"]}`)); !ok { + t.Error("valid probe rejected") + } +} + +func TestNeighborRowCarriesCapsAndBaseUrl(t *testing.T) { + nb := newNeighbor( + NodeInfo{Name: "meta-sort", Instance: "metasort-app"}, + []Resource{{ID: "service", Caps: []string{"metamesh.service/meta-sort"}, Endpoints: map[string]string{"ui": "https://x"}}}, + "10.0.0.1", time.Now(), + ) + if nb.BaseUrl != "https://x" || len(nb.Caps) != 1 || !nb.Matches(CapAnyService) { + t.Fatalf("unexpected row: %+v", nb) + } +} + +// End to end over loopback: a probe with a non-matching want gets no reply; +// a matching one gets the advertise; advertise then bye updates the view. +func TestProbeAndRecord(t *testing.T) { + port := freePort(t) + n := New(Config{ + Name: "meta-core", Instance: "core-1", Port: port, Interval: time.Hour, + Payload: func() (string, []Resource) { + return "running", []Resource{{ID: "core", Caps: []string{CapCore}}} + }, + }) + if err := n.Start(); err != nil { + t.Fatal(err) + } + defer n.Stop() + + c, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + t.Fatal(err) + } + defer c.Close() + dst := &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: port} + buf := make([]byte, ReadBufferSize) + + send := func(m Message) { + b, _ := json.Marshal(m) + if _, err := c.WriteToUDP(b, dst); err != nil { + t.Fatal(err) + } + } + + send(Message{Proto: Proto, V: Version, Type: TypeProbe, From: "cli", Want: []string{"metamesh.service/*"}}) + c.SetReadDeadline(time.Now().Add(300 * time.Millisecond)) + if _, _, err := c.ReadFromUDP(buf); err == nil { + t.Fatal("non-matching probe got a reply") + } + + send(Message{Proto: Proto, V: Version, Type: TypeProbe, From: "cli", Want: []string{CapCore}}) + c.SetReadDeadline(time.Now().Add(2 * time.Second)) + nr, _, err := c.ReadFromUDP(buf) + if err != nil { + t.Fatalf("matching probe got no reply: %v", err) + } + if m, ok := Parse(buf[:nr]); !ok || m.Type != TypeAdvertise || m.Node.Instance != "core-1" { + t.Fatalf("bad reply: %s", buf[:nr]) + } + + peer := &NodeInfo{Name: "meta-sort", Instance: "sort-1"} + send(Message{Proto: Proto, V: Version, Type: TypeAdvertise, Node: peer, + Resources: []Resource{{ID: "service", Caps: []string{"metamesh.service/meta-sort"}}}}) + waitFor(t, func() bool { return len(n.Neighbors()) == 1 }) + send(Message{Proto: Proto, V: Version, Type: TypeBye, Node: peer}) + waitFor(t, func() bool { return len(n.Neighbors()) == 0 }) +} + +func freePort(t *testing.T) int { + c, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + t.Fatal(err) + } + defer c.Close() + return c.LocalAddr().(*net.UDPAddr).Port +} + +func waitFor(t *testing.T, cond func() bool) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(20 * time.Millisecond) + } + t.Fatal("condition not reached") +}