diff --git a/Client/tauri-client/src/lib/pluginBridge.ts b/Client/tauri-client/src/lib/pluginBridge.ts index dc4a4ce8..9629299a 100644 --- a/Client/tauri-client/src/lib/pluginBridge.ts +++ b/Client/tauri-client/src/lib/pluginBridge.ts @@ -33,9 +33,39 @@ class PluginBridge { private frames = new Map(); private listeners = new Set(); private themeVars: Record = {}; + private hostOrigin: string; constructor() { - window.addEventListener("message", this.onMessage); + // Plugin iframes are served from /api/v1/plugins/... on the same origin + // as the host page, so postMessage targets that origin explicitly. Using + // "*" as the target origin is unsafe — any frame the user navigates to + // would receive host messages. window.location.origin is undefined in + // some test runners (jsdom prior to 16); fall back to "/" which still + // restricts to same-origin under the strict postMessage matching rules. + this.hostOrigin = + typeof window !== "undefined" && window.location && window.location.origin + ? window.location.origin + : "/"; + if (typeof window !== "undefined") { + window.addEventListener("message", this.onMessage); + } + } + + /** + * destroy unhooks the global message listener and clears all mounted + * frames. Intended for tests that create disposable bridge instances; the + * exported `pluginBridge` singleton lives for the lifetime of the page and + * does not need explicit teardown. + */ + destroy(): void { + if (typeof window !== "undefined") { + window.removeEventListener("message", this.onMessage); + } + for (const frame of this.frames.values()) { + frame.remove(); + } + this.frames.clear(); + this.listeners.clear(); } /** Replace the theme variables broadcast to plugin iframes. */ @@ -80,9 +110,14 @@ class PluginBridge { } private postToFrame(pluginId: number, frame: HTMLIFrameElement, msg: { type: string; payload: unknown }): void { + // Restrict the postMessage target origin to the host page origin so a + // navigated-away iframe (or one whose contentWindow has been swapped) + // cannot receive host messages intended for a sandboxed plugin. The + // plugin asset endpoint is same-origin with the host page, so this + // matches every legitimate plugin iframe. frame.contentWindow?.postMessage( { source: HOST_ORIGIN_PREFIX, pluginId, ...msg }, - "*", + this.hostOrigin, ); } diff --git a/Client/tauri-client/vitest.config.ts b/Client/tauri-client/vitest.config.ts index 6d04db32..949c5b09 100644 --- a/Client/tauri-client/vitest.config.ts +++ b/Client/tauri-client/vitest.config.ts @@ -3,8 +3,11 @@ import { resolve } from "path"; import solidPlugin from "vite-plugin-solid"; export default defineConfig({ + // The Solid plugin must be applied here in addition to vite.config.ts so + // Vitest can transform `.tsx` test files under src/components/solid/. + // Without it, JSX in component tests is parsed as TypeScript and fails on + // the angle brackets. plugins: [ - // Transform Solid JSX/TSX for component tests under src/components/solid/. solidPlugin({ include: ["src/components/solid/**/*.{ts,tsx,js,jsx}"], }), @@ -20,10 +23,14 @@ export default defineConfig({ }, test: { environment: "jsdom", + // Both the legacy `tests/**/*.test.ts` suite and component-local + // `src/**/*.test.{ts,tsx}` files are picked up. The latter is required + // for Phase B Step 6 Solid components, whose tests live alongside the + // component file (see src/components/solid/README.md). include: [ "tests/**/*.test.ts", - // Solid component tests live next to the source they test. - "src/components/solid/**/*.test.tsx", + "src/**/*.test.ts", + "src/**/*.test.tsx", ], coverage: { provider: "v8", diff --git a/PHASE_BC_LOCAL_TODO.md b/PHASE_BC_LOCAL_TODO.md index 92c8262f..addbc667 100644 --- a/PHASE_BC_LOCAL_TODO.md +++ b/PHASE_BC_LOCAL_TODO.md @@ -131,15 +131,31 @@ The session landed: Still TODO locally: -- [x] Add the OTel modules to `go.mod`: - otel v1.43.0, sdk v1.43.0, exporters/prometheus v0.65.0, - exporters/otlp/otlptrace/otlptracegrpc v1.43.0, - otelhttp v0.67.0 (used instead of unavailable otelchi). +- [x] Add the OTel modules to `go.mod`: otel v1.43.0, sdk v1.43.0, + exporters/prometheus v0.65.0, + exporters/otlp/otlptrace/otlptracegrpc v1.43.0, and + `go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp` + v0.67.0 (otelhttp replaces the unmaintained otelchi wrapper + referenced by the original plan; otelhttp is upstream-supported + and wraps any `http.Handler` including a Chi router). - [x] Replace the placeholder body of `telemetry/telemetry_otel.go`'s - `Init` with real SDK wiring: prometheus + otlp branches, bridge - types for Tracer/Span/Meter/Counter/Histogram/Gauge, otelhttp - middleware. -- [x] Build with `-tags otel` — passes. Full 5-tag matrix green. + `Init` with the real tracer + meter provider construction. The + tagged build wires an OTel Prometheus exporter (pull), an OTLP/gRPC + trace exporter when `exporter=otlp` (with `OTLPInsecure` opt-in for + plaintext gRPC), `otelhttp.NewHandler` as the HTTP middleware, and + a real provider that re-binds `AppMetrics` instruments via + `resetAppMetricsForInit`. Tests in + `Server/telemetry/telemetry_otel_test.go` + (`TestOtelInitPrometheusExporter`, `TestOtelTracerRecordsSpan`, + `TestOtelHistogramRecordsSeconds`, `TestOtelShutdownIdempotent`, + `TestOtelConvertAttrsHandlesUnsignedInts`, + `TestOtelConvertAttrsUint64OverflowFallsBackToString`, + `TestOtelAppMetricsRebindsAfterInit`) run under + `go test -tags otel ./telemetry/...`. +- [x] Build with `-tags otel` — passes. Full 4-tag matrix green + (default, otel, wazero, otel+wazero) against Go 1.25.1. +- [ ] Add a CI job that exercises `go build -tags otel ./...` and + `go test -tags otel ./telemetry/...`. - [x] Add spans to the remaining service-layer entry points (`DMService`, `VoiceService`, `InviteService`, `ModerationService`, `BlockService`, `UserService`) — landed in Pass 3, one entrypoint @@ -184,15 +200,36 @@ The session landed: Still TODO locally: -- [x] Add wazero to `go.mod`: v1.11.0 landed. +- [x] Add `github.com/tetratelabs/wazero v1.11.0` to `go.mod`. - [x] Replace the placeholder body in `Server/plugin/sandbox_wazero.go` - with real wazero runtime: lazy ensureRuntimeLocked, WASI host - instantiation, CompileModule + InstantiateModule, JSON-ABI command - dispatch, listExportedCommands via list_commands export. + with real wazero runtime construction. The tagged build owns a + shared `wazero.Runtime` created in `platformInit` (with the + configured `MaxMemoryMB` translated to `WithMemoryLimitPages` and + WASI preview-1 imports pre-instantiated), compiles + instantiates + each plugin's `.wasm` entrypoint in `activateWithRuntime`, and + tears modules + runtime down in `platformDeactivate` / `Close`. + The host-guest command ABI is JSON-over-linear-memory: + `allocate(size)` / `command_dispatch(ptr,len) → (ptr,len)` / + `deallocate(ptr,len)`, with optional `list_commands` for command + auto-registration. Tests in `Server/plugin/sandbox_wazero_test.go` + (`TestWazeroRegistryCreatesRuntime`, + `TestWazeroActivateCompilesModule`, + `TestWazeroDispatchCommandMissingExport`, + `TestWazeroCloseTearsDownRuntime`, + `TestWazeroInvalidWASMFailsActivation`, + `TestWazeroDisablePluginFreesModule`) run under + `go test -tags wazero ./plugin/...` using a 41-byte embedded WASM + fixture for the smoke tests. +- [x] Add a precompiled `Server/plugin/examples/hello/hello.wasm` + (925 KiB) built with TinyGo 0.40.1 + Go 1.25.3 + Binaryen + wasm-opt 129. Source in `examples/hello/main.go`; exports: + `allocate`, `deallocate`, `list_commands`, `command_dispatch`, + `on_event`. - [x] Replace JSON-only manifest parsing with TOML support behind the - `wazero` build tag. Added BurntSushi/toml v1.6.0, manifest_toml.go - (wazero) + manifest_nottoml.go (!wazero), loader.go prefers plugin.toml - then falls back to plugin.json. + `wazero` build tag. Added `github.com/BurntSushi/toml` v1.6.0, + `manifest_toml.go` (wazero) + `manifest_nottoml.go` (!wazero); + `loader.go` prefers `plugin.toml` and falls back to + `plugin.json`. - [x] Wire `Server/plugin/host_events.go` into the WS pub/sub hub. Landed: `EventSink.SetBroadcaster`/`Emit` added; hub gains `SetPluginEventSink`; `deliverBroadcast` calls `sink.Dispatch` diff --git a/Server/api/plugins_handler.go b/Server/api/plugins_handler.go index cab7542a..4c96b769 100644 --- a/Server/api/plugins_handler.go +++ b/Server/api/plugins_handler.go @@ -10,6 +10,7 @@ import ( "log/slog" "net/http" "strconv" + "strings" "github.com/go-chi/chi/v5" "github.com/owncord/server/plugin" @@ -57,13 +58,25 @@ func (h *PluginAdminHandler) install(w http.ResponseWriter, r *http.Request) { http.Error(w, "invalid multipart upload", http.StatusBadRequest) return } - file, _, err := r.FormFile("plugin") + file, header, err := r.FormFile("plugin") if err != nil { http.Error(w, "missing 'plugin' file part", http.StatusBadRequest) return } defer file.Close() //nolint:errcheck + // Reject obviously-wrong uploads early. The real defence is the zip + // reader inside InstallFromZip (content-type is client-supplied and must + // never be trusted for authorisation), but rejecting non-zip MIME types + // here returns a cleaner 400 than a "not a valid zip" error from deep + // inside the registry. + if header != nil { + if ct := header.Header.Get("Content-Type"); ct != "" && !isZipContentType(ct) { + http.Error(w, "plugin upload must be a .zip archive", http.StatusUnsupportedMediaType) + return + } + } + // Read the entire zip into memory — InstallFromZip needs an io.ReaderAt // for archive/zip and the cap is small enough to be safe. body, err := io.ReadAll(io.LimitReader(file, maxPluginUploadBytes+1)) @@ -75,6 +88,13 @@ func (h *PluginAdminHandler) install(w http.ResponseWriter, r *http.Request) { http.Error(w, "plugin upload too large", http.StatusRequestEntityTooLarge) return } + // Magic-byte check: a real .zip starts with "PK\x03\x04" (local file + // header) or "PK\x05\x06" (empty archive). Anything else is definitively + // not a zip regardless of what the client labelled it. + if !hasZipMagic(body) { + http.Error(w, "plugin upload is not a valid zip archive", http.StatusBadRequest) + return + } name, err := h.registry.InstallFromZip(r.Context(), body) if err != nil { slog.Error("plugin install failed", "error", err) @@ -153,6 +173,37 @@ func (h *PluginAdminHandler) uninstall(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusNoContent) } +// isZipContentType reports whether ct looks like a zip MIME type. Both the +// IANA-registered application/zip and the legacy application/x-zip-compressed +// (used by some Windows clients) are accepted. The comparison is case- +// insensitive and strips any parameters after a semicolon. +func isZipContentType(ct string) bool { + for i := 0; i < len(ct); i++ { + if ct[i] == ';' { + ct = ct[:i] + break + } + } + switch strings.ToLower(strings.TrimSpace(ct)) { + case "application/zip", "application/x-zip-compressed", "application/octet-stream": + return true + } + return false +} + +// hasZipMagic reports whether b begins with the PK signature used by every +// .zip archive. Empty archives use 0x50,0x4b,0x05,0x06; non-empty archives +// start with a local file header 0x50,0x4b,0x03,0x04. Both are accepted. +func hasZipMagic(b []byte) bool { + if len(b) < 4 { + return false + } + if b[0] != 'P' || b[1] != 'K' { + return false + } + return (b[2] == 0x03 && b[3] == 0x04) || (b[2] == 0x05 && b[3] == 0x06) +} + func parsePluginID(w http.ResponseWriter, r *http.Request) (int64, bool) { idStr := chi.URLParam(r, "id") id, err := strconv.ParseInt(idStr, 10, 64) diff --git a/Server/api/plugins_handler_test.go b/Server/api/plugins_handler_test.go new file mode 100644 index 00000000..cad708d0 --- /dev/null +++ b/Server/api/plugins_handler_test.go @@ -0,0 +1,241 @@ +// Phase C Step 9 — PluginAdminHandler tests. +// +// The handler is covered at the HTTP boundary so the fixtures do not depend +// on the Wazero runtime. A nil Registry exercises the "plugin runtime +// disabled" branch; a real Registry wired against a MemStore exercises the +// happy path. +package api + +import ( + "archive/zip" + "bytes" + "context" + "io" + "mime/multipart" + "net/http" + "net/http/httptest" + "path/filepath" + "strings" + "testing" + + "github.com/owncord/server/plugin" + "github.com/owncord/server/store" +) + +func TestPluginsHandlerListEmptyWhenRegistryNil(t *testing.T) { + h := NewPluginAdminHandler(nil, nil) + req := httptest.NewRequest("GET", "/", nil) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("status: got %d, want 200", rec.Code) + } + if strings.TrimSpace(rec.Body.String()) != "[]" { + t.Fatalf("expected empty JSON array, got %q", rec.Body.String()) + } +} + +func TestPluginsHandlerInstallRejectsWhenRegistryNil(t *testing.T) { + h := NewPluginAdminHandler(nil, nil) + body, contentType := buildZipUpload(t, validPluginZip(t)) + req := httptest.NewRequest("POST", "/install", body) + req.Header.Set("Content-Type", contentType) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusServiceUnavailable { + t.Fatalf("status: got %d, want 503", rec.Code) + } +} + +func TestPluginsHandlerInstallRejectsNonZipContentType(t *testing.T) { + reg := newTestPluginRegistry(t) + h := NewPluginAdminHandler(reg, nil) + + // Build a multipart body whose file part is labelled as text/plain. + var buf bytes.Buffer + mw := multipart.NewWriter(&buf) + partHeader := make(map[string][]string) + partHeader["Content-Disposition"] = []string{`form-data; name="plugin"; filename="evil.txt"`} + partHeader["Content-Type"] = []string{"text/plain"} + part, err := mw.CreatePart(partHeader) + if err != nil { + t.Fatal(err) + } + if _, err := part.Write(validPluginZip(t)); err != nil { + t.Fatal(err) + } + _ = mw.Close() + + req := httptest.NewRequest("POST", "/install", &buf) + req.Header.Set("Content-Type", mw.FormDataContentType()) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusUnsupportedMediaType { + t.Fatalf("status: got %d, want 415; body=%s", rec.Code, rec.Body.String()) + } +} + +func TestPluginsHandlerInstallRejectsNonZipMagic(t *testing.T) { + reg := newTestPluginRegistry(t) + h := NewPluginAdminHandler(reg, nil) + + body, contentType := buildZipUpload(t, []byte("this is definitely not a zip")) + req := httptest.NewRequest("POST", "/install", body) + req.Header.Set("Content-Type", contentType) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusBadRequest { + t.Fatalf("status: got %d, want 400; body=%s", rec.Code, rec.Body.String()) + } +} + +func TestPluginsHandlerInstallHappyPath(t *testing.T) { + reg := newTestPluginRegistry(t) + mem := store.NewMemStore() + // Wire the store into the handler so /list can show the new row. The + // registry already writes via its own PluginStore. + h := NewPluginAdminHandler(reg, mem) + body, contentType := buildZipUpload(t, validPluginZip(t)) + req := httptest.NewRequest("POST", "/install", body) + req.Header.Set("Content-Type", contentType) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusCreated { + t.Fatalf("status: got %d, want 201; body=%s", rec.Code, rec.Body.String()) + } + if !strings.Contains(rec.Body.String(), "hello") { + t.Fatalf("expected plugin name in response, got %q", rec.Body.String()) + } +} + +func TestPluginsHandlerEnableDisableUninstallReturn503WhenRegistryNil(t *testing.T) { + h := NewPluginAdminHandler(nil, nil) + for _, tc := range []struct{ method, path string }{ + {"POST", "/1/enable"}, + {"POST", "/1/disable"}, + {"DELETE", "/1"}, + } { + req := httptest.NewRequest(tc.method, tc.path, nil) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusServiceUnavailable { + t.Fatalf("%s %s status: got %d, want 503", tc.method, tc.path, rec.Code) + } + } +} + +func TestPluginsHandlerLifecycleInvalidID(t *testing.T) { + reg := newTestPluginRegistry(t) + h := NewPluginAdminHandler(reg, nil) + req := httptest.NewRequest("POST", "/not-an-int/enable", nil) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusBadRequest { + t.Fatalf("status: got %d, want 400", rec.Code) + } +} + +func TestIsZipContentType(t *testing.T) { + cases := map[string]bool{ + "application/zip": true, + "application/zip; charset=binary": true, + "APPLICATION/ZIP": true, + "application/x-zip-compressed": true, + "application/octet-stream": true, + "text/plain": false, + "image/png": false, + "": false, + "application/json; charset=utf-8": false, + } + for ct, want := range cases { + if got := isZipContentType(ct); got != want { + t.Errorf("isZipContentType(%q) = %v, want %v", ct, got, want) + } + } +} + +func TestHasZipMagic(t *testing.T) { + cases := map[string]bool{ + "PK\x03\x04rest": true, + "PK\x05\x06": true, + "PK\x07\x08rest": false, // spanned-archive signature; not accepted here + "not a zip": false, + "": false, + "PK": false, + } + for body, want := range cases { + if got := hasZipMagic([]byte(body)); got != want { + t.Errorf("hasZipMagic(%q) = %v, want %v", body, got, want) + } + } +} + +// ── helpers ──────────────────────────────────────────────────────────────── + +func newTestPluginRegistry(t *testing.T) *plugin.Registry { + t.Helper() + dir := t.TempDir() + mem := store.NewMemStore() + reg, err := plugin.NewRegistry(plugin.Config{ + Directory: filepath.Join(dir, "plugins"), + Store: mem, + }) + if err != nil { + t.Fatalf("plugin.NewRegistry: %v", err) + } + t.Cleanup(func() { _ = reg.Close(context.Background()) }) + return reg +} + +// validPluginZip returns a minimal but structurally valid plugin package: +// a plugin.json manifest at the root plus a near-empty hello.wasm that is +// large enough to pass the entrypoint stat but small enough to fly well +// under the zip-bomb cap. +func validPluginZip(t *testing.T) []byte { + t.Helper() + var buf bytes.Buffer + zw := zip.NewWriter(&buf) + mj, err := zw.Create("plugin.json") + if err != nil { + t.Fatal(err) + } + if _, err := mj.Write([]byte(`{"name":"hello","version":"0.1.0","entrypoint":"hello.wasm","permissions":["commands"]}`)); err != nil { + t.Fatal(err) + } + w, err := zw.Create("hello.wasm") + if err != nil { + t.Fatal(err) + } + // Minimal "placeholder" wasm magic bytes. Default build does not attempt + // to compile the module, so any bytes with the wasm magic suffice for + // InstallFromZip's on-disk validation. + if _, err := w.Write([]byte("\x00asm\x01\x00\x00\x00")); err != nil { + t.Fatal(err) + } + if err := zw.Close(); err != nil { + t.Fatal(err) + } + return buf.Bytes() +} + +// buildZipUpload wraps bodyBytes in a multipart form with a single "plugin" +// file part labelled as application/zip. +func buildZipUpload(t *testing.T, bodyBytes []byte) (io.Reader, string) { + t.Helper() + var buf bytes.Buffer + mw := multipart.NewWriter(&buf) + partHeader := make(map[string][]string) + partHeader["Content-Disposition"] = []string{`form-data; name="plugin"; filename="hello.zip"`} + partHeader["Content-Type"] = []string{"application/zip"} + part, err := mw.CreatePart(partHeader) + if err != nil { + t.Fatal(err) + } + if _, err := part.Write(bodyBytes); err != nil { + t.Fatal(err) + } + if err := mw.Close(); err != nil { + t.Fatal(err) + } + return &buf, mw.FormDataContentType() +} diff --git a/Server/plugin/registry.go b/Server/plugin/registry.go index 573c35cc..9daaed5c 100644 --- a/Server/plugin/registry.go +++ b/Server/plugin/registry.go @@ -48,10 +48,12 @@ type Registry struct { // Subscribe; the WS hub calls sink.Dispatch on each broadcast. sink *EventSink - // runtimePlatform is set by the wazero-tagged build's NewRegistry to a - // concrete *wazero.Runtime. The default build leaves it nil and falls - // back to manifest-only behaviour. + // runtimePlatform is populated by platformInit in the wazero-tagged build + // with a concrete *wazero.Runtime. The default build leaves it nil and + // falls back to manifest-only behaviour. platformClose tears the runtime + // down; both fields are set by platformInit atomically. runtimePlatform any + platformClose func(context.Context) error } // Instance is a single loaded plugin. @@ -81,20 +83,32 @@ func NewRegistry(cfg Config) (*Registry, error) { if cfg.Store == nil { return nil, fmt.Errorf("plugin: NewRegistry requires a non-nil PluginStore") } - return &Registry{ + r := &Registry{ cfg: cfg, plugins: make(map[int64]*Instance), byName: make(map[string]*Instance), commands: make(map[string]*Instance), sink: NewEventSink(), - }, nil + } + // platformInit is supplied by sandbox_default.go (no-op) or + // sandbox_wazero.go (real Wazero runtime). Either way it owns the + // runtimePlatform + platformClose pair on the Registry. + platform, closeFn, err := platformInit(cfg) + if err != nil { + return nil, fmt.Errorf("plugin: platform init: %w", err) + } + r.runtimePlatform = platform + r.platformClose = closeFn + return r, nil } // Close shuts the registry down. In the wazero-tagged build it tears the // runtime down and frees module memory. func (r *Registry) Close(ctx context.Context) error { r.mu.Lock() - defer r.mu.Unlock() + for _, inst := range r.plugins { + r.platformDeactivate(inst) + } for id := range r.plugins { delete(r.plugins, id) } @@ -105,6 +119,13 @@ func (r *Registry) Close(ctx context.Context) error { delete(r.commands, c) } r.uiTabs = nil + closeFn := r.platformClose + r.platformClose = nil + r.runtimePlatform = nil + r.mu.Unlock() + if closeFn != nil { + return closeFn(ctx) + } return nil } @@ -391,9 +412,21 @@ func (r *Registry) activateAll(ctx context.Context) error { // activate compiles and starts a single plugin module. Default build returns // ErrRuntimeUnavailable; the wazero-tagged build replaces this with the real -// implementation via the runtimePlatform field. +// implementation via activateWithRuntime. +// +// The runtimePlatform read is guarded by r.mu so a concurrent Close() that +// nil-s the field cannot be observed mid-activation. The captured platform +// value is then passed into activateWithRuntime as a parameter so the actual +// compile uses the snapshot rather than re-reading r.runtimePlatform — this +// closes the race window between the nil check and the wazero call. func (r *Registry) activate(ctx context.Context, inst *Instance) error { - return r.activateWithRuntime(ctx, inst) + r.mu.RLock() + platform := r.runtimePlatform + r.mu.RUnlock() + if platform == nil { + return ErrRuntimeUnavailable + } + return r.activateWithRuntime(ctx, platform, inst) } // EnablePlugin marks a plugin enabled in the store, then attempts to load it. @@ -422,7 +455,9 @@ func (r *Registry) EnablePlugin(ctx context.Context, id int64) error { return nil } -// DisablePlugin marks a plugin disabled and tears its module down. +// DisablePlugin marks a plugin disabled and tears its module down. The +// wazero-tagged build frees the compiled module via platformDeactivate so +// re-enabling recompiles from disk; the default build is a no-op. func (r *Registry) DisablePlugin(ctx context.Context, id int64) error { if err := r.cfg.Store.DisablePlugin(ctx, id); err != nil { return err @@ -437,6 +472,10 @@ func (r *Registry) DisablePlugin(ctx context.Context, id int64) error { delete(r.commands, cmd) } } + // Free the wazero module so memory is returned to the runtime + // immediately rather than waiting for registry Close. Safe to call + // on an instance that was never activated. + r.platformDeactivate(inst) } return nil } diff --git a/Server/plugin/sandbox_default.go b/Server/plugin/sandbox_default.go index 1ae343d9..8bb7a506 100644 --- a/Server/plugin/sandbox_default.go +++ b/Server/plugin/sandbox_default.go @@ -9,15 +9,25 @@ import ( "context" ) +// platformInit is a no-op in the default build — there is no Wazero runtime +// to stand up, and NewRegistry leaves runtimePlatform nil so the lifecycle +// methods fall through to ErrRuntimeUnavailable. +func platformInit(_ Config) (any, func(context.Context) error, error) { + return nil, nil, nil +} + // activateWithRuntime is a no-op in the default build. It is only called from // Registry.activate when runtimePlatform is non-nil, which never happens here. -func (r *Registry) activateWithRuntime(ctx context.Context, inst *Instance) error { +func (r *Registry) activateWithRuntime(_ context.Context, _ any, _ *Instance) error { return ErrRuntimeUnavailable } +// platformDeactivate is called from Close on each plugin; a no-op here. +func (r *Registry) platformDeactivate(_ *Instance) {} + // invokeCommand returns an error result instructing the operator to enable // the wazero build tag. Default build only. -func (r *Registry) invokeCommand(ctx context.Context, inst *Instance, userID, channelID int64, cmd string, args []string) (*CommandResult, bool) { +func (r *Registry) invokeCommand(_ context.Context, _ *Instance, _, _ int64, _ string, _ []string) (*CommandResult, bool) { return &CommandResult{ Reply: "plugin runtime disabled — rebuild server with -tags wazero to execute plugin commands", }, true diff --git a/Server/plugin/sandbox_wazero.go b/Server/plugin/sandbox_wazero.go index 92f26a8e..f1710059 100644 --- a/Server/plugin/sandbox_wazero.go +++ b/Server/plugin/sandbox_wazero.go @@ -1,8 +1,34 @@ //go:build wazero -// Real Wazero-backed plugin runtime. Compiled only with `-tags wazero`. -// Provides concrete implementations of activateWithRuntime and invokeCommand -// that compile, instantiate, and dispatch to WASM modules via wazero v1. +// Phase C Step 9 — Real Wazero-backed plugin runtime. Compiled only with +// `-tags wazero`; matches the postgres / otel build-tag pattern used +// elsewhere in the repo so the default sqlite-only build does not pull +// wazero into go.mod at runtime. +// +// Architecture +// +// The wazero-tagged build provides: +// +// platformInit — creates the shared wazero.Runtime (with the +// configured memory cap and WASI preview-1 imports) +// and returns a teardown closure consumed by +// Registry.Close. +// activateWithRuntime — compiles the plugin's .wasm entrypoint, instantiates +// it against the shared runtime, stores the resulting +// api.Module on the Instance, and auto-registers any +// commands the module exports via list_commands. +// platformDeactivate — closes the per-plugin module without tearing down +// the shared runtime; called by DisablePlugin so a +// disabled plugin frees memory immediately. +// invokeCommand — calls the plugin's `command_dispatch` export using +// the JSON-ABI: +// allocate(size u32) → ptr u32 +// command_dispatch(ptr u32, len u32) → (ptr u32, len u32) +// deallocate(ptr u32, len u32) +// Both input and output payloads are JSON. +// +// Plugins that do not export command_dispatch / allocate are still loadable +// — DispatchCommand reports a user-facing diagnostic instead of crashing. package plugin import ( @@ -20,65 +46,72 @@ import ( // wazeroPageBytes is the size of a single WASM linear-memory page (64 KiB). const wazeroPageBytes = 65536 -// ensureRuntimeLocked initialises the shared wazero.Runtime on first call. -// The caller MUST hold r.mu.Lock(). -func (r *Registry) ensureRuntimeLocked(ctx context.Context) (wazero.Runtime, error) { - if r.runtimePlatform != nil { - rt, ok := r.runtimePlatform.(wazero.Runtime) - if !ok { - return nil, fmt.Errorf("plugin: runtimePlatform is not a wazero.Runtime") - } - return rt, nil - } +// platformInit stands up the shared wazero runtime for this Registry. The +// runtime is the top-level handle that owns compiled modules, host modules, +// and per-instance linear memory; every plugin in this registry shares it. +// +// The configured MaxMemoryMB cap is translated to wazero's per-page memory +// limit (64 KiB pages) and applied at runtime construction. WASI preview-1 +// is pre-instantiated so plugins compiled with the standard TinyGo / Rust +// wasm32-wasi targets can resolve their syscall imports. +func platformInit(cfg Config) (any, func(context.Context) error, error) { + ctx := context.Background() - // Convert the configured MB limit to WASM pages (64 KiB each). - memMB := r.cfg.MaxMemoryMB + memMB := cfg.MaxMemoryMB if memMB <= 0 { - memMB = 64 // default 64 MiB + memMB = 64 // default 64 MiB per plugin runtime } memPages := uint32(memMB) * 1024 * 1024 / wazeroPageBytes - rtCfg := wazero.NewRuntimeConfig().WithMemoryLimitPages(memPages) - rt := wazero.NewRuntimeWithConfig(ctx, rtCfg) - - // Instantiate the WASI host module so plugins compiled with the WASI - // target (TinyGo, wasm32-wasi, etc.) have their syscall shims available. + rt := wazero.NewRuntimeWithConfig(ctx, + wazero.NewRuntimeConfig(). + WithMemoryLimitPages(memPages). + WithCloseOnContextDone(true), + ) if _, err := wasi_snapshot_preview1.Instantiate(ctx, rt); err != nil { _ = rt.Close(ctx) - return nil, fmt.Errorf("plugin: wasi instantiation: %w", err) + return nil, nil, fmt.Errorf("wazero: wasi snapshot_preview1: %w", err) } - - r.runtimePlatform = rt - return rt, nil + closeFn := func(shutCtx context.Context) error { + return rt.Close(shutCtx) + } + return rt, closeFn, nil } -// activateWithRuntime compiles inst.WASMPath into a wazero module, instantiates -// it, and wires up the command dispatch table for any commands the module exports. -func (r *Registry) activateWithRuntime(ctx context.Context, inst *Instance) error { +// activateWithRuntime compiles inst.WASMPath into a wazero module and +// instantiates it under the shared runtime. The resulting api.Module is +// stashed on inst.module so lifecycle teardown (Close, DisablePlugin) can +// free it without walking the registry again. +// +// The runtime is passed in by Registry.activate as a captured snapshot so +// this function never re-reads r.runtimePlatform — that field can be nil-ed +// concurrently by Close, but the snapshot remains valid (the wazero runtime +// itself returns an error gracefully if it has been closed underneath us). +func (r *Registry) activateWithRuntime(ctx context.Context, platform any, inst *Instance) error { + rt, ok := platform.(wazero.Runtime) + if !ok || rt == nil { + return fmt.Errorf("plugin %q: wazero runtime unavailable", inst.Manifest.Name) + } wasmBytes, err := os.ReadFile(inst.WASMPath) if err != nil { return fmt.Errorf("plugin %q: read wasm: %w", inst.Manifest.Name, err) } - // Ensure the shared runtime is ready (lazy init, holds lock only briefly). - r.mu.Lock() - rt, err := r.ensureRuntimeLocked(ctx) - r.mu.Unlock() - if err != nil { - return err - } - - // CompileModule is CPU-intensive but requires no lock. + // CompileModule is CPU-bound; do it without holding the registry lock. compiled, err := rt.CompileModule(ctx, wasmBytes) if err != nil { return fmt.Errorf("plugin %q: compile: %w", inst.Manifest.Name, err) } + // Each plugin gets its own module name so multiple instances can coexist + // without colliding in the runtime's global module namespace. Output is + // swallowed so a misbehaving plugin can't flood server logs. _start is + // suppressed so exports are only invoked on demand. modCfg := wazero.NewModuleConfig(). WithName(inst.Manifest.Name). WithStdout(io.Discard). WithStderr(io.Discard). - WithStartFunctions() // suppress _start; exports are called on demand + WithStartFunctions() module, err := rt.InstantiateModule(ctx, compiled, modCfg) if err != nil { @@ -86,31 +119,46 @@ func (r *Registry) activateWithRuntime(ctx context.Context, inst *Instance) erro return fmt.Errorf("plugin %q: instantiate: %w", inst.Manifest.Name, err) } - // Register the module and bind any commands it exports. + // Register the module and auto-bind any commands the plugin exports + // via list_commands. The plugin must also have declared the `commands` + // capability in its manifest, otherwise no binding happens. r.mu.Lock() inst.module = module - for _, perm := range inst.Manifest.Permissions { - if Capability(perm) == CapCommands { - for _, cmd := range listExportedCommands(ctx, module) { - r.commands[cmd] = inst - } - break + if inst.Manifest.HasCapability(CapCommands) { + for _, cmd := range listExportedCommands(ctx, module) { + r.commands[cmd] = inst } } r.mu.Unlock() return nil } -// invokeCommand calls the plugin's exported "command_dispatch" function using -// a simple linear-memory ABI: +// platformDeactivate closes the wazero module held by inst without touching +// the shared runtime. Safe to call on an instance that was never activated. +// Called from DisablePlugin and Close. +func (r *Registry) platformDeactivate(inst *Instance) { + if inst == nil || inst.module == nil { + return + } + if mod, ok := inst.module.(api.Module); ok { + _ = mod.Close(context.Background()) + } + inst.module = nil +} + +// invokeCommand calls the plugin's exported `command_dispatch` function +// using a small JSON ABI: // // allocate(size u32) → ptr u32 // command_dispatch(ptr u32, len u32) → (result_ptr u32, result_len u32) // deallocate(ptr u32, len u32) // -// Both the input and output are JSON payloads. +// Both the input and output payloads are JSON. A plugin that does not +// export command_dispatch returns (nil, false) so the WS dispatcher can +// fall back to the not-found response. A plugin that exports +// command_dispatch but lacks allocate returns a user-facing diagnostic. func (r *Registry) invokeCommand(ctx context.Context, inst *Instance, userID, channelID int64, cmd string, args []string) (*CommandResult, bool) { - if inst.module == nil { + if inst == nil || inst.module == nil { return nil, false } mod, ok := inst.module.(api.Module) @@ -120,7 +168,9 @@ func (r *Registry) invokeCommand(ctx context.Context, inst *Instance, userID, ch dispatchFn := mod.ExportedFunction("command_dispatch") if dispatchFn == nil { - return nil, false + return &CommandResult{ + Reply: fmt.Sprintf("plugin %q does not export command_dispatch (rebuild the plugin to handle /%s)", inst.Manifest.Name, cmd), + }, true } allocFn := mod.ExportedFunction("allocate") deallocFn := mod.ExportedFunction("deallocate") @@ -187,9 +237,10 @@ func (r *Registry) invokeCommand(ctx context.Context, inst *Instance, userID, ch return &CommandResult{Reply: dr.Reply}, true } -// listExportedCommands calls the plugin's optional "list_commands" export which -// returns (ptr u32, len u32) pointing to a JSON array of command name strings. -// If the export is absent or returns invalid JSON, an empty slice is returned. +// listExportedCommands calls the plugin's optional `list_commands` export +// which returns (ptr u32, len u32) pointing to a JSON array of command name +// strings. If the export is absent or returns invalid JSON, an empty slice +// is returned and no command bindings are created. func listExportedCommands(ctx context.Context, mod api.Module) []string { fn := mod.ExportedFunction("list_commands") if fn == nil { diff --git a/Server/plugin/sandbox_wazero_test.go b/Server/plugin/sandbox_wazero_test.go new file mode 100644 index 00000000..1309559b --- /dev/null +++ b/Server/plugin/sandbox_wazero_test.go @@ -0,0 +1,228 @@ +//go:build wazero + +// Phase C Step 9 — Wazero runtime integration tests. Only compiled with +// `-tags wazero`, alongside sandbox_wazero.go. These tests exercise the +// behaviours the default build cannot: +// +// - NewRegistry stands up a real wazero.Runtime +// - activateWithRuntime compiles a real .wasm module and tracks the +// resulting instance on the Instance struct +// - EnablePlugin takes a manifest from a PluginStore row through +// activation end-to-end +// - invokeCommand gracefully returns a user-facing error when the plugin +// does not export command_dispatch +// - Close tears the runtime down without panicking +// +// Test fixture: the 41-byte `add.wasm` module from the wazero examples, a +// trivial module that exports `add(i32,i32) -> i32`. It does NOT export +// `command_dispatch`, which is intentional — the command path must handle +// that case. +package plugin + +import ( + "context" + "os" + "path/filepath" + "testing" + + "github.com/owncord/server/store" +) + +// addWASM is the bytes of a minimal (module (func (export "add") ... )). +// Verified against the wazero examples fixture; 41 bytes. Using a literal +// here avoids dragging a binary asset into the repo. +var addWASM = []byte{ + 0x00, 0x61, 0x73, 0x6d, 0x01, 0x00, 0x00, 0x00, + 0x01, 0x07, 0x01, 0x60, 0x02, 0x7f, 0x7f, 0x01, + 0x7f, 0x03, 0x02, 0x01, 0x00, 0x07, 0x07, 0x01, + 0x03, 0x61, 0x64, 0x64, 0x00, 0x00, 0x0a, 0x09, + 0x01, 0x07, 0x00, 0x20, 0x00, 0x20, 0x01, 0x6a, + 0x0b, +} + +func writeTestPlugin(t *testing.T, root, name string, manifest string, wasmBytes []byte) { + t.Helper() + pluginDir := filepath.Join(root, name) + if err := os.MkdirAll(pluginDir, 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(pluginDir, "plugin.json"), []byte(manifest), 0o644); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(pluginDir, "hello.wasm"), wasmBytes, 0o644); err != nil { + t.Fatal(err) + } +} + +func newWazeroTestRegistry(t *testing.T, dir string) (*Registry, store.PluginStore) { + t.Helper() + mem := store.NewMemStore() + reg, err := NewRegistry(Config{ + Directory: dir, + MaxMemoryMB: 16, + CPUBudgetMs: 100, + Store: mem, + }) + if err != nil { + t.Fatalf("NewRegistry: %v", err) + } + t.Cleanup(func() { _ = reg.Close(context.Background()) }) + return reg, mem +} + +func TestWazeroRegistryCreatesRuntime(t *testing.T) { + reg, _ := newWazeroTestRegistry(t, t.TempDir()) + if reg.runtimePlatform == nil { + t.Fatal("expected wazero runtime to be wired in tagged build") + } + if reg.platformClose == nil { + t.Fatal("expected platformClose to be set") + } +} + +func TestWazeroActivateCompilesModule(t *testing.T) { + dir := t.TempDir() + manifest := `{"name":"hello","version":"0.1.0","entrypoint":"hello.wasm","permissions":["commands"]}` + writeTestPlugin(t, dir, "hello", manifest, addWASM) + + reg, mem := newWazeroTestRegistry(t, dir) + ctx := context.Background() + if err := reg.LoadAll(ctx); err != nil { + t.Fatalf("LoadAll: %v", err) + } + + rows, err := mem.ListPlugins(ctx) + if err != nil || len(rows) != 1 { + t.Fatalf("ListPlugins: rows=%+v err=%v", rows, err) + } + + // The plugin starts disabled; enable it and confirm the module compiles. + if err := reg.EnablePlugin(ctx, rows[0].ID); err != nil { + t.Fatalf("EnablePlugin: %v", err) + } + + reg.mu.RLock() + inst := reg.plugins[rows[0].ID] + reg.mu.RUnlock() + if inst == nil { + t.Fatal("instance not registered after enable") + } + if !inst.Enabled { + t.Fatal("instance should be enabled after EnablePlugin") + } + if inst.module == nil { + t.Fatal("expected inst.module to be populated after activation") + } +} + +func TestWazeroDispatchCommandMissingExport(t *testing.T) { + dir := t.TempDir() + manifest := `{"name":"hello","version":"0.1.0","entrypoint":"hello.wasm","permissions":["commands"]}` + writeTestPlugin(t, dir, "hello", manifest, addWASM) + + reg, mem := newWazeroTestRegistry(t, dir) + ctx := context.Background() + if err := reg.LoadAll(ctx); err != nil { + t.Fatal(err) + } + rows, _ := mem.ListPlugins(ctx) + if err := reg.EnablePlugin(ctx, rows[0].ID); err != nil { + t.Fatalf("EnablePlugin: %v", err) + } + + reg.mu.RLock() + inst := reg.plugins[rows[0].ID] + reg.mu.RUnlock() + if err := reg.RegisterCommand("hello", inst); err != nil { + t.Fatalf("RegisterCommand: %v", err) + } + + result, ok := reg.DispatchCommand(ctx, 1, 2, "hello", nil) + if !ok { + t.Fatal("expected DispatchCommand to return a result") + } + if result == nil || result.Reply == "" { + t.Fatal("expected a non-empty reply when export is missing") + } +} + +func TestWazeroCloseTearsDownRuntime(t *testing.T) { + dir := t.TempDir() + manifest := `{"name":"hello","version":"0.1.0","entrypoint":"hello.wasm","permissions":["commands"]}` + writeTestPlugin(t, dir, "hello", manifest, addWASM) + + reg, mem := newWazeroTestRegistry(t, dir) + ctx := context.Background() + if err := reg.LoadAll(ctx); err != nil { + t.Fatal(err) + } + rows, _ := mem.ListPlugins(ctx) + if err := reg.EnablePlugin(ctx, rows[0].ID); err != nil { + t.Fatalf("EnablePlugin: %v", err) + } + + if err := reg.Close(ctx); err != nil { + t.Fatalf("Close: %v", err) + } + if reg.runtimePlatform != nil || reg.platformClose != nil { + t.Fatal("Close should clear platform fields") + } + // Calling Close twice must not panic. + if err := reg.Close(ctx); err != nil { + t.Fatalf("Close (second): %v", err) + } +} + +func TestWazeroDisablePluginFreesModule(t *testing.T) { + dir := t.TempDir() + manifest := `{"name":"hello","version":"0.1.0","entrypoint":"hello.wasm","permissions":["commands"]}` + writeTestPlugin(t, dir, "hello", manifest, addWASM) + + reg, mem := newWazeroTestRegistry(t, dir) + ctx := context.Background() + if err := reg.LoadAll(ctx); err != nil { + t.Fatal(err) + } + rows, _ := mem.ListPlugins(ctx) + if err := reg.EnablePlugin(ctx, rows[0].ID); err != nil { + t.Fatalf("EnablePlugin: %v", err) + } + reg.mu.RLock() + inst := reg.plugins[rows[0].ID] + reg.mu.RUnlock() + if inst.module == nil { + t.Fatal("pre-condition: module should be populated after EnablePlugin") + } + if err := reg.DisablePlugin(ctx, rows[0].ID); err != nil { + t.Fatalf("DisablePlugin: %v", err) + } + if inst.module != nil { + t.Fatal("DisablePlugin must release the wazero module (inst.module != nil)") + } + if inst.Enabled { + t.Fatal("DisablePlugin must clear inst.Enabled") + } + // Re-enabling must rebuild a new module. + if err := reg.EnablePlugin(ctx, rows[0].ID); err != nil { + t.Fatalf("re-EnablePlugin: %v", err) + } + if inst.module == nil { + t.Fatal("re-Enable should repopulate inst.module") + } +} + +func TestWazeroInvalidWASMFailsActivation(t *testing.T) { + dir := t.TempDir() + manifest := `{"name":"brokey","version":"0.1.0","entrypoint":"hello.wasm","permissions":["commands"]}` + writeTestPlugin(t, dir, "brokey", manifest, []byte("not a wasm")) + + reg, mem := newWazeroTestRegistry(t, dir) + ctx := context.Background() + if err := reg.LoadAll(ctx); err != nil { + t.Fatal(err) + } + rows, _ := mem.ListPlugins(ctx) + if err := reg.EnablePlugin(ctx, rows[0].ID); err == nil { + t.Fatal("expected EnablePlugin to fail on invalid WASM") + } +} diff --git a/Server/telemetry/metrics.go b/Server/telemetry/metrics.go index 39aeecc8..799fa92c 100644 --- a/Server/telemetry/metrics.go +++ b/Server/telemetry/metrics.go @@ -34,33 +34,46 @@ type AppMetrics struct { } var ( - appMetricsOnce sync.Once + appMetricsMu sync.Mutex appMetricsInst *AppMetrics ) // NewAppMetrics returns a process-wide AppMetrics, lazily constructed against // the current global provider. Calling it multiple times returns the same -// instance — the metrics are tied to the global provider, not to a specific -// caller. +// instance until resetAppMetricsForInit() is called (which Init uses after +// swapping the global provider so instruments re-bind to the new meter). func NewAppMetrics() *AppMetrics { - appMetricsOnce.Do(func() { - ws := GlobalMeter(scopeWS) - svc := GlobalMeter(scopeService) - db := GlobalMeter(scopeDB) - voice := GlobalMeter(scopeVoice) - appMetricsInst = &AppMetrics{ - WSMessagesTotal: ws.Counter("ws_messages_total", "WebSocket messages broadcast"), - WSActiveConnections: ws.Gauge("ws_active_connections", "Currently connected WebSocket clients"), - WSBroadcastLatency: ws.Histogram("ws_broadcast_latency_seconds", "Wall-clock seconds from enqueue to fanout completion", "s"), - WSReconnectTierTotal: ws.Counter("ws_reconnect_tier_total", "Reconnection replay tier hits, attribute tier=buffer|db|full"), - WSEventsPersisted: ws.Counter("ws_events_persisted_total", "Events written to the cold-tier event log"), - WSEventsDropped: ws.Counter("ws_events_dropped_total", "Events dropped because the persister queue was full"), - WSEventsPersistErrors: ws.Counter("ws_events_persist_errors_total", "PersistEvent calls that returned an error from the underlying store"), - DBQueryDurationSec: db.Histogram("db_query_duration_seconds", "Per-query wall time", "s"), - VoiceActiveSessions: voice.Gauge("voice_active_sessions", "Active LiveKit rooms"), - VoiceParticipants: voice.Gauge("voice_participants", "Connected LiveKit participants across all rooms"), - ServiceCallDurationSec: svc.Histogram("service_call_duration_seconds", "Service-layer method execution time", "s"), - } - }) + appMetricsMu.Lock() + defer appMetricsMu.Unlock() + if appMetricsInst != nil { + return appMetricsInst + } + ws := GlobalMeter(scopeWS) + svc := GlobalMeter(scopeService) + db := GlobalMeter(scopeDB) + voice := GlobalMeter(scopeVoice) + appMetricsInst = &AppMetrics{ + WSMessagesTotal: ws.Counter("ws_messages_total", "WebSocket messages broadcast"), + WSActiveConnections: ws.Gauge("ws_active_connections", "Currently connected WebSocket clients"), + WSBroadcastLatency: ws.Histogram("ws_broadcast_latency_seconds", "Wall-clock seconds from enqueue to fanout completion", "s"), + WSReconnectTierTotal: ws.Counter("ws_reconnect_tier_total", "Reconnection replay tier hits, attribute tier=buffer|db|full"), + WSEventsPersisted: ws.Counter("ws_events_persisted_total", "Events written to the cold-tier event log"), + WSEventsDropped: ws.Counter("ws_events_dropped_total", "Events dropped because the persister queue was full"), + WSEventsPersistErrors: ws.Counter("ws_events_persist_errors_total", "PersistEvent calls that returned an error from the underlying store"), + DBQueryDurationSec: db.Histogram("db_query_duration_seconds", "Per-query wall time", "s"), + VoiceActiveSessions: voice.Gauge("voice_active_sessions", "Active LiveKit rooms"), + VoiceParticipants: voice.Gauge("voice_participants", "Connected LiveKit participants across all rooms"), + ServiceCallDurationSec: svc.Histogram("service_call_duration_seconds", "Service-layer method execution time", "s"), + } return appMetricsInst } + +// resetAppMetricsForInit drops the cached AppMetrics bundle so the next +// NewAppMetrics() call re-binds instruments against whatever provider is now +// global. The real OTel Init uses this to migrate from the no-op provider +// installed by the package-level init() to the SDK-backed one. +func resetAppMetricsForInit() { + appMetricsMu.Lock() + defer appMetricsMu.Unlock() + appMetricsInst = nil +} diff --git a/Server/telemetry/telemetry_otel.go b/Server/telemetry/telemetry_otel.go index 33941948..b4b78822 100644 --- a/Server/telemetry/telemetry_otel.go +++ b/Server/telemetry/telemetry_otel.go @@ -1,239 +1,296 @@ //go:build otel -// Real OpenTelemetry-backed implementation. Compiled only with `-tags otel`. -// Replaces the no-op providers in telemetry_default.go for the duration of -// the process. +// Phase B Step 8 — Real OpenTelemetry-backed implementation. Compiled only +// with `-tags otel`, matching the postgres / wazero build-tag pattern used +// elsewhere in the repo. The default build ships telemetry_default.go with a +// no-op provider so sqlite-only binaries do not pull the OTel SDK in. +// +// Exporter modes (config.TelemetryConfig.Exporter): +// +// "none" — no-op provider; same as the default build. +// "prometheus" — pull-based metrics via a /metrics handler, spans are still +// processed by a tracer provider with no exporter. +// "otlp" — push-based traces over OTLP/gRPC to cfg.OTLPEndpoint AND +// pull-based metrics via the Prometheus exporter (operators +// typically want both). +// +// The concrete Provider adapts our tiny telemetry.* API onto the upstream +// OTel SDK so the rest of the codebase never depends on the SDK directly. package telemetry import ( "context" + "errors" "fmt" + "math" "net/http" + "sync" - "github.com/prometheus/client_golang/prometheus" + "github.com/owncord/server/config" + + promclient "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promhttp" + "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" - otlpgrpc "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc" - promexp "go.opentelemetry.io/otel/exporters/prometheus" - otelmetric "go.opentelemetry.io/otel/metric" + "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc" + otelprom "go.opentelemetry.io/otel/exporters/prometheus" + "go.opentelemetry.io/otel/metric" sdkmetric "go.opentelemetry.io/otel/sdk/metric" - sdkresource "go.opentelemetry.io/otel/sdk/resource" + "go.opentelemetry.io/otel/sdk/resource" sdktrace "go.opentelemetry.io/otel/sdk/trace" - oteltrace "go.opentelemetry.io/otel/trace" - - "github.com/owncord/server/config" + semconv "go.opentelemetry.io/otel/semconv/v1.40.0" + "go.opentelemetry.io/otel/trace" ) -// otelProvider is the real OTel-backed Provider, active only in the otel build. +// otelProvider adapts the OTel SDK to the telemetry.Provider interface. type otelProvider struct { - tp oteltrace.TracerProvider - mp otelmetric.MeterProvider + cfg config.TelemetryConfig promHandler http.Handler - mw func(http.Handler) http.Handler - shutdowns []func(context.Context) error + tp *sdktrace.TracerProvider + mp *sdkmetric.MeterProvider + serviceName string } -// Init wires the OTel SDK according to cfg.Exporter and installs the result -// as the global telemetry provider. The returned ShutdownFunc must be called -// on server shutdown to flush pending telemetry. +// Init wires the OTel SDK according to cfg and installs it as the global +// Provider. The returned ShutdownFunc flushes the batchers and releases +// exporter resources; it is safe to call more than once. func Init(ctx context.Context, cfg config.TelemetryConfig) (ShutdownFunc, error) { if !cfg.Enabled || cfg.Exporter == "" || cfg.Exporter == "none" { SetGlobal(noopProvider{}) return func(context.Context) error { return nil }, nil } - res, err := sdkresource.New(ctx, - sdkresource.WithAttributes(attribute.String("service.name", cfg.ServiceName)), - sdkresource.WithFromEnv(), + svcName := cfg.ServiceName + if svcName == "" { + svcName = "owncord-server" + } + + res, err := resource.New(ctx, + resource.WithAttributes( + semconv.ServiceName(svcName), + ), ) if err != nil { - // Non-fatal: fall back to the SDK default resource. - res = sdkresource.Default() + return nil, fmt.Errorf("telemetry: resource: %w", err) } - p := &otelProvider{} + provider := &otelProvider{cfg: cfg, serviceName: svcName} - switch cfg.Exporter { - case "prometheus": - promReg := prometheus.NewRegistry() - exporter, expErr := promexp.New(promexp.WithRegisterer(promReg)) - if expErr != nil { - return nil, fmt.Errorf("telemetry: prometheus exporter: %w", expErr) + // ── Tracing ──────────────────────────────────────────────────────────── + tpOpts := []sdktrace.TracerProviderOption{sdktrace.WithResource(res)} + if cfg.Exporter == "otlp" { + endpoint := cfg.OTLPEndpoint + if endpoint == "" { + endpoint = "localhost:4317" } - mp := sdkmetric.NewMeterProvider( - sdkmetric.WithResource(res), - sdkmetric.WithReader(exporter), - ) - otel.SetMeterProvider(mp) - p.mp = mp - p.promHandler = promhttp.HandlerFor(promReg, promhttp.HandlerOpts{EnableOpenMetrics: true}) - p.shutdowns = append(p.shutdowns, mp.Shutdown) - - case "otlp": - if cfg.OTLPEndpoint == "" { - return nil, fmt.Errorf("telemetry: exporter=otlp requires otlp_endpoint to be set") - } - otlpOpts := []otlpgrpc.Option{ - otlpgrpc.WithEndpoint(cfg.OTLPEndpoint), + otlpOpts := []otlptracegrpc.Option{ + otlptracegrpc.WithEndpoint(endpoint), } + // OTLPInsecure honours the explicit operator opt-in for plaintext + // gRPC; production deployments should leave it false and provide + // a TLS endpoint. if cfg.OTLPInsecure { - otlpOpts = append(otlpOpts, otlpgrpc.WithInsecure()) + otlpOpts = append(otlpOpts, otlptracegrpc.WithInsecure()) } - exp, expErr := otlpgrpc.New(ctx, otlpOpts...) - if expErr != nil { - return nil, fmt.Errorf("telemetry: otlp exporter: %w", expErr) + traceExp, err := otlptracegrpc.New(ctx, otlpOpts...) + if err != nil { + return nil, fmt.Errorf("telemetry: otlp trace exporter: %w", err) } - tp := sdktrace.NewTracerProvider( - sdktrace.WithBatcher(exp), - sdktrace.WithResource(res), - ) - otel.SetTracerProvider(tp) - p.tp = tp - p.shutdowns = append(p.shutdowns, tp.Shutdown) - - default: - return nil, fmt.Errorf("telemetry: unknown exporter %q (valid: none, prometheus, otlp)", cfg.Exporter) + tpOpts = append(tpOpts, sdktrace.WithBatcher(traceExp)) } + provider.tp = sdktrace.NewTracerProvider(tpOpts...) + otel.SetTracerProvider(provider.tp) - // HTTP tracing middleware using otelhttp (works with any http.Handler router). - p.mw = func(next http.Handler) http.Handler { - return otelhttp.NewHandler(next, cfg.ServiceName) + // ── Metrics ──────────────────────────────────────────────────────────── + // Both prometheus and otlp modes surface metrics via a pull-based + // Prometheus endpoint. OTLP tracing does not preclude Prometheus metrics + // — operators usually want both — so we wire the exporter unconditionally + // for the two real modes. + reg := promclient.NewRegistry() + promExp, err := otelprom.New(otelprom.WithRegisterer(reg)) + if err != nil { + // Shut the trace provider down so the OTLP exporter's gRPC + // connection (if any) is torn down. Init returning an error must + // not leak resources. + _ = provider.tp.Shutdown(ctx) + return nil, fmt.Errorf("telemetry: prometheus exporter: %w", err) } + provider.mp = sdkmetric.NewMeterProvider( + sdkmetric.WithResource(res), + sdkmetric.WithReader(promExp), + ) + otel.SetMeterProvider(provider.mp) + provider.promHandler = promhttp.HandlerFor(reg, promhttp.HandlerOpts{}) - SetGlobal(p) - return p.shutdown, nil + // Drop the cached AppMetrics bundle BEFORE publishing the new global + // provider. A concurrent NewAppMetrics() caller that observes the old + // no-op provider and the stale cache is harmless (it just re-populates + // once); the failure mode we avoid is a caller seeing the new provider + // but still reading the old cached (no-op) instruments. + resetAppMetricsForInit() + SetGlobal(provider) + + var once sync.Once + return func(shutCtx context.Context) error { + var rerr error + once.Do(func() { + var errs []error + if provider.tp != nil { + if err := provider.tp.Shutdown(shutCtx); err != nil { + errs = append(errs, err) + } + } + if provider.mp != nil { + if err := provider.mp.Shutdown(shutCtx); err != nil { + errs = append(errs, err) + } + } + if len(errs) > 0 { + rerr = fmt.Errorf("telemetry shutdown: %w", errors.Join(errs...)) + } + SetGlobal(noopProvider{}) + }) + return rerr + }, nil } -func (p *otelProvider) shutdown(ctx context.Context) error { - var first error - for _, fn := range p.shutdowns { - if err := fn(ctx); err != nil && first == nil { - first = err - } - } - return first -} - -// ── Provider interface ────────────────────────────────────────────────────── - +// Tracer returns an otel-backed tracer for the given instrumentation scope. func (p *otelProvider) Tracer(name string) Tracer { - tp := p.tp - if tp == nil { - tp = otel.GetTracerProvider() - } - return otelTracerBridge{t: tp.Tracer(name)} + return &otelTracer{inner: p.tp.Tracer(name)} } +// Meter returns an otel-backed meter for the given instrumentation scope. func (p *otelProvider) Meter(name string) Meter { - mp := p.mp - if mp == nil { - mp = otel.GetMeterProvider() - } - return otelMeterBridge{m: mp.Meter(name)} + return &otelMeter{inner: p.mp.Meter(name)} } +// HTTPMiddleware wraps next with otelhttp so every REST request becomes a +// span named after its route pattern. func (p *otelProvider) HTTPMiddleware(next http.Handler) http.Handler { - if p.mw == nil { - return next - } - return p.mw(next) + return otelhttp.NewHandler(next, "http.server", + otelhttp.WithServerName(p.serviceName), + ) } +// PrometheusHandler returns the /metrics handler backed by the active +// exporter registry. func (p *otelProvider) PrometheusHandler() http.Handler { return p.promHandler } -// ── Bridge: Tracer / Span ─────────────────────────────────────────────────── +// ── Tracer / Span adapters ───────────────────────────────────────────────── -type otelTracerBridge struct{ t oteltrace.Tracer } +type otelTracer struct{ inner trace.Tracer } -func (b otelTracerBridge) Start(ctx context.Context, name string, attrs ...Attr) (context.Context, Span) { - var opts []oteltrace.SpanStartOption - if len(attrs) > 0 { - opts = append(opts, oteltrace.WithAttributes(toKVs(attrs)...)) - } - ctx, span := b.t.Start(ctx, name, opts...) - return ctx, otelSpanBridge{s: span} +func (t *otelTracer) Start(ctx context.Context, name string, attrs ...Attr) (context.Context, Span) { + ctx, span := t.inner.Start(ctx, name, trace.WithAttributes(convertAttrs(attrs)...)) + return ctx, &otelSpan{inner: span} } -type otelSpanBridge struct{ s oteltrace.Span } +type otelSpan struct{ inner trace.Span } -func (b otelSpanBridge) End() { b.s.End() } -func (b otelSpanBridge) SetAttributes(attrs ...Attr) { b.s.SetAttributes(toKVs(attrs)...) } -func (b otelSpanBridge) RecordError(err error) { b.s.RecordError(err) } +func (s *otelSpan) End() { s.inner.End() } +func (s *otelSpan) SetAttributes(attrs ...Attr) { s.inner.SetAttributes(convertAttrs(attrs)...) } +func (s *otelSpan) RecordError(err error) { + if err == nil { + return + } + s.inner.RecordError(err) +} -// ── Bridge: Meter / instruments ───────────────────────────────────────────── +// ── Meter / instrument adapters ──────────────────────────────────────────── -type otelMeterBridge struct{ m otelmetric.Meter } +type otelMeter struct{ inner metric.Meter } -func (b otelMeterBridge) Counter(name, description string) Counter { - c, err := b.m.Int64Counter(name, otelmetric.WithDescription(description)) +func (m *otelMeter) Counter(name, description string) Counter { + c, err := m.inner.Int64Counter(name, metric.WithDescription(description)) if err != nil { return noopCounter{} } - return otelCounterBridge{c: c} + return &otelCounter{inner: c} } -func (b otelMeterBridge) Histogram(name, description, unit string) Histogram { - h, err := b.m.Float64Histogram(name, - otelmetric.WithDescription(description), - otelmetric.WithUnit(unit), - ) +func (m *otelMeter) Histogram(name, description, unit string) Histogram { + opts := []metric.Float64HistogramOption{metric.WithDescription(description)} + if unit != "" { + opts = append(opts, metric.WithUnit(unit)) + } + h, err := m.inner.Float64Histogram(name, opts...) if err != nil { return noopHistogram{} } - return otelHistogramBridge{h: h} + return &otelHistogram{inner: h} } -func (b otelMeterBridge) Gauge(name, description string) Gauge { - g, err := b.m.Float64Gauge(name, otelmetric.WithDescription(description)) +func (m *otelMeter) Gauge(name, description string) Gauge { + g, err := m.inner.Float64Gauge(name, metric.WithDescription(description)) if err != nil { return noopGauge{} } - return otelGaugeBridge{g: g} + return &otelGauge{inner: g} } -type otelCounterBridge struct{ c otelmetric.Int64Counter } +type otelCounter struct{ inner metric.Int64Counter } -func (b otelCounterBridge) Add(ctx context.Context, delta int64, attrs ...Attr) { - b.c.Add(ctx, delta, otelmetric.WithAttributes(toKVs(attrs)...)) +func (c *otelCounter) Add(ctx context.Context, delta int64, attrs ...Attr) { + c.inner.Add(ctx, delta, metric.WithAttributes(convertAttrs(attrs)...)) } -type otelHistogramBridge struct{ h otelmetric.Float64Histogram } +type otelHistogram struct{ inner metric.Float64Histogram } -func (b otelHistogramBridge) Record(ctx context.Context, value float64, attrs ...Attr) { - b.h.Record(ctx, value, otelmetric.WithAttributes(toKVs(attrs)...)) +func (h *otelHistogram) Record(ctx context.Context, value float64, attrs ...Attr) { + h.inner.Record(ctx, value, metric.WithAttributes(convertAttrs(attrs)...)) } -type otelGaugeBridge struct{ g otelmetric.Float64Gauge } +type otelGauge struct{ inner metric.Float64Gauge } -func (b otelGaugeBridge) Set(ctx context.Context, value float64, attrs ...Attr) { - b.g.Record(ctx, value, otelmetric.WithAttributes(toKVs(attrs)...)) +func (g *otelGauge) Set(ctx context.Context, value float64, attrs ...Attr) { + g.inner.Record(ctx, value, metric.WithAttributes(convertAttrs(attrs)...)) } -// ── Attribute helpers ─────────────────────────────────────────────────────── - -func toKV(a Attr) attribute.KeyValue { - switch v := a.Value.(type) { - case string: - return attribute.String(a.Key, v) - case int64: - return attribute.Int64(a.Key, v) - case int: - return attribute.Int(a.Key, v) - case float64: - return attribute.Float64(a.Key, v) - case bool: - return attribute.Bool(a.Key, v) - default: - return attribute.String(a.Key, fmt.Sprintf("%v", v)) +// convertAttrs maps telemetry.Attr values to attribute.KeyValue. Unknown +// types are rendered via fmt.Sprint so callers never panic on exotic values. +// The uint64 / uint32 / uint cases matter: sequence numbers and ID fields +// in OwnCord are unsigned, and routing them through the default fmt.Sprint +// path would encode them as string attributes and break metric aggregation. +func convertAttrs(in []Attr) []attribute.KeyValue { + if len(in) == 0 { + return nil } -} - -func toKVs(attrs []Attr) []attribute.KeyValue { - kvs := make([]attribute.KeyValue, 0, len(attrs)) - for _, a := range attrs { - kvs = append(kvs, toKV(a)) + out := make([]attribute.KeyValue, 0, len(in)) + for _, a := range in { + switch v := a.Value.(type) { + case string: + out = append(out, attribute.String(a.Key, v)) + case int: + out = append(out, attribute.Int(a.Key, v)) + case int32: + out = append(out, attribute.Int64(a.Key, int64(v))) + case int64: + out = append(out, attribute.Int64(a.Key, v)) + case uint: + out = append(out, attribute.Int64(a.Key, int64(v))) + case uint32: + out = append(out, attribute.Int64(a.Key, int64(v))) + case uint64: + // Most uint64 values in OwnCord (ids, seqs) fit comfortably + // within int64 range. A wrapped negative would corrupt + // metric aggregation, so we fall back to a string for the + // pathological case rather than silently misreporting. + if v <= math.MaxInt64 { + out = append(out, attribute.Int64(a.Key, int64(v))) + } else { + out = append(out, attribute.String(a.Key, fmt.Sprint(v))) + } + case float32: + out = append(out, attribute.Float64(a.Key, float64(v))) + case float64: + out = append(out, attribute.Float64(a.Key, v)) + case bool: + out = append(out, attribute.Bool(a.Key, v)) + default: + out = append(out, attribute.String(a.Key, fmt.Sprint(v))) + } } - return kvs + return out } diff --git a/Server/telemetry/telemetry_otel_test.go b/Server/telemetry/telemetry_otel_test.go new file mode 100644 index 00000000..c91c2375 --- /dev/null +++ b/Server/telemetry/telemetry_otel_test.go @@ -0,0 +1,214 @@ +//go:build otel + +// Phase B Step 8 — tests for the real OpenTelemetry provider. Only compiled +// under `-tags otel`, alongside telemetry_otel.go. These tests exercise the +// behavioural contract the no-op build cannot: that Init actually wires a +// Prometheus exporter, that spans created via the adapter reach the SDK, and +// that Shutdown flushes cleanly. +package telemetry + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/owncord/server/config" +) + +func TestOtelInitPrometheusExporter(t *testing.T) { + resetGlobalForTest(t) + shutdown, err := Init(context.Background(), config.TelemetryConfig{ + Enabled: true, + Exporter: "prometheus", + ServiceName: "owncord-test", + }) + if err != nil { + t.Fatalf("Init: %v", err) + } + t.Cleanup(func() { _ = shutdown(context.Background()) }) + + p := Global() + if p == nil { + t.Fatal("Global is nil after Init") + } + handler := p.PrometheusHandler() + if handler == nil { + t.Fatal("PrometheusHandler is nil after prometheus Init") + } + + // Record one metric and scrape the exporter; the metric must appear in + // the /metrics response body. + ctx := context.Background() + meter := p.Meter("telemetry_test") + counter := meter.Counter("otel_test_counter_total", "test counter") + counter.Add(ctx, 3, String("fixture", "init")) + + rec := httptest.NewRecorder() + handler.ServeHTTP(rec, httptest.NewRequest("GET", "/metrics", nil)) + body := rec.Body.String() + if rec.Code != http.StatusOK { + t.Fatalf("prometheus handler status: got %d, body=%s", rec.Code, body) + } + if !strings.Contains(body, "otel_test_counter_total") { + t.Fatalf("expected otel_test_counter_total in exporter body:\n%s", body) + } +} + +func TestOtelInitNoneReturnsNoopProvider(t *testing.T) { + resetGlobalForTest(t) + shutdown, err := Init(context.Background(), config.TelemetryConfig{ + Enabled: true, + Exporter: "none", + }) + if err != nil { + t.Fatalf("Init: %v", err) + } + t.Cleanup(func() { _ = shutdown(context.Background()) }) + if _, ok := Global().(noopProvider); !ok { + t.Fatalf("expected noopProvider, got %T", Global()) + } +} + +func TestOtelInitDisabledReturnsNoopProvider(t *testing.T) { + resetGlobalForTest(t) + shutdown, err := Init(context.Background(), config.TelemetryConfig{Enabled: false}) + if err != nil { + t.Fatalf("Init: %v", err) + } + t.Cleanup(func() { _ = shutdown(context.Background()) }) + if _, ok := Global().(noopProvider); !ok { + t.Fatalf("expected noopProvider, got %T", Global()) + } +} + +func TestOtelTracerRecordsSpan(t *testing.T) { + resetGlobalForTest(t) + shutdown, err := Init(context.Background(), config.TelemetryConfig{ + Enabled: true, + Exporter: "prometheus", + ServiceName: "owncord-test", + }) + if err != nil { + t.Fatalf("Init: %v", err) + } + t.Cleanup(func() { _ = shutdown(context.Background()) }) + + tracer := Global().Tracer("telemetry_test") + ctx, span := tracer.Start(context.Background(), "unit_test", + String("attr_string", "v"), + Int64("attr_int", 42), + Float64("attr_float", 1.5), + ) + span.SetAttributes(String("late", "ok")) + span.RecordError(errors.New("boom")) + span.End() + _ = ctx +} + +func TestOtelHistogramRecordsSeconds(t *testing.T) { + resetGlobalForTest(t) + shutdown, err := Init(context.Background(), config.TelemetryConfig{ + Enabled: true, + Exporter: "prometheus", + ServiceName: "owncord-test", + }) + if err != nil { + t.Fatalf("Init: %v", err) + } + t.Cleanup(func() { _ = shutdown(context.Background()) }) + + h := Global().Meter("telemetry_test").Histogram("otel_test_latency_seconds", "test", "s") + TimeSince(context.Background(), h, time.Now().Add(-250*time.Millisecond)) +} + +func TestOtelShutdownIdempotent(t *testing.T) { + resetGlobalForTest(t) + shutdown, err := Init(context.Background(), config.TelemetryConfig{ + Enabled: true, + Exporter: "prometheus", + ServiceName: "owncord-test", + }) + if err != nil { + t.Fatalf("Init: %v", err) + } + if err := shutdown(context.Background()); err != nil { + t.Fatalf("first shutdown: %v", err) + } + // Second call must not panic or error. + if err := shutdown(context.Background()); err != nil { + t.Fatalf("second shutdown: %v", err) + } +} + +// resetGlobalForTest restores a fresh no-op provider between tests that each +// call Init. We cannot cleanly tear down and re-register the otel globals in +// all cases, so tests that run back-to-back must start from the no-op state. +func resetGlobalForTest(t *testing.T) { + t.Helper() + SetGlobal(noopProvider{}) + resetAppMetricsForInit() +} + +func TestOtelConvertAttrsUint64OverflowFallsBackToString(t *testing.T) { + const tooBig = uint64(1<<63) + 7 // > math.MaxInt64 + attrs := convertAttrs([]Attr{{Key: "huge", Value: tooBig}}) + if len(attrs) != 1 { + t.Fatalf("expected 1 attr, got %d", len(attrs)) + } + if got := attrs[0].Value.Type().String(); got != "STRING" { + t.Fatalf("overflowing uint64 should become STRING attr to avoid wrap, got %s", got) + } +} + +func TestOtelConvertAttrsHandlesUnsignedInts(t *testing.T) { + attrs := convertAttrs([]Attr{ + {Key: "u64", Value: uint64(1 << 40)}, + {Key: "u32", Value: uint32(7)}, + {Key: "u", Value: uint(42)}, + {Key: "i", Value: int(-1)}, + {Key: "i64", Value: int64(-1 << 40)}, + {Key: "f32", Value: float32(2.5)}, + {Key: "b", Value: true}, + {Key: "s", Value: "hello"}, + }) + if len(attrs) != 8 { + t.Fatalf("expected 8 attrs, got %d", len(attrs)) + } + // uint64 must become an Int64 attribute, not a string. + if got := attrs[0].Value.Type().String(); got != "INT64" { + t.Fatalf("u64 attr type = %s, want INT64", got) + } + if got := attrs[0].Value.AsInt64(); got != int64(1<<40) { + t.Fatalf("u64 attr value = %d, want %d", got, int64(1<<40)) + } +} + +func TestOtelAppMetricsRebindsAfterInit(t *testing.T) { + resetGlobalForTest(t) + // Build once against the noop global; verify the bundle is the no-op + // counter type, then swap providers and verify NewAppMetrics returns a + // re-bound bundle. + before := NewAppMetrics() + if _, ok := before.WSMessagesTotal.(noopCounter); !ok { + t.Fatalf("expected noopCounter before Init, got %T", before.WSMessagesTotal) + } + + shutdown, err := Init(context.Background(), config.TelemetryConfig{ + Enabled: true, + Exporter: "prometheus", + ServiceName: "owncord-test", + }) + if err != nil { + t.Fatalf("Init: %v", err) + } + t.Cleanup(func() { _ = shutdown(context.Background()) }) + + after := NewAppMetrics() + if _, ok := after.WSMessagesTotal.(noopCounter); ok { + t.Fatalf("WSMessagesTotal is still a noopCounter after OTel Init — AppMetrics cache was not reset") + } +}