Compare commits

..
2 Commits
Author SHA1 Message Date
Worph 27ed45999b beacon v2 discovery: advertise metamesh.core + service resources on 239.255.99.1:9099, /api/neighbors ?cap= filter
Build and Push Docker Image / build (push) Failing after 14s
2026-10-04 11:51:03 +00:00
Worph 9254f043a4 README: sync with current code 2026-10-03 11:51:55 +00:00
10 changed files with 689 additions and 358 deletions
+144 -88
View File
@@ -1,34 +1,42 @@
# meta-core # meta-core
Standalone Go service that owns Redis, the leader-info file, the HTTP+SSE Standalone Go service that owns Redis, the HTTP+SSE metadata surface, the
metadata surface, and a WebDAV file server for the MetaMesh dev/prod stack. User Data Layer store and signing keystore, and a WebDAV file server for the
MetaMesh dev/prod stack.
## Overview ## Overview
meta-core runs as its own container (`metacore-app`). Other MetaMesh services meta-core runs as its own container (`metacore-app`). Other MetaMesh services
(meta-sort, meta-fuse, meta-stremio, meta-dup, meta-share) discover it by (meta-sort, meta-fuse, meta-stremio, meta-dup, meta-search, meta-share,
reading `/meta-core/locks/kv-leader.info` and then call its HTTP API. Redis meta-gateway, meta-watch, …) locate it over UDP multicast (beacon v2 —
is not exposed outside the container — all metadata I/O is mediated by the its `metamesh.core` advertise carries the full `/urls` payload) and then call its
HTTP API. 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)).
### Key capabilities ### Key capabilities
| Feature | Description | | Feature | Description |
|---------|-------------| |---------|-------------|
| **Leader gate** | Bash `flock` loop in `docker/leader-election.sh` only `exec`s supervisord on the winner. The Go binary therefore only ever runs as leader. | | **Redis-volume mutex** | Bash `flock` loop in `docker/leader-election.sh` only `exec`s supervisord on the winner, so two containers sharing one `/meta-core` volume can never run two Redis writers. It is a mutex, not discovery — nothing reads the lock. |
| **Redis owner** | Supervisord starts Redis with AOF + RDB persistence inside the container; meta-core connects to it on localhost. | | **Redis owner** | Supervisord starts Redis with AOF + RDB persistence inside the container; meta-core connects to it on localhost. |
| **HTTP API** | Typed REST surface (gorilla/mux) for metadata, KV browsing, snapshots, schema inference, mounts, watchers, neighbours, and file access by CID. | | **HTTP API** | Typed REST surface (gorilla/mux) for metadata, KV browsing, snapshots, schema inference, mounts, watchers, neighbours, identity, UDL, and file access by CID. |
| **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. | | **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. | | **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`. | | **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. | | **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. | | **UUID-rooted storage** | Roots are UUIDv7 (Crockford Base32, ULID layout); CIDs are reverse-index aliases. Schema-version sentinel refuses to boot against stale data. |
| **Identity keystore** | Multi-account secp256k1 signing keys under `/meta-core/identity/accounts/`, created from the dashboard's Identity menu (never auto-generated). A private key is returned only by generate, or by reveal against a proof-of-possession signature (delete needs the same proof); `/api/identity/sign` refuses challenge-domain payloads. |
| **User Data Layer** | Version-gated store for per-user signed (and, for private-tier keys, encrypted) records — My List, Continue Watching, ratings — indexed by user and CID. meta-core never inspects the crypto. See [User Data Layer](../../docs/project-architecture/user-data-layer.md). |
| **Dashboard + editor** | nginx serves a React dashboard at `/` and the standalone metadata editor at `/editor/`. |
### Design characteristics ### Design characteristics
- **Single-leader** — flock on a shared volume; no external consensus. - **Single writer** — flock on the `/meta-core` volume; no external consensus.
- **HTTP-only externally** — Redis is bound to the container, never published. - **HTTP-only externally** — Redis is never published by the compose stacks;
debug with `docker exec metacore-app redis-cli …`.
- **Schema-gated boot** — `EnsureSchemaVersion` aborts startup when the - **Schema-gated boot** — `EnsureSchemaVersion` aborts startup when the
on-disk Redis layout predates the current build (alpha clean-wipe policy; on-disk Redis layout predates the current build (alpha clean-wipe policy;
see `docs/uuid-rooted-metadata.md`). see `docs/uuid-rooted-metadata.md`).
@@ -41,34 +49,31 @@ HTTP API.
┌──────────────────────────────────────────────────────────────────────────┐ ┌──────────────────────────────────────────────────────────────────────────┐
│ metacore-app (standalone container) │ │ metacore-app (standalone container) │
│ │ │ │
│ leader-election.sh (flock /meta-core/locks/kv-leader.lock, │
│ then exec supervisord) │
│ supervisord │ │ supervisord │
│ ├─ leader-election.sh (flock /meta-core/locks/kv-leader.lock, │
│ │ writes kv-leader.info, exec supervisord) │
│ ├─ redis-server (AOF + RDB on /meta-core/db/redis) │ │ ├─ redis-server (AOF + RDB on /meta-core/db/redis) │
│ ├─ nginx (TLS termination, proxy_cache for /webdav) │ │ ├─ nginx :80 (dashboard, /editor, API proxy, │
│ ├─ rclone rcd (mount daemon, RC API) │ │ │ proxy_cache for /webdav, rclone GUI /rclone) │
│ ├─ rclone rcd (mount daemon, RC API on :5572) │
│ ├─ mount-watcher.sh (rclone mount lifecycle) │ │ ├─ mount-watcher.sh (rclone mount lifecycle) │
│ └─ meta-core (Go) ─► HTTP API + SSE + WebDAV on :9000 │ │ └─ meta-core (Go) ─► HTTP API + SSE + WebDAV on :9000 │
│ ─► UDP advertise/listen 239.255.99.1:9099 │
└──────────────────────────────────────────────────────────────────────────┘ └──────────────────────────────────────────────────────────────────────────┘
│ │
┌──────────────────────┴──────────────────────┐ ┌──────────────────────┴──────────────────────┐
│ /meta-core (shared volume) │ │ /meta-core (meta-core's own volume) │
│ locks/kv-leader.{lock,info} │ │ locks/kv-leader.lock │
│ db/redis/ │ │ db/redis/ │
│ services/*.json │ │ identity/accounts/*.json │
│ mounts/{mounts.json, errors/} │ │ mounts/{mounts.json, errors/} │
│ watchers.json │ │ watchers.json │
│ cache/ (nginx proxy_cache) │ │ cache/ (nginx proxy_cache) │
└─────────────────────────────────────────────┘ └─────────────────────────────────────────────┘
▲
┌──────────────────────────────┼────────────────────────────────┐ meta-sort / meta-fuse / meta-stremio / meta-search / meta-share / …
│ │ │ │ 1. hear the metamesh.core advertise on UDP multicast (carries /urls)
┌─────────────┐ ┌─────────────┐ ┌─────────────┐ └─ 2. HTTP + SSE (+ WebDAV) to metacore-app — never Redis, never the volume
│ meta-sort │ │ meta-fuse │ │ meta-stremio│
│ │ │ │ │ │
│ reads │ HTTP+SSE │ reads │ HTTP+SSE │ reads │
│ leader.info├──────────────►│ leader.info├───────────────►│ leader.info│
└─────────────┘ └─────────────┘ └─────────────┘
``` ```
## Quick start ## Quick start
@@ -82,8 +87,9 @@ make test # run Go unit tests
``` ```
`make docker` tags the image as `:1.0.0`, `:latest`, **and** `:local`. The `make docker` tags the image as `:1.0.0`, `:latest`, **and** `:local`. The
`:local` tag is the one every sub-stack docker-compose.yml references `:local` tag is the one the sub-stack docker-compose files reference
(meta-share 10-peer federation, meta-gateway dev stacks); without it the (meta-share federation, meta-read and meta-listen dev stacks); the meta-gateway
dev stack (`docker-compose.gateway.yml`) uses `:latest`. Without these tags the
sub-stacks can't see a freshly-built image by name. sub-stacks can't see a freshly-built image by name.
> **Source-change → recreate.** `make docker` rebuilds and re-tags the > **Source-change → recreate.** `make docker` rebuilds and re-tags the
@@ -94,12 +100,14 @@ sub-stacks can't see a freshly-built image by name.
> >
> ```bash > ```bash
> make docker > make docker
> docker compose -f packages/meta-gateway/dev/docker-compose.yml \ > docker compose -f packages/meta-gateway/dev/docker-compose.gateway.yml \
> up -d --force-recreate metagateway-generic-core > up -d --force-recreate metagateway-hub-metacore
> docker compose -f packages/meta-gateway/dev/docker-compose.torznab.yml \
> up -d --force-recreate metagateway-torznab-core
> docker compose -f packages/meta-share/dev/docker-compose.yml \ > docker compose -f packages/meta-share/dev/docker-compose.yml \
> --profile share-10node up -d --force-recreate metashare-core metashare-core-test > --profile share-10node up -d --force-recreate metashare-core metashare-core-test
> docker compose -f packages/meta-read/dev/docker-compose.yml \
> up -d --force-recreate metaread-dev-core
> docker compose -f packages/meta-listen/dev/docker-compose.yml \
> up -d --force-recreate metalisten-dev-core
> ``` > ```
> >
> The symptom of a missed recreate is missing API fields on `/urls` > The symptom of a missed recreate is missing API fields on `/urls`
@@ -109,7 +117,7 @@ sub-stacks can't see a freshly-built image by name.
> >
> ```bash > ```bash
> docker inspect meta-core:local --format '{{.Id}}' # fresh SHA > docker inspect meta-core:local --format '{{.Id}}' # fresh SHA
> docker ps --filter ancestor=meta-core:local --format '{{.Names}}' > docker ps --filter ancestor=meta-core:local --format '{{.Names}}' # repeat for :latest
> # any container whose Image SHA doesn't match the fresh SHA needs --force-recreate. > # any container whose Image SHA doesn't match the fresh SHA needs --force-recreate.
> ``` > ```
@@ -126,36 +134,39 @@ HTTP port (no auth, bypasses Caddy) is `http://localhost:18083`.
## Configuration ## Configuration
Environment variables consumed by `internal/config/config.go`: Environment variables (from `internal/config/config.go` unless noted):
| Variable | Default | Description | | Variable | Default | Description |
|---|---|---| |---|---|---|
| `META_CORE_PATH` | `/meta-core` | Shared volume root (locks, db, services, mounts, cache). | | `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. | | `FILES_PATH` | `/files` | Files volume root. |
| `SERVICE_NAME` | `meta-core` | Identity in the service registry. | | `SERVICE_NAME` | `meta-core` | Name in the beacon v2 advertise. |
| `SERVICE_VERSION` | `1.0.0` | Reported via `/status` and registry. | | `SERVICE_VERSION` | `1.0.0` | Reported via `/status`. |
| `API_PORT` | `8180` | External port baked into the leader-info `baseUrl`. | | `API_PORT` | `8180` | External port baked into the constructed `baseUrl` when `BASE_URL` is unset. |
| `BASE_URL` | _empty_ | Overrides the constructed `baseUrl` if set (used for HTTPS perimeter). | | `BASE_URL` | _empty_ | Overrides the constructed `baseUrl` (the Caddy/HTTPS perimeter URL). |
| `META_CORE_PUBLIC_URL` | _empty_ | Browser-facing nav URL in the announce; wins over `BASE_URL` (e.g. a debug-direct port with no Caddy). |
| `REDIS_PORT` | `6379` | Local Redis port. | | `REDIS_PORT` | `6379` | Local Redis port. |
| `META_CORE_HTTP_PORT` | `9000` | Go HTTP+SSE+WebDAV port. | | `META_CORE_HTTP_PORT` | `9000` | Go HTTP+SSE+WebDAV port. |
| `META_CORE_HTTP_HOST` | `0.0.0.0` | HTTP bind. | | `META_CORE_HTTP_HOST` | `0.0.0.0` | HTTP bind. |
| `HEALTH_CHECK_INTERVAL_MS` | `5000` | Internal health-loop cadence. | | `ENABLE_FILE_WATCHER` | `true` | Disable to suppress the in-process watcher entirely (`POST /api/files/register` stays available). |
| `HEARTBEAT_INTERVAL_MS` | `30000` | Service registry heartbeat. | | `ENABLE_UDP_DISCOVERY` | `true` | beacon v2 advertise + listen. |
| `STALE_THRESHOLD_MS` | `60000` | Age past which a registry entry is marked stale. | | `BEACON_GROUP` / `BEACON_PORT` | `239.255.99.1` / `9099` | Multicast group and UDP port. |
| `CLEANUP_INTERVAL_MS` | `600000` | Dead-service cleaner cadence. | | `BEACON_INTERVAL_MS` | `10000` | Unsolicited advertise interval. |
| `DEAD_SERVICE_THRESHOLD_MS` | `600000` | Age past which a stale entry is deleted. | | `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`). |
| `WATCH_INTERVAL_MS` | `1000` | Watcher poll cadence. | | `ELECTION_RETRY_SECS` | `5` | flock retry interval (read by `docker/leader-election.sh`, not the Go binary). |
| `DEBOUNCE_MS` | `30000` | File-event debounce window. |
| `ENABLE_FILE_WATCHER` | `true` | Disable to suppress the in-process watcher entirely. |
Note: there is no `META_CORE_REDIS_PORT`, no `META_CORE_SERVICE_NAME`, and `HEALTH_CHECK_INTERVAL_MS`, `HEARTBEAT_INTERVAL_MS`, `STALE_THRESHOLD_MS`,
no `META_CORE_LEADER_TIMEOUT`. The Redis URL is discovered locally; service `CLEANUP_INTERVAL_MS`, `DEAD_SERVICE_THRESHOLD_MS`, `WATCH_INTERVAL_MS` and
identity uses `SERVICE_NAME`; leader timeout is governed by the bash flock `DEBOUNCE_MS` are still parsed by `internal/config/config.go` but nothing reads
retry interval (`ELECTION_RETRY_SECS`, default `5`). them (leftovers of the file-based registry); setting them has no effect. There
is no `META_CORE_REDIS_PORT` / `META_CORE_SERVICE_NAME` /
`META_CORE_LEADER_TIMEOUT`, and no `WATCH_PATHS` — watched folders are watcher
configs created via `/api/watchers`.
## API reference ## API reference
All endpoints below are served by `internal/api/server.go`. Examples assume All endpoints below are registered in `internal/api/server.go` (mounts,
watcher and watchers routes in their own packages' `handlers.go`). Examples assume
the in-container Go port `:9000`; from the host use the in-container Go port `:9000`; from the host use
`https://metacore-dev.localhost:8083` (Caddy) or `http://localhost:18083` `https://metacore-dev.localhost:8083` (Caddy) or `http://localhost:18083`
(debug-direct). (debug-direct).
@@ -167,14 +178,14 @@ GET /health # storage + role
GET /status # version, uptime, etc. GET /status # version, uptime, etc.
GET /leader # current leader info (hostname, PID, timestamps) GET /leader # current leader info (hostname, PID, timestamps)
GET /urls # baseUrl, apiUrl, webdavUrl, webdavUrlInternal (redisUrl is intentionally empty) 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/neighbors # same, under /api
GET /api/stats # dashboard counters (see below) GET /api/stats # dashboard counters (see below)
``` ```
`/services`, `/api/services`, `/services/{name}` and `/services/cleanup/stats` `/services`, `/api/services`, `/services/{name}` and `/services/cleanup/stats`
are **gone** — the registration-and-heartbeat registry they served was replaced 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 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 than a 404; that is what made the dashboard fail with
`SyntaxError: Unexpected token '<'` until 1.0.21. `SyntaxError: Unexpected token '<'` until 1.0.21.
@@ -229,6 +240,8 @@ GET /api/meta/{cid} # resolve CID → metadata document
GET /api/file/{cid}/info # CID resolution metadata GET /api/file/{cid}/info # CID resolution metadata
GET /file/{cid} # serve file bytes (range-aware) GET /file/{cid} # serve file bytes (range-aware)
POST /file/cid # compute CID for an uploaded file POST /file/cid # compute CID for an uploaded file
POST /api/files/register # mint + alias one file under /files/plugin by midhash
# (the watcher's path for one file; works with the watcher OFF)
HEAD /data/{hash} # existence + size HEAD /data/{hash} # existence + size
GET /data/{hash}/path # absolute path on /files GET /data/{hash}/path # absolute path on /files
``` ```
@@ -280,6 +293,44 @@ signalling).
```bash ```bash
GET /api/events/files # file:events stream (watcher events) GET /api/events/files # file:events stream (watcher events)
GET /api/events/meta # meta:events stream (metadata mutations) GET /api/events/meta # meta:events stream (metadata mutations)
GET /api/events/poll # DEPRECATED long-poll over file:events — use SSE
```
### Identity (signing keystore)
Reached through meta-search's identity proxy (and directly by meta-gateway,
meta-read, meta-listen) for record signing, UDL writes and keypair sign-in.
```bash
GET /api/identity # status (never the private key)
GET /api/identity/accounts
POST /api/identity/generate # the only call that returns a new private key
POST /api/identity/import
POST /api/identity/challenge # one-shot challenge for proof-of-possession
POST /api/identity/reveal # needs proof-of-possession
DELETE /api/identity # needs proof-of-possession; reports purged UDL counts
POST /api/identity/sign
POST /api/identity/sign-batch
GET /api/identity/aead-key
```
### User Data Layer
Records arrive signed (private-tier ones encrypted) as opaque base64 CBOR;
writes are version-gated (stale → `409 stale_version`, or `accepted:false` per
cell in a batch). Redis model in `internal/storage/udl.go`.
```bash
GET /api/udl/record?uid&cid&key
PUT /api/udl/record
PUT /api/udl/records # batch
GET /api/udl/user/{uid}/key/{key}
GET /api/udl/user/{uid}/cid/{cid}
GET /api/udl/user/{uid}/all
GET /api/udl/user/{uid}/count
GET /api/udl/cid/{cid}/users
GET /api/udl/cid/{cid}/aggregate # meta-watch Aggregate wire shape
GET /api/udl/users/stats
``` ```
### Mounts ### Mounts
@@ -319,6 +370,9 @@ return a deprecation pointer at the `/api/watchers/*` routes
```bash ```bash
POST /api/admin/migrate-dual-roots # reunify stranded midhash-rooted entries POST /api/admin/migrate-dual-roots # reunify stranded midhash-rooted entries
POST /api/admin/migrate-domain-screen # domain film|tv → screen, workForm backfill (idempotent)
POST /api/admin/migrate-user-cids # backfill udl:idx:user:<uid>:cids; until run, pre-index
# profiles export as EMPTY. Returns {fixed}
POST /api/admin/sweep-literature-echoes[?apply=true] POST /api/admin/sweep-literature-echoes[?apply=true]
# strip query echoes off literature cards: anilistid on site cards, # strip query echoes off literature cards: anilistid on site cards,
# workcid/chapterNumber/volumeNumber/chapterStart/chapterEnd on any card. # workcid/chapterNumber/volumeNumber/chapterStart/chapterEnd on any card.
@@ -329,32 +383,29 @@ GET/PUT/DELETE /webdav/... # mounted on /files
No `/metrics` endpoint is exposed; observability is via container logs and No `/metrics` endpoint is exposed; observability is via container logs and
the `/health` / `/status` JSON. the `/health` / `/status` JSON.
## Leader gate ## Redis-volume mutex
Leader election is a bash flock loop in `docker/leader-election.sh`, not a `docker/leader-election.sh` is a bash flock loop, not a Go-level state
Go-level state machine. When `flock` succeeds on `kv-leader.lock`, the machine. When `flock` succeeds on `/meta-core/locks/kv-leader.lock`, the
script writes the leader API URL to `kv-leader.info` and `exec`s script `exec`s supervisord; a second container mounting the same volume
supervisord. When the process tree dies, the kernel releases the lock and retries every `ELECTION_RETRY_SECS` and stays dormant. When the process tree
any standby instance can take over. There is no follower path inside the dies, the kernel releases the lock. There is no follower path inside the Go
Go binary — there is no `OnBecomeLeader` / `OnBecomeFollower` callback API. binary.
`kv-leader.info` is a single plaintext line containing the API base URL: 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
http://10.0.1.50:9000 (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`
Other services read this file, then call `GET /urls` for full discovery. and the UDP announce share.
The dev compose runs a single `metacore-app`; the flock primitive is
retained as a correctness anchor (one process owns Redis and the
writable state under `/meta-core`), not as live failover.
## Metadata storage shape (one-line summary) ## Metadata storage shape (one-line summary)
Roots are opaque UUIDv7 strings; every property is a separate Redis Roots are opaque UUIDv7 strings; every property is a separate Redis
STRING at `file:<uuid>/<property>`; every known CID for a file is a STRING at `file:<uuid>/<property>`; every known CID is a key-set member
reverse-index entry `cid:<algorithm>:<value> → <uuid>`. Full details in `file:<uuid>/cids/<bareCid>` plus a reverse-index entry `cid:<bareCid> → <uuid>`
(the canonical CID is derived by rank on read, never stored). Full details in
[`docs/metadata-storage-structure.md`](docs/metadata-storage-structure.md) [`docs/metadata-storage-structure.md`](docs/metadata-storage-structure.md)
and design rationale in [`docs/uuid-rooted-metadata.md`](docs/uuid-rooted-metadata.md). and design rationale in [`docs/uuid-rooted-metadata.md`](docs/uuid-rooted-metadata.md).
The authoritative registry of field semantics and value formats lives in The authoritative registry of field semantics and value formats lives in
@@ -364,14 +415,15 @@ the repo-root [`METADATA_KEYS.md`](../../METADATA_KEYS.md).
| Package | Purpose | | Package | Purpose |
|---------|---------| |---------|---------|
| `cmd/meta-core` | Entry point; Redis connect + schema sentinel + service wiring. | | `cmd/meta-core` | Entry point; Redis connect + schema sentinel + UDP discovery + service wiring. |
| `internal/config` | Env-driven configuration, path helpers. | | `internal/config` | Env-driven configuration, path helpers. |
| `internal/leader` | `LeaderInfoProvider` (builds the leader info from local hostname/IP/config; no election logic). | | `internal/leader` | `LeaderInfoProvider` — builds the `/urls` payload from hostname/IP/config; no election logic. |
| `internal/storage` | Redis wrapper (`client.go`), UUIDv7 minting (`uuid.go`), CID reverse index (`cid_resolution.go`), schema sentinel (`schema_version.go`), dual-root migration (`dual_root_migration.go`). | | `internal/meshdisco` | beacon v2 node (advertise, probe, neighbour map) — the Go port of the spec. |
| `internal/cid` | CID parsing + algorithm ranking for canonical-CID selection. | | `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/discovery` | Service registry + dead-service cleaner. | | `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. |
| `internal/events` | Keyspace-notification → `meta:events` stream publisher (`meta_publisher.go`). | | `internal/events` | Keyspace-notification → `meta:events` stream publisher (`meta_publisher.go`). |
| `internal/api` | HTTP server + all handlers + SSE event endpoints. | | `internal/api` | HTTP server + all handlers + SSE event endpoints + in-memory search index + `/api/stats`. |
| `internal/schema` | Live per-field schema indexer (consumes `meta:events`). | | `internal/schema` | Live per-field schema indexer (consumes `meta:events`). |
| `internal/snapshot` | Snapshot export / import / wipe. | | `internal/snapshot` | Snapshot export / import / wipe. |
| `internal/watcher` | File system scanner, MidHash256 computation, dispatcher, state registry. | | `internal/watcher` | File system scanner, MidHash256 computation, dispatcher, state registry. |
@@ -386,10 +438,12 @@ the repo-root [`METADATA_KEYS.md`](../../METADATA_KEYS.md).
2. Build LeaderInfoProvider (no election — the bash gate already won) 2. Build LeaderInfoProvider (no election — the bash gate already won)
3. Connect to local Redis (retry up to 30× / 1s) 3. Connect to local Redis (retry up to 30× / 1s)
4. EnsureSchemaVersion — abort if the existing Redis layout is stale 4. EnsureSchemaVersion — abort if the existing Redis layout is stale
5. Start service discovery + dead-service cleaner 5. Start beacon v2 node (unless ENABLE_UDP_DISCOVERY=false)
6. Construct API server (initialises watcher, mounts, WebDAV) 6. Construct API server (identity keystore migration, watcher, watchers,
7. Start API server; on first connect, start MetaPublisher + schema Indexer mounts, WebDAV)
and republish existing metadata onto meta:events 7. Start API server + pollers + search-index refresh loop; start
MetaPublisher + schema Indexer, then warm field indexes and republish
existing metadata onto meta:events in the background
8. Wait for SIGINT/SIGTERM, shutdown in reverse order 8. Wait for SIGINT/SIGTERM, shutdown in reverse order
``` ```
@@ -400,11 +454,13 @@ the repo-root [`METADATA_KEYS.md`](../../METADATA_KEYS.md).
| `github.com/gorilla/mux` | HTTP router | | `github.com/gorilla/mux` | HTTP router |
| `github.com/redis/go-redis/v9` | Redis client | | `github.com/redis/go-redis/v9` | Redis client |
| `github.com/google/uuid` | UUIDv7 minting | | `github.com/google/uuid` | UUIDv7 minting |
| `github.com/decred/dcrd/dcrec/secp256k1/v4`, `golang.org/x/crypto`, `github.com/mr-tron/base58` | Identity keys, signing, AEAD key derivation |
| `golang.org/x/net` | WebDAV support | | `golang.org/x/net` | WebDAV support |
| Go stdlib | context, net/http, os, sync, syscall | | `github.com/alicebob/miniredis/v2` | In-memory Redis for tests |
Runtime requirements: Go 1.21+ at build time; the container ships Redis, Runtime requirements: Go 1.21+ at build time; the container ships Redis,
rclone, nginx, and supervisord alongside the binary. rclone, nginx, and supervisord alongside the binary, plus the React dashboard
and metadata editor (built from `dashboard/` and `editor/` in the Dockerfile).
## License ## License
+20 -9
View File
@@ -1,6 +1,7 @@
package main package main
import ( import (
"encoding/json"
"fmt" "fmt"
"log" "log"
"os" "os"
@@ -60,34 +61,44 @@ func main() {
} }
log.Printf("[meta-core] Redis schema version OK (v%d)", storage.SchemaVersion) log.Printf("[meta-core] Redis schema version OK (v%d)", storage.SchemaVersion)
// UDP service discovery (meta-discovery v1). Runs alongside the file // Beacon v2 (docs/project-architecture/beacon-v2.md): advertise meta-core
// registry above for now — nothing consumes the wire yet, so this cannot // as metamesh.core — carrying the /urls block services locate us by — plus
// regress the existing path. The announce payload is rebuilt from // its own metamesh.service/<name> nav entry. The payload is rebuilt from
// leaderProvider on every tick, which is what keeps it and GET /urls from // leaderProvider on every tick, which keeps it and GET /urls from drifting.
// drifting apart.
var mesh *meshdisco.Node var mesh *meshdisco.Node
if cfg.EnableUDPDiscovery { if cfg.EnableUDPDiscovery {
hostname, _ := os.Hostname() hostname, _ := os.Hostname()
mesh = meshdisco.New(meshdisco.Config{ mesh = meshdisco.New(meshdisco.Config{
Name: cfg.ServiceName, Name: cfg.ServiceName,
Instance: hostname, Instance: hostname,
Role: meshdisco.RoleCore,
Version: Version, Version: Version,
Group: cfg.DiscoveryGroup, Group: cfg.DiscoveryGroup,
Port: cfg.DiscoveryPort, Port: cfg.DiscoveryPort,
Interval: time.Duration(cfg.DiscoveryIntervalMS) * time.Millisecond, 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() info := leaderProvider.LeaderInfo()
if info == nil { 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, Hostname: info.Hostname,
BaseUrl: info.BaseUrl, BaseUrl: info.BaseUrl,
ApiUrl: info.ApiUrl, ApiUrl: info.ApiUrl,
WebdavUrl: info.WebdavUrl, WebdavUrl: info.WebdavUrl,
WebdavUrlInternal: info.WebdavUrlInternal, 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 {
+7 -2
View File
@@ -238,6 +238,10 @@ class MetaServiceMenu extends HTMLElement {
for (const n of this._neighbors) { for (const n of this._neighbors) {
const name = n.name || ""; const name = n.name || "";
if (!name || seen.has(name)) continue; 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); seen.add(name);
rows.push(n); rows.push(n);
} }
@@ -275,7 +279,7 @@ class MetaServiceMenu extends HTMLElement {
el.href = url; el.href = url;
el.setAttribute("role", "menuitem"); 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"); const icon = document.createElement("span");
icon.className = "icon"; icon.className = "icon";
@@ -285,7 +289,8 @@ class MetaServiceMenu extends HTMLElement {
label.textContent = titleCase(n.name); label.textContent = titleCase(n.name);
el.append(icon, label); 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"); const dot = document.createElement("span");
dot.className = "dot"; dot.className = "dot";
dot.title = n.status || "running"; dot.title = n.status || "running";
+1 -1
View File
@@ -18,5 +18,5 @@ require (
github.com/cespare/xxhash/v2 v2.2.0 // indirect github.com/cespare/xxhash/v2 v2.2.0 // indirect
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
github.com/yuin/gopher-lua v1.1.0 // 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
) )
+14 -5
View File
@@ -550,11 +550,11 @@ func (s *Server) handleGetFileByCID(w http.ResponseWriter, r *http.Request) {
// handleListNeighbors handles GET /neighbors and GET /api/neighbors. // handleListNeighbors handles GET /neighbors and GET /api/neighbors.
// //
// meta-discovery v1. The list is built from UDP announces held in memory, not // Beacon v2 (docs/project-architecture/beacon-v2.md). The list is this node's
// from files under /meta-core/services, so it works across stacks that share a // in-memory view of the UDP group, so it works across stacks that share a
// network but no volume. Instances of the same service are collapsed to the // network but no volume. Instances of one name collapse to the most recent
// most recent, which preserves the one-row-per-service shape the old // (one row per service, the nav menu's shape); `?all=1` returns every
// /api/services had. `?all=1` returns every instance, for debugging. // instance, `?cap=<pattern>` keeps only nodes with a matching capability.
func (s *Server) handleListNeighbors(w http.ResponseWriter, r *http.Request) { func (s *Server) handleListNeighbors(w http.ResponseWriter, r *http.Request) {
response := map[string]interface{}{ response := map[string]interface{}{
"current": s.config.ServiceName, "current": s.config.ServiceName,
@@ -576,6 +576,15 @@ func (s *Server) handleListNeighbors(w http.ResponseWriter, r *http.Request) {
} else { } else {
list = s.mesh.NeighborsByName() 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["enabled"] = true
response["neighbors"] = list response["neighbors"] = list
+2 -2
View File
@@ -43,7 +43,7 @@ type Server struct {
// challenges backs proof-of-possession on reveal/delete. In-process, so a // challenges backs proof-of-possession on reveal/delete. In-process, so a
// restart invalidates the ones in flight — see identity/challenge.go. // restart invalidates the ones in flight — see identity/challenge.go.
challenges *identity.ChallengeStore 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 // is disabled or failed to bind, in which case /api/neighbors reports an
// empty list rather than erroring — a dashboard nav is decoration and // empty list rather than erroring — a dashboard nav is decoration and
// must not take the API down with it. // 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/files", s.handleEventsFiles).Methods("GET")
s.router.HandleFunc("/api/events/meta", s.handleEventsMeta).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. // which is still served above while consumers migrate.
s.router.HandleFunc("/neighbors", s.handleListNeighbors).Methods("GET") s.router.HandleFunc("/neighbors", s.handleListNeighbors).Methods("GET")
s.router.HandleFunc("/api/neighbors", s.handleListNeighbors).Methods("GET") s.router.HandleFunc("/api/neighbors", s.handleListNeighbors).Methods("GET")
+11 -11
View File
@@ -41,13 +41,13 @@ type Config struct {
// Mount configuration // Mount configuration
MountsDir string // Path to mounts configuration (default: /meta-core/mounts) MountsDir string // Path to mounts configuration (default: /meta-core/mounts)
// UDP service discovery (meta-discovery v1). Replaces the file-based // UDP discovery (beacon v2). Replaces the file-based registry that used
// registry that used to live under /meta-core/services; see // to live under /meta-core/services; see
// docs/project-architecture/service-discovery.md. // docs/project-architecture/beacon-v2.md.
EnableUDPDiscovery bool // Announce + listen on the multicast group (default: true) EnableUDPDiscovery bool // Advertise + listen on the multicast group (default: true)
DiscoveryGroup string // IPv4 multicast group (default: 239.255.77.1) DiscoveryGroup string // BEACON_GROUP: IPv4 multicast group (default: 239.255.99.1)
DiscoveryPort int // UDP port (default: 9399) DiscoveryPort int // BEACON_PORT: UDP port (default: 9099)
DiscoveryIntervalMS int // Unsolicited announce interval in ms (default: 10000) DiscoveryIntervalMS int // BEACON_INTERVAL_MS: unsolicited advertise interval (default: 10000)
} }
// Load creates a Config from environment variables // Load creates a Config from environment variables
@@ -72,9 +72,9 @@ func Load() *Config {
DebounceMS: getEnvInt("DEBOUNCE_MS", 30000), DebounceMS: getEnvInt("DEBOUNCE_MS", 30000),
EnableFileWatcher: getEnvBool("ENABLE_FILE_WATCHER", true), EnableFileWatcher: getEnvBool("ENABLE_FILE_WATCHER", true),
EnableUDPDiscovery: getEnvBool("ENABLE_UDP_DISCOVERY", true), EnableUDPDiscovery: getEnvBool("ENABLE_UDP_DISCOVERY", true),
DiscoveryGroup: getEnv("DISCOVERY_GROUP", "239.255.77.1"), DiscoveryGroup: getEnv("BEACON_GROUP", "239.255.99.1"),
DiscoveryPort: getEnvInt("DISCOVERY_PORT", 9399), DiscoveryPort: getEnvInt("BEACON_PORT", 9099),
DiscoveryIntervalMS: getEnvInt("DISCOVERY_INTERVAL_MS", 10000), DiscoveryIntervalMS: getEnvInt("BEACON_INTERVAL_MS", 10000),
} }
// Set mounts directory // Set mounts directory
@@ -85,7 +85,7 @@ func Load() *Config {
// LockFilePath returns the path to the Redis-volume mutex. Held by // LockFilePath returns the path to the Redis-volume mutex. Held by
// docker/leader-election.sh; nothing reads it, and it is NOT service // 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 { func (c *Config) LockFilePath() string {
return c.MetaCorePath + "/locks/kv-leader.lock" return c.MetaCorePath + "/locks/kv-leader.lock"
} }
+129 -124
View File
@@ -13,24 +13,24 @@ import (
"time" "time"
"golang.org/x/net/ipv4" "golang.org/x/net/ipv4"
"golang.org/x/sys/unix"
) )
// Config configures a Node. Payload is a function rather than a value so the // 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 // advertise is rebuilt from live state on every tick — that is what keeps the
// announced URLs and GET /urls from drifting apart. // advertised core URLs and GET /urls from drifting apart.
type Config struct { type Config struct {
Name string Name string
Instance string Instance string
Role string
Version string Version string
Group string Group string
Port int Port int
Interval time.Duration Interval time.Duration
// Payload returns the identity/URL fields for this announce. Name, // Payload returns this node's status ("starting" / "running") and its
// Instance, Role, Version and the envelope are filled in by the Node. // resources. The envelope and NodeInfo are filled in by the Node.
Payload func() (baseURL string, status string, urls *URLs) Payload func() (status string, resources []Resource)
} }
func (c *Config) applyDefaults() { func (c *Config) applyDefaults() {
@@ -43,13 +43,17 @@ func (c *Config) applyDefaults() {
if c.Interval <= 0 { if c.Interval <= 0 {
c.Interval = DefaultInterval c.Interval = DefaultInterval
} }
if c.Role == "" {
c.Role = RoleService
}
} }
// Node is one participant in the mesh: it announces itself, answers probes, type seen struct {
// and keeps a TTL map of everyone else it has heard from. 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 { type Node struct {
cfg Config cfg Config
group net.IP group net.IP
@@ -60,7 +64,7 @@ type Node struct {
// ifaces are the multicast-capable interfaces we joined and send on. // ifaces are the multicast-capable interfaces we joined and send on.
// A container attached to two docker networks (metashare-app on `pcs` // A container attached to two docker networks (metashare-app on `pcs`
// and `metamesh-mesh`, metawatch-* on `metawatch` and `metamesh-mesh`) // 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. // default route alone silently covers only one network.
ifaces []net.Interface ifaces []net.Interface
@@ -69,7 +73,7 @@ type Node struct {
sendMu sync.Mutex sendMu sync.Mutex
mu sync.RWMutex mu sync.RWMutex
neighbors map[string]*Neighbor nodes map[string]*seen
stopChan chan struct{} stopChan chan struct{}
stopOnce sync.Once stopOnce sync.Once
@@ -81,26 +85,25 @@ func New(cfg Config) *Node {
cfg.applyDefaults() cfg.applyDefaults()
return &Node{ return &Node{
cfg: cfg, cfg: cfg,
neighbors: make(map[string]*Neighbor), nodes: make(map[string]*seen),
stopChan: make(chan struct{}), stopChan: make(chan struct{}),
} }
} }
// Start opens the multicast socket, joins the group on every eligible // Start opens the multicast socket, joins the group on every eligible
// interface, and begins announcing. A join failure on one interface is logged // interface, and begins advertising. A join failure on one interface is logged
// and skipped rather than fatal — matching the ENABLE_MDNS precedent in the // and skipped rather than fatal — a host that blocks multicast degrades
// Rust services, where a host that blocks multicast degrades instead of // instead of failing to boot.
// failing to boot.
func (n *Node) Start() error { func (n *Node) Start() error {
n.group = net.ParseIP(n.cfg.Group) n.group = net.ParseIP(n.cfg.Group)
if n.group == nil || n.group.To4() == nil { 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} lc := net.ListenConfig{Control: reusePort}
conn, err := lc.ListenPacket(context.Background(), "udp4", fmt.Sprintf(":%d", n.cfg.Port)) conn, err := lc.ListenPacket(context.Background(), "udp4", fmt.Sprintf(":%d", n.cfg.Port))
if err != nil { 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 n.closer = conn
@@ -108,41 +111,41 @@ func (n *Node) Start() error {
// TTL 1: link-local only. Discovery never routes off the local segment; // TTL 1: link-local only. Discovery never routes off the local segment;
// the two-host topology keeps two separate meshes by design. // the two-host topology keeps two separate meshes by design.
if err := pc.SetMulticastTTL(1); err != nil { 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.pc = pc
n.ifaces = eligibleInterfaces() n.ifaces = eligibleInterfaces()
if len(n.ifaces) == 0 { 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} group := &net.UDPAddr{IP: n.group}
joined := 0 joined := 0
for i := range n.ifaces { for i := range n.ifaces {
if err := pc.JoinGroup(&n.ifaces[i], group); err != nil { 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 continue
} }
joined++ 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.cfg.Group, n.cfg.Port, n.cfg.Name, n.cfg.Instance, joined, len(n.ifaces))
n.wg.Add(2) n.wg.Add(2)
go n.readLoop() go n.readLoop()
go n.announceLoop() go n.advertiseLoop()
return nil return nil
} }
// Stop closes the socket and waits for the loops to exit. It sends a final // Stop multicasts a bye — so neighbours drop us immediately instead of
// announce with status "stopping" so neighbours drop us immediately instead of // waiting out the staleness window — then closes the socket and waits for the
// waiting out the staleness window. // loops to exit.
func (n *Node) Stop() error { func (n *Node) Stop() error {
var err error var err error
n.stopOnce.Do(func() { n.stopOnce.Do(func() {
n.sendAnnounce(nil, "stopping") n.broadcast(Message{Proto: Proto, V: Version, Type: TypeBye, Node: n.nodeInfo("")})
close(n.stopChan) close(n.stopChan)
if n.closer != nil { if n.closer != nil {
err = n.closer.Close() err = n.closer.Close()
@@ -152,28 +155,27 @@ func (n *Node) Stop() error {
return err return err
} }
// Probe multicasts a discovery message on every interface. Every listener // Probe multicasts a probe on every interface. Every node owning a resource
// replies immediately, so a caller does not have to wait out an announce // matching one of want (every node, when want is empty) replies immediately,
// interval. This is what makes the protocol usable on the boot path. // so a caller does not have to wait out an interval.
func (n *Node) Probe() { func (n *Node) Probe(want ...string) {
n.broadcast(Message{V: ProtocolVersion, Type: TypeDiscovery}) 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 // Neighbors returns every node heard from within the staleness window,
// name. Entries older than that are dropped as a side effect — expiry is a // sorted by (name, instance). Expired entries are dropped as a side effect —
// read-time concern, so there is no reaper goroutine and no equivalent of the // expiry is a read-time concern, so there is no reaper goroutine.
// old cleaner.go.
func (n *Node) Neighbors() []Neighbor { func (n *Node) Neighbors() []Neighbor {
cutoff := time.Now().Add(-n.cfg.Interval * LivenessFactor) cutoff := time.Now().Add(-n.cfg.Interval * LivenessFactor)
n.mu.Lock() n.mu.Lock()
out := make([]Neighbor, 0, len(n.neighbors)) out := make([]Neighbor, 0, len(n.nodes))
for k, nb := range n.neighbors { for k, s := range n.nodes {
if nb.LastSeen.Before(cutoff) { if s.lastSeen.Before(cutoff) {
delete(n.neighbors, k) delete(n.nodes, k)
continue continue
} }
out = append(out, *nb) out = append(out, newNeighbor(s.node, s.resources, s.addr, s.lastSeen))
} }
n.mu.Unlock() n.mu.Unlock()
@@ -186,13 +188,12 @@ func (n *Node) Neighbors() []Neighbor {
return out return out
} }
// NeighborsByName collapses multiple instances of one service down to the most // NeighborsByName collapses multiple instances of one name down to the most
// recently seen, which is the shape the nav menu wants (and what the existing // recently seen — the shape the nav menu wants.
// /api/services dedup contract asserts).
func (n *Node) NeighborsByName() []Neighbor { func (n *Node) NeighborsByName() []Neighbor {
best := make(map[string]Neighbor) best := make(map[string]Neighbor)
for _, nb := range n.Neighbors() { 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 best[nb.Name] = nb
} }
} }
@@ -204,10 +205,27 @@ func (n *Node) NeighborsByName() []Neighbor {
return out return out
} }
// Self returns this node's own announce, so callers can render themselves in // Self returns this node as a row, so callers can render themselves without
// the menu without waiting to hear their own multicast echo back. // waiting to hear their own multicast echo back.
func (n *Node) Self() Neighbor { 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() { func (n *Node) readLoop() {
@@ -228,32 +246,39 @@ func (n *Node) readLoop() {
return return
default: default:
} }
log.Printf("[meshdisco] Read error: %v", err) log.Printf("[beacon] Read error: %v", err)
continue continue
} }
var msg Message msg, ok := Parse(buf[:nRead])
if err := json.Unmarshal(buf[:nRead], &msg); err != nil { if !ok {
// Malformed or foreign traffic on the group — ignore quietly. continue // v1 or foreign traffic on the shared group
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
} }
switch msg.Type { switch msg.Type {
case TypeDiscovery: case TypeProbe:
// Unicast the reply straight back to the prober's source if msg.From == n.cfg.Instance {
// address. The kernel picks the right interface by route, and continue
// the prober's ephemeral source port is where it is listening. }
n.sendAnnounce(src, "running") ad := n.advertiseMsg()
case TypeAnnounce: 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) n.record(msg, src)
} }
} }
@@ -269,25 +294,21 @@ func (n *Node) record(msg Message, src net.Addr) {
} }
} }
if msg.Status == "stopping" {
n.mu.Lock() n.mu.Lock()
delete(n.neighbors, msg.key()) defer n.mu.Unlock()
n.mu.Unlock() if msg.Type == TypeBye {
delete(n.nodes, msg.Node.Instance)
return return
} }
n.nodes[msg.Node.Instance] = &seen{node: *msg.Node, resources: msg.Resources, addr: addr, lastSeen: time.Now()}
nb := &Neighbor{Message: msg, Addr: addr, LastSeen: time.Now()}
n.mu.Lock()
n.neighbors[msg.key()] = nb
n.mu.Unlock()
} }
func (n *Node) announceLoop() { func (n *Node) advertiseLoop() {
defer n.wg.Done() defer n.wg.Done()
// Announce immediately, then probe once so the map fills without waiting // Advertise immediately, then probe once so the map fills without
// a full interval for everyone else's ticker. // waiting a full interval for everyone else's ticker.
n.sendAnnounce(nil, "running") n.broadcast(n.advertiseMsg())
n.Probe() n.Probe()
ticker := time.NewTicker(n.cfg.Interval) ticker := time.NewTicker(n.cfg.Interval)
@@ -298,51 +319,32 @@ func (n *Node) announceLoop() {
case <-n.stopChan: case <-n.stopChan:
return return
case <-ticker.C: case <-ticker.C:
n.sendAnnounce(nil, "running") n.broadcast(n.advertiseMsg())
} }
} }
} }
func (n *Node) buildAnnounce(status string) Message { func encode(msg Message) ([]byte, bool) {
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
}
body, err := json.Marshal(msg) body, err := json.Marshal(msg)
if err != nil { 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 return
} }
n.sendMu.Lock() n.sendMu.Lock()
defer n.sendMu.Unlock() defer n.sendMu.Unlock()
if _, err := n.pc.WriteTo(body, nil, dst); err != nil { 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 { if n.pc == nil {
return return
} }
body, err := json.Marshal(msg) body, ok := encode(msg)
if err != nil { if !ok {
return return
} }
dst := &net.UDPAddr{IP: n.group, Port: n.cfg.Port} dst := &net.UDPAddr{IP: n.group, Port: n.cfg.Port}
@@ -363,11 +365,11 @@ func (n *Node) broadcast(msg Message) {
defer n.sendMu.Unlock() defer n.sendMu.Unlock()
for i := range n.ifaces { for i := range n.ifaces {
if err := n.pc.SetMulticastInterface(&n.ifaces[i]); err != nil { 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 continue
} }
if _, err := n.pc.WriteTo(body, nil, dst); err != nil { 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 { func eligibleInterfaces() []net.Interface {
all, err := net.Interfaces() all, err := net.Interfaces()
if err != nil { 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 return nil
} }
@@ -402,12 +404,15 @@ func eligibleInterfaces() []net.Interface {
return out return out
} }
// reusePort sets SO_REUSEADDR so a second listener (a test, or a sidecar) can // reusePort sets SO_REUSEADDR + SO_REUSEPORT so several listeners (a test, a
// bind the same multicast port on the same host. // sidecar, a beacon v1 responder in the same netns) can bind 9099 on one host.
func reusePort(_, _ string, c syscall.RawConn) error { func reusePort(_, _ string, c syscall.RawConn) error {
var serr error var serr error
err := c.Control(func(fd uintptr) { 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 { if err != nil {
return err return err
+212 -107
View File
@@ -1,75 +1,62 @@
// Package meshdisco implements meta-discovery v1: UDP multicast service // Package meshdisco implements beacon v2: local resource advertise / discover
// discovery for the meta-* services. // 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 // A node advertises Resources, each tagged with capability strings
// /meta-core volume: // ("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 // The wire protocol is documented in docs/project-architecture/beacon-v2.md.
// API URL, read at boot to locate meta-core. // Keep this package and that doc in step — the Rust (meta-feeder-sdk),
// - /meta-core/services/<name>-<host>.json — per-service registration files // TypeScript and Python implementations are written against the doc.
// 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.
package meshdisco package meshdisco
import "time" import (
"encoding/json"
const ( "strings"
// ProtocolVersion is the `v` field on every message. Receivers ignore "time"
// 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"
) )
// URLs mirrors meta-core's existing GET /urls response (api.URLsResponse) const (
// minus redisUrl, which the api-mediated-access lockdown retired. Embedding it // Proto and Version form the envelope every v2 message carries. Anything
// in the announce makes the announce a drop-in for the /urls hop that // else on the group (beacon v1, junk) is dropped silently.
// LeaderClient performs today; /urls itself stays, since meta-search and Proto = "beacon"
// meta-share still call it. 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 { type URLs struct {
Hostname string `json:"hostname"` Hostname string `json:"hostname"`
BaseUrl string `json:"baseUrl"` BaseUrl string `json:"baseUrl"`
@@ -78,55 +65,173 @@ type URLs struct {
WebdavUrlInternal string `json:"webdavUrlInternal"` WebdavUrlInternal string `json:"webdavUrlInternal"`
} }
// Message is the single wire type: both probe and announce are one JSON // NodeInfo says who is advertising.
// datagram. Everything but `v` and `type` is omitted on a probe. 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"`
}
// 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"`
}
// 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 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 { type Message struct {
Proto string `json:"proto"`
V int `json:"v"` V int `json:"v"`
Type string `json:"type"` Type string `json:"type"`
Node *NodeInfo `json:"node,omitempty"`
// Name is the service name, e.g. "meta-core", "meta-sort". Resources []Resource `json:"resources,omitempty"`
Name string `json:"name,omitempty"` // From and Want are probe-only.
// Instance distinguishes two instances of the same Name — the container From string `json:"from,omitempty"`
// hostname. Also used to recognise and drop our own multicast echo. Want []string `json:"want,omitempty"`
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".
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://<local-ip>:<port>.
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 // Parse decodes a datagram, returning ok=false for anything that is not a
// about itself. // 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 { type Neighbor struct {
Message NodeInfo
// Addr is the source address the datagram arrived from. Taken from the Resources []Resource `json:"resources"`
// packet, never from the payload — a sender does not get to claim its // Caps flattens every resource's caps, for consumers that just filter.
// own IP. 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"` Addr string `json:"addr"`
// LastSeen is when we last received an announce from this instance. // LastSeen is unix seconds.
LastSeen time.Time `json:"lastSeen"` LastSeen float64 `json:"lastSeen"`
} }
// key identifies a neighbour entry. Two instances of the same service are // Matches reports whether any resource matches pattern.
// distinct neighbours; callers that want one row per service (the nav menu) func (n Neighbor) Matches(pattern string) bool {
// collapse by Name themselves. for _, r := range n.Resources {
func (m Message) key() string { if r.Matches(pattern) {
if m.Instance == "" { return true
return m.Name
} }
return m.Name + "|" + m.Instance }
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
} }
+140
View File
@@ -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")
}