From 355c666183b45f769726e7b89370df28a496d8e5 Mon Sep 17 00:00:00 2001 From: Worph Date: Sun, 13 Sep 2026 19:48:52 +0200 Subject: [PATCH] chnage service discovery and identity --- cmd/meta-core/main.go | 60 ++++ dashboard/src/meta-service-menu.js | 303 +++++++++++++++++++ docker/nginx.conf | 8 + go.mod | 1 + go.sum | 2 + internal/api/admin_migrate.go | 30 ++ internal/api/handlers.go | 59 +++- internal/api/server.go | 24 +- internal/api/udl.go | 70 +++++ internal/config/config.go | 70 +++-- internal/meshdisco/node.go | 434 ++++++++++++++++++++++++++++ internal/meshdisco/proto.go | 132 +++++++++ internal/storage/udl.go | 181 +++++++++++- internal/storage/udl_admin.go | 8 +- internal/storage/udl_export_test.go | 234 +++++++++++++++ 15 files changed, 1571 insertions(+), 45 deletions(-) create mode 100644 dashboard/src/meta-service-menu.js create mode 100644 internal/meshdisco/node.go create mode 100644 internal/meshdisco/proto.go create mode 100644 internal/storage/udl_export_test.go diff --git a/cmd/meta-core/main.go b/cmd/meta-core/main.go index 0079876..f3e1b88 100644 --- a/cmd/meta-core/main.go +++ b/cmd/meta-core/main.go @@ -12,6 +12,7 @@ import ( "github.com/metazla/meta-core/internal/config" "github.com/metazla/meta-core/internal/discovery" "github.com/metazla/meta-core/internal/leader" + "github.com/metazla/meta-core/internal/meshdisco" "github.com/metazla/meta-core/internal/storage" ) @@ -72,8 +73,47 @@ func main() { log.Printf("[meta-core] Warning: failed to start service cleaner: %v", err) } + // 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. + 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) { + info := leaderProvider.LeaderInfo() + if info == nil { + return "", "starting", nil + } + return coreBaseURL(cfg), "running", &meshdisco.URLs{ + Hostname: info.Hostname, + BaseUrl: info.BaseUrl, + ApiUrl: info.ApiUrl, + WebdavUrl: info.WebdavUrl, + WebdavUrlInternal: info.WebdavUrlInternal, + } + }, + }) + if err := mesh.Start(); err != nil { + log.Printf("[meta-core] Warning: UDP discovery disabled: %v", err) + mesh = nil + } + } else { + log.Println("[meta-core] UDP discovery disabled by ENABLE_UDP_DISCOVERY") + } + // Create and start API server apiServer := api.NewServer(cfg, leaderProvider, disc, cleaner, storageClient) + apiServer.SetMeshNode(mesh) if err := apiServer.Start(); err != nil { log.Fatalf("[meta-core] Failed to start API server: %v", err) } @@ -90,6 +130,12 @@ func main() { log.Printf("[meta-core] Error stopping API server: %v", err) } + if mesh != nil { + if err := mesh.Stop(); err != nil { + log.Printf("[meta-core] Error stopping UDP discovery: %v", err) + } + } + if err := cleaner.Stop(); err != nil { log.Printf("[meta-core] Error stopping service cleaner: %v", err) } @@ -112,3 +158,17 @@ func waitForShutdown() { <-sigChan } +// coreBaseURL is the browser-facing URL announced for the nav menu. It keeps +// the three-way fallback the file registry used (discovery/service.go): +// META_CORE_PUBLIC_URL wins (the reachable URL when there is no Caddy +// perimeter, e.g. a debug-direct port), then BASE_URL (the Caddy URL), then +// the container IP. +func coreBaseURL(cfg *config.Config) string { + if cfg.PublicURL != "" { + return cfg.PublicURL + } + if cfg.BaseURL != "" { + return cfg.BaseURL + } + return fmt.Sprintf("http://%s:%d", meshdisco.LocalIPv4(), cfg.APIPort) +} diff --git a/dashboard/src/meta-service-menu.js b/dashboard/src/meta-service-menu.js new file mode 100644 index 0000000..17fa6b2 --- /dev/null +++ b/dashboard/src/meta-service-menu.js @@ -0,0 +1,303 @@ +/** + * — the neighbouring meta-* services, as a burger menu. + * + * Replaces the eight divergent "services bar" implementations (four copies of + * a React `ServiceNav.tsx`, two mirrored `services.js`, two bespoke inline + * ones) with a single framework-free custom element. It drops into the React + * dashboards as a plain tag — no React dependency — and into the vanilla UIs + * as an ES module. + * + * Data comes from the host service's own `GET /api/neighbors`, same-origin. + * Never cross-origin: the dev stack's self-signed certs and the OIDC redirects + * make a fetch to a neighbour a dead end. Links are plain navigations. + * + * ⚠ MIRRORED FILE. Byte-identical copies live in every meta-* UI tree; see + * `scripts/check-mirrors.sh`. Edit one → edit all. + * + * Shadow DOM is not optional here. The eight UIs use four incompatible token + * vocabularies, and two of them define *different colours under the same + * names* (`--bg-secondary` is #12121a in meta-sort, #16213e in meta-core, + * #161b22 in meta-dup). meta-search and meta-share are also the only + * light-mode-capable UIs, so a hardcoded dark palette would break them. + * The element therefore styles itself from four tokens with dark defaults: + * + * --mm-nav-fg, --mm-nav-bg, --mm-nav-border, --mm-nav-accent + * + * Each host maps its own variables onto those, e.g. + * meta-service-menu { --mm-nav-bg: var(--bg-secondary); } + * + * Attributes: + * current name of the host service, so it can be marked and not linked + * endpoint override the default "/api/neighbors" + * label button tooltip (default "Services") + * + * Property: + * rewriteUrl optional (name, url, neighbour) => url, for hosts that must + * translate internal hostnames into browser-reachable ones + * (meta-gateway does this for its dev/prod split). + */ + +const ICONS = { + "meta-core": "⚙️", + "meta-sort": "📁", + "meta-fuse": "🗂️", + "meta-stremio": "🎬", + "meta-dup": "🔍", + "meta-search": "🔎", + "meta-share": "🔗", + "meta-gateway": "🌉", + "meta-watch": "📺", + "meta-listen": "🎧", + "meta-read": "📖", +}; +const DEFAULT_ICON = "📦"; +const POLL_MS = 30000; + +function titleCase(name) { + return String(name || "") + .split("-") + .map((w) => w.charAt(0).toUpperCase() + w.slice(1)) + .join(" "); +} + +const TEMPLATE = ` + + + +`; + +class MetaServiceMenu extends HTMLElement { + constructor() { + super(); + this.attachShadow({ mode: "open" }); + this.shadowRoot.innerHTML = TEMPLATE; + this._btn = this.shadowRoot.querySelector("button.trigger"); + this._panel = this.shadowRoot.querySelector(".panel"); + this._neighbors = []; + this._timer = null; + this._open = false; + + this._onDocClick = (e) => { + if (!this._open) return; + if (e.composedPath().includes(this)) return; + this._setOpen(false); + }; + this._onKey = (e) => { + if (e.key === "Escape" && this._open) { + this._setOpen(false); + this._btn.focus(); + } + }; + } + + connectedCallback() { + this._btn.title = this.getAttribute("label") || "Services"; + this._btn.setAttribute("aria-label", this._btn.title); + this._btn.addEventListener("click", () => this._setOpen(!this._open)); + document.addEventListener("click", this._onDocClick); + document.addEventListener("keydown", this._onKey); + + this.refresh(); + this._timer = setInterval(() => this.refresh(), POLL_MS); + } + + disconnectedCallback() { + if (this._timer) clearInterval(this._timer); + this._timer = null; + document.removeEventListener("click", this._onDocClick); + document.removeEventListener("keydown", this._onKey); + } + + get endpoint() { + return this.getAttribute("endpoint") || "/api/neighbors"; + } + + get current() { + return this.getAttribute("current") || ""; + } + + async refresh() { + try { + const res = await fetch(this.endpoint, { + headers: { Accept: "application/json" }, + }); + if (!res.ok) throw new Error(String(res.status)); + const data = await res.json(); + // `services` is the legacy alias some backends still emit. + this._neighbors = data.neighbors || data.services || []; + } catch { + // Silent: the menu is navigation, not a health indicator. Keep whatever + // we last had rather than blanking the list on one failed poll. + } + this._render(); + } + + _setOpen(open) { + this._open = open; + this._panel.hidden = !open; + this._btn.setAttribute("aria-expanded", String(open)); + if (open) this.refresh(); + } + + _render() { + const panel = this._panel; + panel.textContent = ""; + + const seen = new Set(); + const rows = []; + for (const n of this._neighbors) { + const name = n.name || ""; + if (!name || seen.has(name)) continue; + seen.add(name); + rows.push(n); + } + // Always show the host itself, even before its own echo comes back. + if (this.current && !seen.has(this.current)) { + rows.push({ name: this.current, baseUrl: "" }); + } + rows.sort((a, b) => String(a.name).localeCompare(String(b.name))); + + if (rows.length === 0) { + const empty = document.createElement("div"); + empty.className = "empty"; + empty.textContent = "No services found"; + panel.appendChild(empty); + return; + } + + for (const n of rows) { + const isActive = n.name === this.current; + let url = n.baseUrl || ""; + if (typeof this.rewriteUrl === "function") { + try { + url = this.rewriteUrl(n.name, url, n) || ""; + } catch { + /* a bad hook must not blank the menu */ + } + } + + // A neighbour with no reachable URL is shown but not linked — better + // than a link that goes nowhere. + const el = document.createElement(isActive || !url ? "span" : "a"); + el.className = + "item" + (isActive ? " active" : "") + (!isActive && !url ? " dead" : ""); + if (el.tagName === "A") { + el.href = url; + el.setAttribute("role", "menuitem"); + } + if (!isActive && !url) el.title = "No reachable URL announced"; + + const icon = document.createElement("span"); + icon.className = "icon"; + icon.textContent = ICONS[n.name] || DEFAULT_ICON; + + const label = document.createElement("span"); + label.textContent = titleCase(n.name); + + el.append(icon, label); + if (n.status === "running" || isActive) { + const dot = document.createElement("span"); + dot.className = "dot"; + dot.title = n.status || "running"; + el.appendChild(dot); + } + panel.appendChild(el); + } + } +} + +if (!customElements.get("meta-service-menu")) { + customElements.define("meta-service-menu", MetaServiceMenu); +} + +export default MetaServiceMenu; diff --git a/docker/nginx.conf b/docker/nginx.conf index 513d3b4..c7bfd54 100644 --- a/docker/nginx.conf +++ b/docker/nginx.conf @@ -135,6 +135,14 @@ http { proxy_set_header Host $host; } + # meta-discovery v1 neighbours (supersedes /services, which stays + # while consumers migrate). + location /neighbors { + proxy_pass http://meta-core-api/neighbors; + proxy_http_version 1.1; + proxy_set_header Host $host; + } + # ========================================================================= # Leader/role endpoints # ========================================================================= diff --git a/go.mod b/go.mod index 977f2fb..63fd171 100644 --- a/go.mod +++ b/go.mod @@ -18,4 +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 ) diff --git a/go.sum b/go.sum index 43c5375..0eb8064 100644 --- a/go.sum +++ b/go.sum @@ -34,3 +34,5 @@ golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ss golang.org/x/net v0.21.0 h1:AQyQV4dYCvJ7vGmJyKki9+PBdyvhkSd8EIx/qb0AYv4= golang.org/x/net v0.21.0/go.mod h1:bIjVDfnllIU7BJ2DNgfnXvpSvtn8VRwhlsaeUTyUS44= golang.org/x/sys v0.0.0-20190204203706-41f3e6584952/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA= +golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= diff --git a/internal/api/admin_migrate.go b/internal/api/admin_migrate.go index 4d54628..13247f7 100644 --- a/internal/api/admin_migrate.go +++ b/internal/api/admin_migrate.go @@ -99,3 +99,33 @@ func (s *Server) handleSweepLiteratureEchoes(w http.ResponseWriter, r *http.Requ "byKey": res.ByKey, }) } + +// handleMigrateUserCids triggers the one-shot sweep that builds +// `udl:idx:user::cids` for User Data Layer records written before that +// index existed. See storage.UDLBackfillUserCids. +// +// The index is what makes whole-profile export a set read instead of a +// keyspace scan, and it is maintained by the upsert script from now on — so +// this only ever matters once, for data already on disk. Until it has run, a +// profile that predates the index exports as **empty**, which means a +// cross-instance recovery reports success having restored nothing. +// +// Returns {fixed: N} — cids added, not cids examined. Safe to call repeatedly: +// a second pass adds nothing. +func (s *Server) handleMigrateUserCids(w http.ResponseWriter, r *http.Request) { + if !s.storage.IsConnected() { + writeError(w, http.StatusServiceUnavailable, "storage not connected") + return + } + + fixed, err := s.storage.UDLBackfillUserCids() + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + + writeJSON(w, http.StatusOK, map[string]interface{}{ + "success": true, + "fixed": fixed, + }) +} diff --git a/internal/api/handlers.go b/internal/api/handlers.go index 117e098..d01afdf 100644 --- a/internal/api/handlers.go +++ b/internal/api/handlers.go @@ -14,6 +14,7 @@ import ( "github.com/gorilla/mux" "github.com/metazla/meta-core/internal/leader" + "github.com/metazla/meta-core/internal/meshdisco" ) // contentTypeByExt maps file extensions to MIME types @@ -79,13 +80,13 @@ type ErrorResponse struct { // Stable error-slug strings. Pinned in the doc; consumers may match on // these. Do not rename without a deprecation cycle. const ( - ErrAliasCollision = "alias_collision" - ErrUnknownRoot = "unknown_root" - ErrUnknownCID = "unknown_cid" - ErrSchemaViolation = "schema_violation" + ErrAliasCollision = "alias_collision" + ErrUnknownRoot = "unknown_root" + ErrUnknownCID = "unknown_cid" + ErrSchemaViolation = "schema_violation" ErrStorageUnavailable = "storage_unavailable" - ErrInternal = "internal" - ErrDeprecatedField = "deprecated_field" + ErrInternal = "internal" + ErrDeprecatedField = "deprecated_field" ) // rejectDeprecatedField reports whether a field name is one of the removed @@ -594,6 +595,44 @@ func (s *Server) handleGetService(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, service) } +// handleListNeighbors handles GET /neighbors and GET /api/neighbors. +// +// This is the meta-discovery v1 replacement for /api/services: 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 contract the old endpoint had. +// +// A `?all=1` query returns every instance uncollapsed, for debugging. +func (s *Server) handleListNeighbors(w http.ResponseWriter, r *http.Request) { + response := map[string]interface{}{ + "current": s.config.ServiceName, + "neighbors": []interface{}{}, + "count": 0, + } + + if s.mesh == nil { + // Discovery off or failed to bind. Report empty rather than 503 — + // the nav is decoration and callers already treat [] as "hide". + response["enabled"] = false + writeJSON(w, http.StatusOK, response) + return + } + + var list []meshdisco.Neighbor + if r.URL.Query().Get("all") != "" { + list = s.mesh.Neighbors() + } else { + list = s.mesh.NeighborsByName() + } + + response["enabled"] = true + response["neighbors"] = list + response["count"] = len(list) + response["self"] = s.mesh.Self() + writeJSON(w, http.StatusOK, response) +} + // handleCleanupStats handles GET /services/cleanup/stats func (s *Server) handleCleanupStats(w http.ResponseWriter, r *http.Request) { if s.cleaner == nil { @@ -959,10 +998,10 @@ func (s *Server) handleComputeFileCID(w http.ResponseWriter, r *http.Request) { // CIDv1 format: version (0x01) + codec (0x55) + multihash // Multihash format: hash-code (0x12) + length (0x20) + hash cidBytes := make([]byte, 0, 2+2+32) - cidBytes = append(cidBytes, 0x01) // CIDv1 - cidBytes = append(cidBytes, 0x55) // raw codec - cidBytes = append(cidBytes, 0x12) // sha256 code - cidBytes = append(cidBytes, 0x20) // 32 bytes + cidBytes = append(cidBytes, 0x01) // CIDv1 + cidBytes = append(cidBytes, 0x55) // raw codec + cidBytes = append(cidBytes, 0x12) // sha256 code + cidBytes = append(cidBytes, 0x20) // 32 bytes cidBytes = append(cidBytes, hashBytes...) // Encode as base32lower with 'b' prefix (multibase) diff --git a/internal/api/server.go b/internal/api/server.go index a1e3b17..2480e4f 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -13,6 +13,7 @@ import ( "github.com/metazla/meta-core/internal/events" "github.com/metazla/meta-core/internal/identity" "github.com/metazla/meta-core/internal/leader" + "github.com/metazla/meta-core/internal/meshdisco" "github.com/metazla/meta-core/internal/mounts" "github.com/metazla/meta-core/internal/schema" "github.com/metazla/meta-core/internal/storage" @@ -45,8 +46,19 @@ 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 - router *mux.Router - server *http.Server + // mesh is the UDP discovery node (meta-discovery v1). 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. + mesh *meshdisco.Node + router *mux.Router + server *http.Server +} + +// SetMeshNode attaches the UDP discovery node. Separate from NewServer so the +// server's constructor signature (and its tests) stay untouched. +func (s *Server) SetMeshNode(n *meshdisco.Node) { + s.mesh = n } // NewServer creates a new API server @@ -216,6 +228,8 @@ func (s *Server) setupRoutes() { s.router.HandleFunc("/api/udl/records", s.handleUDLRecordsPut).Methods("PUT") s.router.HandleFunc("/api/udl/user/{uid}/key/{key}", s.handleUDLUserKey).Methods("GET") s.router.HandleFunc("/api/udl/user/{uid}/cid/{cid}", s.handleUDLUserCid).Methods("GET") + s.router.HandleFunc("/api/udl/user/{uid}/all", s.handleUDLUserAll).Methods("GET") + s.router.HandleFunc("/api/udl/user/{uid}/count", s.handleUDLUserCount).Methods("GET") s.router.HandleFunc("/api/udl/cid/{cid}/users", s.handleUDLCidUsers).Methods("GET") s.router.HandleFunc("/api/udl/cid/{cid}/aggregate", s.handleUDLAggregate).Methods("GET") s.router.HandleFunc("/api/udl/users/stats", s.handleUDLUserStats).Methods("GET") @@ -264,6 +278,7 @@ func (s *Server) setupRoutes() { // One-shot vocabulary sweep: domain film|tv → screen, workForm backfill // (METADATA_KEYS.md §14.17). Idempotent. s.router.HandleFunc("/api/admin/migrate-domain-screen", s.handleMigrateDomainScreen).Methods("POST") + s.router.HandleFunc("/api/admin/migrate-user-cids", s.handleMigrateUserCids).Methods("POST") // One-shot sweep of query echoes on literature cards (meta-read // reading-model.md §7). Dry-run unless ?apply=true. Idempotent. s.router.HandleFunc("/api/admin/sweep-literature-echoes", s.handleSweepLiteratureEchoes).Methods("POST") @@ -282,6 +297,11 @@ func (s *Server) setupRoutes() { s.router.HandleFunc("/services/cleanup/stats", s.handleCleanupStats).Methods("GET") s.router.HandleFunc("/services/{name}", s.handleGetService).Methods("GET") + // meta-discovery v1: 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") + // Mount management routes (if manager initialized) if s.mountsHandlers != nil { s.mountsHandlers.RegisterRoutes(s.router) diff --git a/internal/api/udl.go b/internal/api/udl.go index 56adc49..77c767d 100644 --- a/internal/api/udl.go +++ b/internal/api/udl.go @@ -13,6 +13,7 @@ import ( "encoding/json" "math" "net/http" + "strconv" "strings" "github.com/gorilla/mux" @@ -271,6 +272,75 @@ func (s *Server) handleUDLUserCid(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, map[string]interface{}{"entries": entries}) } +// handleUDLUserAll handles GET /api/udl/user/{uid}/all?after=&limit= +// — every record a uid owns, for cross-instance profile recovery. +// +// The other user-scoped reads answer a question with one coordinate already +// known. Recovery knows only the uid: a box that has just imported a secret key +// has no idea which cids or keys the profile ever touched, so it needs the +// whole set or it gets nothing. +// +// Paged on a cid boundary — `nextCid` empty means the export is complete. The +// response is opaque signed records, exactly as stored; whoever consumes it is +// responsible for verifying each one before trusting it (meta-core has never +// held the account key and cannot verify on anyone's behalf). +func (s *Server) handleUDLUserAll(w http.ResponseWriter, r *http.Request) { + if !s.storage.IsConnected() { + writeError(w, http.StatusServiceUnavailable, "storage not connected") + return + } + uid := mux.Vars(r)["uid"] + if uid == "" { + writeError(w, http.StatusBadRequest, "uid is required") + return + } + limit := maxUDLBatch + if v := r.URL.Query().Get("limit"); v != "" { + n, err := strconv.Atoi(v) + if err != nil || n <= 0 { + writeError(w, http.StatusBadRequest, "limit must be a positive integer") + return + } + if n < limit { + limit = n + } + } + entries, nextCid, err := s.storage.UDLAllForUser(uid, r.URL.Query().Get("after"), limit) + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + writeJSON(w, http.StatusOK, map[string]interface{}{ + "entries": entries, + "nextCid": nextCid, + }) +} + +// handleUDLUserCount handles GET /api/udl/user/{uid}/count — how many cids a +// uid holds records for. +// +// Exists so a gateway mirror can enforce a per-profile quota without walking +// anything: 32 random bytes are a valid uid, so an unbounded mirror is a +// disk-fill DoS that needs no account. A quota check that itself cost a scan +// would just move the denial of service one layer up. +func (s *Server) handleUDLUserCount(w http.ResponseWriter, r *http.Request) { + if !s.storage.IsConnected() { + writeError(w, http.StatusServiceUnavailable, "storage not connected") + return + } + uid := mux.Vars(r)["uid"] + if uid == "" { + writeError(w, http.StatusBadRequest, "uid is required") + return + } + n, err := s.storage.UDLCountUserCids(uid) + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + writeJSON(w, http.StatusOK, map[string]interface{}{"cids": n}) +} + // handleUDLCidUsers handles GET /api/udl/cid/{cid}/users (uids_for_cid). func (s *Server) handleUDLCidUsers(w http.ResponseWriter, r *http.Request) { if !s.storage.IsConnected() { diff --git a/internal/config/config.go b/internal/config/config.go index f2873c9..42139a9 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -27,42 +27,54 @@ type Config struct { HTTPHost string // HTTP API host (default: "0.0.0.0" — meta-core always runs in a container and must answer siblings) // Timing configuration - HealthCheckIntervalMS int // Health check interval in ms (default: 5000) - HeartbeatIntervalMS int // Service heartbeat interval in ms (default: 30000) - StaleThresholdMS int // Stale service threshold in ms (default: 60000) - CleanupIntervalMS int // Dead service cleanup interval in ms (default: 600000) - DeadServiceThresholdMS int // Dead service threshold in ms (default: 600000) + HealthCheckIntervalMS int // Health check interval in ms (default: 5000) + HeartbeatIntervalMS int // Service heartbeat interval in ms (default: 30000) + StaleThresholdMS int // Stale service threshold in ms (default: 60000) + CleanupIntervalMS int // Dead service cleanup interval in ms (default: 600000) + DeadServiceThresholdMS int // Dead service threshold in ms (default: 600000) // File watcher configuration - WatchIntervalMS int // Polling interval for network mounts (default: 1000) - DebounceMS int // File change debounce time (default: 30000) - EnableFileWatcher bool // Enable file watcher (default: true) + WatchIntervalMS int // Polling interval for network mounts (default: 1000) + DebounceMS int // File change debounce time (default: 30000) + EnableFileWatcher bool // Enable file watcher (default: true) // Mount configuration MountsDir string // Path to mounts configuration (default: /meta-core/mounts) + + // UDP service discovery (meta-discovery v1). Replaces the file-based + // registry under ServicesDir(); 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) } // Load creates a Config from environment variables func Load() *Config { cfg := &Config{ - MetaCorePath: getEnv("META_CORE_PATH", "/meta-core"), - FilesPath: getEnv("FILES_PATH", "/files"), - ServiceName: getEnv("SERVICE_NAME", "meta-core"), - ServiceVersion: getEnv("SERVICE_VERSION", "1.0.0"), - APIPort: getEnvInt("API_PORT", 8180), - BaseURL: getEnv("BASE_URL", ""), - PublicURL: getEnv("META_CORE_PUBLIC_URL", ""), - RedisPort: getEnvInt("REDIS_PORT", 6379), - HTTPPort: getEnvInt("META_CORE_HTTP_PORT", 9000), - HTTPHost: getEnv("META_CORE_HTTP_HOST", "0.0.0.0"), + MetaCorePath: getEnv("META_CORE_PATH", "/meta-core"), + FilesPath: getEnv("FILES_PATH", "/files"), + ServiceName: getEnv("SERVICE_NAME", "meta-core"), + ServiceVersion: getEnv("SERVICE_VERSION", "1.0.0"), + APIPort: getEnvInt("API_PORT", 8180), + BaseURL: getEnv("BASE_URL", ""), + PublicURL: getEnv("META_CORE_PUBLIC_URL", ""), + RedisPort: getEnvInt("REDIS_PORT", 6379), + HTTPPort: getEnvInt("META_CORE_HTTP_PORT", 9000), + HTTPHost: getEnv("META_CORE_HTTP_HOST", "0.0.0.0"), HealthCheckIntervalMS: getEnvInt("HEALTH_CHECK_INTERVAL_MS", 5000), - HeartbeatIntervalMS: getEnvInt("HEARTBEAT_INTERVAL_MS", 30000), - StaleThresholdMS: getEnvInt("STALE_THRESHOLD_MS", 60000), - CleanupIntervalMS: getEnvInt("CLEANUP_INTERVAL_MS", 600000), + HeartbeatIntervalMS: getEnvInt("HEARTBEAT_INTERVAL_MS", 30000), + StaleThresholdMS: getEnvInt("STALE_THRESHOLD_MS", 60000), + CleanupIntervalMS: getEnvInt("CLEANUP_INTERVAL_MS", 600000), DeadServiceThresholdMS: getEnvInt("DEAD_SERVICE_THRESHOLD_MS", 600000), - WatchIntervalMS: getEnvInt("WATCH_INTERVAL_MS", 1000), - DebounceMS: getEnvInt("DEBOUNCE_MS", 30000), - EnableFileWatcher: getEnvBool("ENABLE_FILE_WATCHER", true), + WatchIntervalMS: getEnvInt("WATCH_INTERVAL_MS", 1000), + 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), } // Set mounts directory @@ -161,11 +173,11 @@ func (c *Config) DefaultWatchPath() string { // // - /files/watch — user-curated media. The main library. // - /files/plugin — plugin output (TMDB posters/backdrops, extracted -// subtitles, etc.). Indexed so each artefact gets a -// midhash256 alias and CID-based references from -// other files (e.g. `backdrop=`) resolve through -// the standard reverse index instead of needing a -// path companion field. +// subtitles, etc.). Indexed so each artefact gets a +// midhash256 alias and CID-based references from +// other files (e.g. `backdrop=`) resolve through +// the standard reverse index instead of needing a +// path companion field. func (c *Config) DefaultWatcherPaths() []string { return []string{ c.FilesPath + "/watch", diff --git a/internal/meshdisco/node.go b/internal/meshdisco/node.go new file mode 100644 index 0000000..0deecc7 --- /dev/null +++ b/internal/meshdisco/node.go @@ -0,0 +1,434 @@ +package meshdisco + +import ( + "context" + "encoding/json" + "fmt" + "log" + "net" + "os" + "sort" + "sync" + "syscall" + "time" + + "golang.org/x/net/ipv4" +) + +// 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. +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) +} + +func (c *Config) applyDefaults() { + if c.Group == "" { + c.Group = DefaultGroup + } + if c.Port == 0 { + c.Port = DefaultPort + } + 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 Node struct { + cfg Config + group net.IP + + pc *ipv4.PacketConn + closer interface{ Close() error } + + // 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 + // default route alone silently covers only one network. + ifaces []net.Interface + + // sendMu serialises SetMulticastInterface + WriteTo, which are a + // two-step operation on one shared socket. + sendMu sync.Mutex + + mu sync.RWMutex + neighbors map[string]*Neighbor + + stopChan chan struct{} + stopOnce sync.Once + wg sync.WaitGroup +} + +// New creates a Node. It does not touch the network until Start. +func New(cfg Config) *Node { + cfg.applyDefaults() + return &Node{ + cfg: cfg, + neighbors: make(map[string]*Neighbor), + 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. +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) + } + + 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) + } + n.closer = conn + + pc := ipv4.NewPacketConn(conn) + // 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) + } + n.pc = pc + + n.ifaces = eligibleInterfaces() + if len(n.ifaces) == 0 { + log.Println("[meshdisco] 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) + continue + } + joined++ + } + log.Printf("[meshdisco] 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() + + 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. +func (n *Node) Stop() error { + var err error + n.stopOnce.Do(func() { + n.sendAnnounce(nil, "stopping") + close(n.stopChan) + if n.closer != nil { + err = n.closer.Close() + } + n.wg.Wait() + }) + 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}) +} + +// 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. +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) + continue + } + out = append(out, *nb) + } + n.mu.Unlock() + + sort.Slice(out, func(i, j int) bool { + if out[i].Name != out[j].Name { + return out[i].Name < out[j].Name + } + return out[i].Instance < out[j].Instance + }) + 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). +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) { + best[nb.Name] = nb + } + } + out := make([]Neighbor, 0, len(best)) + for _, nb := range best { + out = append(out, nb) + } + sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name }) + 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. +func (n *Node) Self() Neighbor { + return Neighbor{Message: n.buildAnnounce("running"), LastSeen: time.Now()} +} + +func (n *Node) readLoop() { + defer n.wg.Done() + buf := make([]byte, ReadBufferSize) + + for { + select { + case <-n.stopChan: + return + default: + } + + nRead, _, src, err := n.pc.ReadFrom(buf) + if err != nil { + select { + case <-n.stopChan: + return + default: + } + log.Printf("[meshdisco] 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 + } + + 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: + n.record(msg, src) + } + } +} + +func (n *Node) record(msg Message, src net.Addr) { + addr := "" + if src != nil { + if udp, ok := src.(*net.UDPAddr); ok { + addr = udp.IP.String() + } else { + addr = src.String() + } + } + + if msg.Status == "stopping" { + n.mu.Lock() + delete(n.neighbors, msg.key()) + n.mu.Unlock() + return + } + + nb := &Neighbor{Message: msg, Addr: addr, LastSeen: time.Now()} + n.mu.Lock() + n.neighbors[msg.key()] = nb + n.mu.Unlock() +} + +func (n *Node) announceLoop() { + 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") + n.Probe() + + ticker := time.NewTicker(n.cfg.Interval) + defer ticker.Stop() + + for { + select { + case <-n.stopChan: + return + case <-ticker.C: + n.sendAnnounce(nil, "running") + } + } +} + +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 + } + body, err := json.Marshal(msg) + if err != nil { + 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) + } +} + +// broadcast writes one message to the group once per eligible interface. +// Writing once and letting the routing table choose would reach exactly one +// network, which is the failure mode this protocol exists to avoid. +func (n *Node) broadcast(msg Message) { + if n.pc == nil { + return + } + body, err := json.Marshal(msg) + if err != nil { + return + } + dst := &net.UDPAddr{IP: n.group, Port: n.cfg.Port} + + n.sendMu.Lock() + 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) + 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) + } + } +} + +// eligibleInterfaces returns every up, non-loopback, multicast-capable +// interface that has an IPv4 address. +func eligibleInterfaces() []net.Interface { + all, err := net.Interfaces() + if err != nil { + log.Printf("[meshdisco] Warning: could not enumerate interfaces: %v", err) + return nil + } + + var out []net.Interface + for _, ifi := range all { + if ifi.Flags&net.FlagUp == 0 || + ifi.Flags&net.FlagLoopback != 0 || + ifi.Flags&net.FlagMulticast == 0 { + continue + } + addrs, err := ifi.Addrs() + if err != nil { + continue + } + for _, a := range addrs { + if ipnet, ok := a.(*net.IPNet); ok && ipnet.IP.To4() != nil { + out = append(out, ifi) + break + } + } + } + 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. +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) + }) + if err != nil { + return err + } + return serr +} + +// LocalIPv4 returns this host's first non-loopback IPv4 address, or the +// hostname when none can be determined. Same fallback the file-based registry +// used (discovery/service.go), kept identical so announced URLs don't shift. +func LocalIPv4() string { + addrs, err := net.InterfaceAddrs() + if err == nil { + for _, addr := range addrs { + if ipnet, ok := addr.(*net.IPNet); ok && !ipnet.IP.IsLoopback() { + if ipnet.IP.To4() != nil { + return ipnet.IP.String() + } + } + } + } + host, _ := os.Hostname() + return host +} diff --git a/internal/meshdisco/proto.go b/internal/meshdisco/proto.go new file mode 100644 index 0000000..5ea943f --- /dev/null +++ b/internal/meshdisco/proto.go @@ -0,0 +1,132 @@ +// Package meshdisco implements meta-discovery v1: UDP multicast service +// discovery for the meta-* services. +// +// It replaces two file-based mechanisms that both required a shared +// /meta-core 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. +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" +) + +// 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. +type URLs struct { + Hostname string `json:"hostname"` + BaseUrl string `json:"baseUrl"` + ApiUrl string `json:"apiUrl"` + WebdavUrl string `json:"webdavUrl"` + 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". + 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"` +} + +// 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 + } + return m.Name + "|" + m.Instance +} diff --git a/internal/storage/udl.go b/internal/storage/udl.go index 5e20294..875187b 100644 --- a/internal/storage/udl.go +++ b/internal/storage/udl.go @@ -17,6 +17,7 @@ package storage // udl:rec:// HASH { record, version, ts, value? } // udl:idx:user::key: SET of -> active_for_user_key (My List, Continue Watching) // udl:idx:user::cid: SET of -> list_for_user_cid +// udl:idx:user::cids SET of -> whole-profile export (recovery) // udl:idx:cid::key: SET of -> aggregate // udl:idx:cid::users SET of -> uids_for_cid // @@ -27,7 +28,9 @@ package storage import ( "context" "fmt" + "sort" "strconv" + "strings" "time" "github.com/redis/go-redis/v9" @@ -66,11 +69,24 @@ func udlIdxUserCid(uid, cid string) string { return fmt.Sprintf("udl:idx:user:%s func udlIdxCidKey(cid, key string) string { return fmt.Sprintf("udl:idx:cid:%s:key:%s", cid, key) } func udlIdxCidUsers(cid string) string { return fmt.Sprintf("udl:idx:cid:%s:users", cid) } +// udlIdxUserCids is the set of every cid a uid has ANY record for — the one +// index that makes "export this whole profile" a set read instead of a +// keyspace scan. +// +// The other four indexes each answer a question with one coordinate already +// known (a key, a cid). Cross-instance recovery knows neither: it has a uid and +// nothing else. Without this set the only ways to enumerate a profile are to +// iterate the hardcoded key list — which silently misses any key added later — +// or to SCAN `udl:rec:/*`, which is the full-keyspace-read pathology that +// already made `/api/udl/users/stats` admin-only. +func udlIdxUserCids(uid string) string { return fmt.Sprintf("udl:idx:user:%s:cids", uid) } + // udlUpsertScript performs the version-gated upsert + index maintenance // atomically. Accepts (returns 1) only when the incoming version strictly // exceeds the stored one; otherwise returns 0 (stale). // // KEYS: 1=cell 2=idx:user:key 3=idx:user:cid 4=idx:cid:key 5=idx:cid:users +// 6=idx:user:cids // ARGV: 1=version 2=ts 3=record 4=value 5=hasValue("1"/"0") // 6=cid 7=key 8=uid 9=tombstone("1"/"0") // @@ -78,9 +94,16 @@ func udlIdxCidUsers(cid string) string { return fmt.Sprintf("udl:idx:cid:%s // would let an older write resurrect the value — but it SREMs the cid from the // user+key index. That index is what `active_for_user_key` returns, so leaving // deleted cids in it means every My List / Continue Watching read carries every -// title ever un-liked or finished, forever. The other three indexes are left +// title ever un-liked or finished, forever. The other four indexes are left // alone: they answer "who has an opinion about this cid", where a tombstone is // still an answer. +// +// ⚠ `idx:user:cids` (KEYS[6]) in particular must NEVER be pruned on a +// tombstone, even though it looks symmetric with KEYS[2]. It is the export +// index: a cid dropped from it is a cid whose tombstone never travels, and a +// tombstone that does not travel is exactly how an un-liked title comes back +// from the dead on the other box — the peer still holds the older `like` at a +// lower version, and with no tombstone to beat it, nothing ever supersedes it. var udlUpsertScript = redis.NewScript(` local stored = redis.call('HGET', KEYS[1], 'version') if stored and tonumber(stored) >= tonumber(ARGV[1]) then @@ -100,6 +123,7 @@ end redis.call('SADD', KEYS[3], ARGV[7]) redis.call('SADD', KEYS[4], ARGV[8]) redis.call('SADD', KEYS[5], ARGV[8]) +redis.call('SADD', KEYS[6], ARGV[6]) return 1 `) @@ -126,6 +150,7 @@ func (c *Client) UDLUpsertIfNewer(uid, cid, key string, version, ts int64, recor c.buildKey(udlIdxUserCid(uid, cid)), c.buildKey(udlIdxCidKey(cid, key)), c.buildKey(udlIdxCidUsers(cid)), + c.buildKey(udlIdxUserCids(uid)), } hv := "0" if hasValue { @@ -250,6 +275,160 @@ func (c *Client) UDLListUserCid(uid, cid string) ([]UDLUserCidEntry, error) { return out, nil } +// UDLUserAllEntry is one row of a whole-profile export. +type UDLUserAllEntry struct { + Cid string `json:"cid"` + Key string `json:"key"` + Record string `json:"record"` + Version int64 `json:"version"` + Ts int64 `json:"ts"` +} + +// UDLAllForUser returns every cell a uid owns, for cross-instance recovery. +// +// Ordered by cid (sorted, so paging is stable across calls even though the +// underlying index is a Redis SET), starting strictly after `afterCid`. A cid's +// keys are never split across a page: the page ends on a cid boundary once +// `limit` rows have been emitted, and `nextCid` is the last cid included — pass +// it back as `afterCid` to continue. `nextCid` is empty when the export is +// complete. +// +// Tombstones are included. They are the CRDT state that stops a deleted value +// resurrecting on the box that receives this export, so omitting them would +// make recovery actively wrong rather than merely incomplete. +func (c *Client) UDLAllForUser(uid, afterCid string, limit int) ([]UDLUserAllEntry, string, error) { + c.mu.RLock() + defer c.mu.RUnlock() + + if c.client == nil { + return nil, "", fmt.Errorf("not connected") + } + if limit <= 0 { + limit = 1000 + } + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + cids, err := c.client.SMembers(ctx, c.buildKey(udlIdxUserCids(uid))).Result() + if err != nil { + return nil, "", fmt.Errorf("smembers failed: %w", err) + } + sort.Strings(cids) + + out := make([]UDLUserAllEntry, 0, limit) + nextCid := "" + for _, cid := range cids { + if afterCid != "" && cid <= afterCid { + continue + } + if len(out) >= limit { + nextCid = out[len(out)-1].Cid + break + } + keys, err := c.client.SMembers(ctx, c.buildKey(udlIdxUserCid(uid, cid))).Result() + if err != nil { + return nil, "", fmt.Errorf("smembers failed: %w", err) + } + sort.Strings(keys) + for _, key := range keys { + cell, ok, err := c.udlReadCell(ctx, udlRecKey(uid, cid, key)) + if err != nil { + return nil, "", err + } + if !ok { + continue + } + out = append(out, UDLUserAllEntry{ + Cid: cid, Key: key, Record: cell.Record, Version: cell.Version, Ts: cell.Ts, + }) + } + } + return out, nextCid, nil +} + +// UDLCountUserCids is the number of distinct cids a uid holds records for. +// +// O(1) (a Redis SCARD), which is what makes it usable as a per-push quota +// check on the gateway mirror — a quota that cost a keyspace walk would be a +// denial-of-service in its own right. Counts cids rather than cells because +// that is what the index holds; a profile is bounded by the titles it has +// touched, not by how many keys it has per title. +func (c *Client) UDLCountUserCids(uid string) (int64, error) { + c.mu.RLock() + defer c.mu.RUnlock() + + if c.client == nil { + return 0, fmt.Errorf("not connected") + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + n, err := c.client.SCard(ctx, c.buildKey(udlIdxUserCids(uid))).Result() + if err != nil { + return 0, fmt.Errorf("scard failed: %w", err) + } + return n, nil +} + +// UDLBackfillUserCids rebuilds the `idx:user::cids` sets for data written +// before that index existed, and reports how many cids it added. +// +// One-shot: the upsert script maintains the set from now on. It is a SCAN, but +// a targeted one (`udl:idx:user:*:cid:*`, which is one key per (uid,cid), not +// per record) run once at boot rather than per read — the distinction that +// makes the difference between a migration and the pathology this index exists +// to avoid. +func (c *Client) UDLBackfillUserCids() (int, error) { + c.mu.Lock() + defer c.mu.Unlock() + + if c.client == nil { + return 0, fmt.Errorf("not connected") + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) + defer cancel() + + pattern := c.buildKey("udl:idx:user:*:cid:*") + prefix := c.buildKey("udl:idx:user:") + added := 0 + var cursor uint64 + for { + keys, next, err := c.client.Scan(ctx, cursor, pattern, 500).Result() + if err != nil { + return added, fmt.Errorf("scan failed: %w", err) + } + for _, k := range keys { + // :cid: — split on the FIRST separator, not the + // last. A uid is multibase base58btc, whose alphabet contains no + // colon at all, so the first ":cid:" always ends the uid. A cid + // that happens to contain ":cid:" in its own name would make the + // last occurrence split in the wrong place. + rest := strings.TrimPrefix(k, prefix) + i := strings.Index(rest, ":cid:") + if i < 0 { + continue + } + uid, cid := rest[:i], rest[i+len(":cid:"):] + if uid == "" || cid == "" { + continue + } + n, err := c.client.SAdd(ctx, c.buildKey(udlIdxUserCids(uid)), cid).Result() + if err != nil { + return added, fmt.Errorf("sadd failed: %w", err) + } + added += int(n) + } + cursor = next + if cursor == 0 { + break + } + } + return added, nil +} + // UDLCidUsers returns every uid that has any record for cid — uids_for_cid. func (c *Client) UDLCidUsers(cid string) ([]string, error) { c.mu.RLock() diff --git a/internal/storage/udl_admin.go b/internal/storage/udl_admin.go index 5c7f521..b0238f5 100644 --- a/internal/storage/udl_admin.go +++ b/internal/storage/udl_admin.go @@ -179,7 +179,7 @@ func (c *Client) UDLAllUserStats() (map[string]UDLUserStats, error) { } // UDLPurgeUser removes every trace of one account from the User Data Layer: -// its cells, its two user-scoped indexes, and its membership in the shared +// its cells, its three user-scoped indexes, and its membership in the shared // cid-scoped indexes. Returns what was removed. // // The shared indexes are the part that is easy to miss and expensive to leave. @@ -225,8 +225,10 @@ func (c *Client) UDLPurgeUser(uid string) (UDLUserStats, error) { stats.Cids = len(distinctCids) stats.Keys = len(distinctKeys) - // The two user-scoped indexes are enumerable directly, and are dropped - // whole rather than member-by-member. Scanning them separately also catches + // The three user-scoped indexes (`:key:*`, `:cid:*`, `:cids`) are enumerable + // directly, and are dropped whole rather than member-by-member. The glob is + // what keeps this correct as indexes are added — `:cids` needed no change + // here when it landed. Scanning them separately also catches // index entries whose cell is already gone — exactly the state a previously // half-finished purge leaves behind. idxPrefix := c.buildKey("udl:idx:user:") + uid + ":" diff --git a/internal/storage/udl_export_test.go b/internal/storage/udl_export_test.go new file mode 100644 index 0000000..9ffeef5 --- /dev/null +++ b/internal/storage/udl_export_test.go @@ -0,0 +1,234 @@ +package storage + +import ( + "context" + "testing" +) + +// The whole-profile export is what cross-instance recovery reads: a box that +// has just imported a secret key knows the uid and nothing else — not which +// cids the profile touched, not which keys. Every one of these tests is about +// that one caller. + +func TestUDLAllForUser_ReturnsEveryCellAcrossCidsAndKeys(t *testing.T) { + c, _ := newTestClient(t, "") + + const uid = "zAlice" + writes := []struct{ cid, key, rec string }{ + {"bagcsaaaaone", "like", "rec-one-like"}, + {"bagcsaaaaone", "seek", "rec-one-seek"}, + {"bagcsaaaatwo", "rating", "rec-two-rating"}, + {"bagcsaaaatwo", "profile:name", "rec-two-name"}, + {"bagcsaaathree", "watched", "rec-three-watched"}, + } + for _, w := range writes { + if ok, err := c.UDLUpsertIfNewer(uid, w.cid, w.key, 1, 100, w.rec, "", false, false); err != nil || !ok { + t.Fatalf("upsert %s/%s: ok=%v err=%v", w.cid, w.key, ok, err) + } + } + + entries, next, err := c.UDLAllForUser(uid, "", 0) + if err != nil { + t.Fatalf("export: %v", err) + } + if next != "" { + t.Fatalf("export of 5 rows should be complete, got nextCid=%q", next) + } + if len(entries) != len(writes) { + t.Fatalf("want %d rows, got %d: %+v", len(writes), len(entries), entries) + } + seen := map[string]string{} + for _, e := range entries { + seen[e.Cid+"/"+e.Key] = e.Record + } + for _, w := range writes { + if got := seen[w.cid+"/"+w.key]; got != w.rec { + t.Errorf("%s/%s: want record %q, got %q", w.cid, w.key, w.rec, got) + } + } +} + +// The failure this guards is silent and total: a uid whose cids were never +// indexed exports as empty, and recovery reports success having restored +// nothing. +func TestUDLAllForUser_IsEmptyForAnUnknownUid(t *testing.T) { + c, _ := newTestClient(t, "") + + if ok, err := c.UDLUpsertIfNewer("zAlice", "bagcsaaaaone", "like", 1, 100, "rec", "", false, false); err != nil || !ok { + t.Fatalf("upsert: ok=%v err=%v", ok, err) + } + entries, _, err := c.UDLAllForUser("zBob", "", 0) + if err != nil { + t.Fatalf("export: %v", err) + } + if len(entries) != 0 { + t.Fatalf("Bob owns nothing; got %+v", entries) + } +} + +// ⚠ The tombstone rule, and the reason the cids index is never pruned. +// +// A tombstone must EXPORT. It is the only thing that can beat the older live +// value the other box still holds at a lower version — drop it from the export +// and an un-liked title comes back from the dead on every peer that already +// saw the `like`. This is the one place the cids index deliberately diverges +// from the user+key index, which does prune. +func TestUDLAllForUser_ExportsTombstones(t *testing.T) { + c, _ := newTestClient(t, "") + + const ( + uid = "zAlice" + cid = "bagcsaaaaone" + key = "like" + ) + if ok, err := c.UDLUpsertIfNewer(uid, cid, key, 1, 100, "rec-liked", "", false, false); err != nil || !ok { + t.Fatalf("live upsert: ok=%v err=%v", ok, err) + } + if ok, err := c.UDLUpsertIfNewer(uid, cid, key, 2, 200, "rec-unliked", "", false, true); err != nil || !ok { + t.Fatalf("tombstone upsert: ok=%v err=%v", ok, err) + } + + // The user+key index prunes it — that is what keeps My List short. + live, err := c.UDLListUserKey(uid, key) + if err != nil { + t.Fatalf("active_for_user_key: %v", err) + } + if len(live) != 0 { + t.Fatalf("tombstoned cid should leave the user+key index, got %+v", live) + } + + // The export must still carry it, at the tombstone's version. + entries, _, err := c.UDLAllForUser(uid, "", 0) + if err != nil { + t.Fatalf("export: %v", err) + } + if len(entries) != 1 { + t.Fatalf("tombstone must export, got %+v", entries) + } + if entries[0].Record != "rec-unliked" || entries[0].Version != 2 { + t.Fatalf("export must carry the tombstone, not the superseded value: %+v", entries[0]) + } +} + +func TestUDLAllForUser_PagesOnACidBoundaryAndResumes(t *testing.T) { + c, _ := newTestClient(t, "") + + const uid = "zAlice" + cids := []string{"bagcsaaaaa1", "bagcsaaaaa2", "bagcsaaaaa3"} + for _, cid := range cids { + for _, key := range []string{"like", "seek"} { + if ok, err := c.UDLUpsertIfNewer(uid, cid, key, 1, 100, "rec-"+cid+"-"+key, "", false, false); err != nil || !ok { + t.Fatalf("upsert: ok=%v err=%v", ok, err) + } + } + } + + // A cid's keys are never split across a page, so a limit of 1 still yields + // both of the first cid's rows. + page1, next1, err := c.UDLAllForUser(uid, "", 1) + if err != nil { + t.Fatalf("page 1: %v", err) + } + if len(page1) != 2 || page1[0].Cid != cids[0] { + t.Fatalf("page 1 should hold both rows of %s, got %+v", cids[0], page1) + } + if next1 != cids[0] { + t.Fatalf("cursor should name the last cid emitted, want %q got %q", cids[0], next1) + } + + // Walking the cursor must visit every row exactly once. + all := append([]UDLUserAllEntry{}, page1...) + cursor := next1 + for cursor != "" { + page, next, err := c.UDLAllForUser(uid, cursor, 1) + if err != nil { + t.Fatalf("page after %q: %v", cursor, err) + } + all = append(all, page...) + cursor = next + } + if len(all) != 6 { + t.Fatalf("want 6 rows across all pages, got %d: %+v", len(all), all) + } + seen := map[string]bool{} + for _, e := range all { + k := e.Cid + "/" + e.Key + if seen[k] { + t.Fatalf("%s emitted twice across pages", k) + } + seen[k] = true + } +} + +// Data written before the cids index existed must still export, or every +// profile that predates this feature silently recovers as empty. +func TestUDLBackfillUserCids_MakesLegacyDataExportable(t *testing.T) { + c, _ := newTestClient(t, "") + + const ( + uid = "zAlice" + cid = "bagcsaaaaone" + key = "like" + ) + if ok, err := c.UDLUpsertIfNewer(uid, cid, key, 1, 100, "rec", "", false, false); err != nil || !ok { + t.Fatalf("upsert: ok=%v err=%v", ok, err) + } + + // Simulate the pre-index state: the cell and the other indexes exist, the + // cids set does not. + ctx := context.Background() + if err := c.client.Del(ctx, c.buildKey(udlIdxUserCids(uid))).Err(); err != nil { + t.Fatalf("del cids index: %v", err) + } + if entries, _, err := c.UDLAllForUser(uid, "", 0); err != nil || len(entries) != 0 { + t.Fatalf("precondition: legacy data should export as empty, got %+v err=%v", entries, err) + } + + added, err := c.UDLBackfillUserCids() + if err != nil { + t.Fatalf("backfill: %v", err) + } + if added != 1 { + t.Fatalf("want 1 cid backfilled, got %d", added) + } + entries, _, err := c.UDLAllForUser(uid, "", 0) + if err != nil { + t.Fatalf("export after backfill: %v", err) + } + if len(entries) != 1 || entries[0].Cid != cid || entries[0].Key != key { + t.Fatalf("backfilled data should export, got %+v", entries) + } + + // Idempotent: a second run adds nothing. + if added, err := c.UDLBackfillUserCids(); err != nil || added != 0 { + t.Fatalf("second backfill should be a no-op, added=%d err=%v", added, err) + } +} + +// The backfill splits `:cid:` on the FIRST separator, because a uid +// is multibase base58btc and its alphabet has no colon. Splitting on the last +// occurrence — the obvious-looking choice — mis-parses a cid that contains +// ":cid:" in its own name into uid "zAlice:cid:weird", cid "name". +func TestUDLBackfillUserCids_SplitsOnTheFirstSeparator(t *testing.T) { + c, _ := newTestClient(t, "") + + const uid = "zAlice" + cid := "weird:cid:name" + if ok, err := c.UDLUpsertIfNewer(uid, cid, "like", 1, 100, "rec", "", false, false); err != nil || !ok { + t.Fatalf("upsert: ok=%v err=%v", ok, err) + } + ctx := context.Background() + if err := c.client.Del(ctx, c.buildKey(udlIdxUserCids(uid))).Err(); err != nil { + t.Fatalf("del cids index: %v", err) + } + if _, err := c.UDLBackfillUserCids(); err != nil { + t.Fatalf("backfill: %v", err) + } + entries, _, err := c.UDLAllForUser(uid, "", 0) + if err != nil { + t.Fatalf("export: %v", err) + } + if len(entries) != 1 || entries[0].Cid != cid { + t.Fatalf("want the cid parsed whole, got %+v", entries) + } +}