From ac9211e87b88e2ba1f6a4313e76fdf91e9381503 Mon Sep 17 00:00:00 2001 From: robertjamesprior <83608739+robertjamesprior@users.noreply.github.com> Date: Wed, 14 Oct 2026 16:55:25 +0000 Subject: [PATCH] Publish a live view failure event when the capture pipeline cannot start --- server/internal/plugins/telemetry/manager.go | 24 ++++++- .../plugins/telemetry/manager_test.go | 60 +++++++++++++++++ server/internal/plugins/telemetry/plugin.go | 2 + server/internal/plugins/telemetry/types.go | 67 +++++++++++++++++++ server/internal/websocket/handler/handler.go | 8 +++ 5 files changed, 160 insertions(+), 1 deletion(-) diff --git a/server/internal/plugins/telemetry/manager.go b/server/internal/plugins/telemetry/manager.go index 9fa0448f9..31e107792 100644 --- a/server/internal/plugins/telemetry/manager.go +++ b/server/internal/plugins/telemetry/manager.go @@ -102,6 +102,22 @@ func (m *Manager) handleDisconnect(id string) { }) } +// reportLiveViewFailure publishes a capture pipeline that could not start. The +// code is the one the dashboard classifies on, so the failure is attributable to +// an org and a session without reading the VM's log. +func (m *Manager) reportLiveViewFailure(sessionID, code, message string) { + m.enqueue(eventPayload{ + Type: eventLiveViewFailed, + Category: categoryConnection, + SourceEvent: "neko.live_view.pipeline_failed", + Data: map[string]any{ + "session_id": sessionID, + "code": code, + "message": message, + }, + }) +} + func (m *Manager) enqueue(ev eventPayload) { select { case m.eventsCh <- ev: @@ -134,9 +150,14 @@ func (m *Manager) worker() { } func (m *Manager) publish(ev eventPayload) { + category := ev.Category + if category == "" { + category = "system" + } + body := publishBody{ Type: ev.Type, - Category: "system", + Category: category, Source: publishSource{ Kind: "local_process", Event: ev.SourceEvent, @@ -169,6 +190,7 @@ func (m *Manager) publish(ev eventPayload) { type eventPayload struct { Type string + Category string SourceEvent string Data map[string]any } diff --git a/server/internal/plugins/telemetry/manager_test.go b/server/internal/plugins/telemetry/manager_test.go index 5460d1134..5bfceb525 100644 --- a/server/internal/plugins/telemetry/manager_test.go +++ b/server/internal/plugins/telemetry/manager_test.go @@ -2,6 +2,7 @@ package telemetry import ( "encoding/json" + "errors" "io" "net/http" "net/http/httptest" @@ -165,3 +166,62 @@ func connect(t *testing.T, sm types.SessionManager, id string) (types.Session, t s.ConnectWebSocketPeer(p) return s, p } + +func TestManager_ReportsLiveViewFailure(t *testing.T) { + srv, captured := newCapturingServer(t) + defer srv.Close() + + sm := newSessionManager(t) + m := NewManager(sm, &Config{Enabled: true, Endpoint: srv.URL}) + require.NoError(t, m.Start()) + defer func() { require.NoError(t, m.Shutdown()) }() + + m.reportLiveViewFailure("session-1", codeGPUEncoderUnavailable, `(pipeline error) no element "nvh264enc"`) + + require.Eventually(t, func() bool { + return len(captured.snapshot()) >= 1 + }, 2*time.Second, 10*time.Millisecond, "expected a failure event") + + ev := captured.snapshot()[0] + require.Equal(t, "live_view_failed", ev.Type) + require.Equal(t, "connection", ev.Category) + require.Equal(t, "local_process", ev.Source.Kind) + require.Equal(t, "neko.live_view.pipeline_failed", ev.Source.Event) + require.Equal(t, "session-1", ev.Data["session_id"]) + require.Equal(t, codeGPUEncoderUnavailable, ev.Data["code"]) + require.NotEmpty(t, ev.Data["message"]) +} + +func TestReportLiveViewFailure_WithoutReporter(t *testing.T) { + setReporter(nil) + + require.NotPanics(t, func() { + ReportLiveViewFailure("session-1", errors.New(`(pipeline error) no element "nvh264enc"`)) + }) + require.NotPanics(t, func() { + ReportLiveViewFailure("", errors.New("pipeline failed")) + }) + require.NotPanics(t, func() { + ReportLiveViewFailure("session-1", nil) + }) +} + +func TestClassifyPipelineError(t *testing.T) { + cases := []struct { + name string + err error + want string + }{ + {"missing encoder element", errors.New(`(pipeline error) no element "nvh264enc"`), codeGPUEncoderUnavailable}, + {"cuda context", errors.New("failed to create cuda context, ret: 0x2"), codeGPUEncoderUnavailable}, + {"plugin load", errors.New("element plugin couldn't be loaded"), codeGPUEncoderUnavailable}, + {"other pipeline failure", errors.New("webrtc stream not found"), codePipelineFailed}, + {"no error", nil, codePipelineFailed}, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + require.Equal(t, c.want, classifyPipelineError(c.err)) + }) + } +} diff --git a/server/internal/plugins/telemetry/plugin.go b/server/internal/plugins/telemetry/plugin.go index aa6d5e87d..97197cfe9 100644 --- a/server/internal/plugins/telemetry/plugin.go +++ b/server/internal/plugins/telemetry/plugin.go @@ -25,9 +25,11 @@ func (p *Plugin) Config() types.PluginConfig { func (p *Plugin) Start(m types.PluginManagers) error { p.manager = NewManager(m.SessionManager, p.config) + setReporter(p.manager.reportLiveViewFailure) return p.manager.Start() } func (p *Plugin) Shutdown() error { + setReporter(nil) return p.manager.Shutdown() } diff --git a/server/internal/plugins/telemetry/types.go b/server/internal/plugins/telemetry/types.go index 72082f0e2..7aef7136c 100644 --- a/server/internal/plugins/telemetry/types.go +++ b/server/internal/plugins/telemetry/types.go @@ -1,3 +1,70 @@ package telemetry +import ( + "regexp" + "sync" +) + const PluginName = "telemetry" + +const ( + // eventLiveViewFailed is published when a live-view capture pipeline cannot + // start, so the viewer connects and never receives a frame. + eventLiveViewFailed = "live_view_failed" + + categoryConnection = "connection" + + // Codes match the vocabulary the dashboard already classifies on, so the + // control plane does not need a second taxonomy. + codeGPUEncoderUnavailable = "gpu_encoder_unavailable" + codePipelineFailed = "live_view_pipeline_failed" +) + +// Matches the strings GStreamer and the NVIDIA stack produce when the encoder +// cannot be created, which is what a live view hits when the GPU is already +// holding an encoder session for the replay recorder. +var gpuEncoderFailurePattern = regexp.MustCompile( + `(?i)cuctxcreate failed|failed to create cuda context|cuda_error_out_of_memory|insufficient memory on nvh264enc|nvh264enc|nvenc|element plugin couldn't be loaded`, +) + +func classifyPipelineError(err error) string { + if err == nil { + return codePipelineFailed + } + if gpuEncoderFailurePattern.MatchString(err.Error()) { + return codeGPUEncoderUnavailable + } + return codePipelineFailed +} + +var ( + reporterMu sync.RWMutex + reporter func(sessionID, code, message string) +) + +func setReporter(fn func(sessionID, code, message string)) { + reporterMu.Lock() + defer reporterMu.Unlock() + reporter = fn +} + +func currentReporter() func(sessionID, code, message string) { + reporterMu.RLock() + defer reporterMu.RUnlock() + return reporter +} + +// ReportLiveViewFailure records a live-view pipeline failure for the session that +// asked for the stream. The websocket handler owns both the session and the error, +// so it reports through this hook instead of holding a plugin reference. No-op +// when the plugin is disabled or has not started. +func ReportLiveViewFailure(sessionID string, err error) { + if sessionID == "" || err == nil { + return + } + fn := currentReporter() + if fn == nil { + return + } + fn(sessionID, classifyPipelineError(err), err.Error()) +} diff --git a/server/internal/websocket/handler/handler.go b/server/internal/websocket/handler/handler.go index 4bf9830cc..47b9fadaf 100644 --- a/server/internal/websocket/handler/handler.go +++ b/server/internal/websocket/handler/handler.go @@ -4,6 +4,7 @@ import ( "github.com/rs/zerolog" "github.com/rs/zerolog/log" + "github.com/m1k1o/neko/server/internal/plugins/telemetry" "github.com/m1k1o/neko/server/pkg/types" "github.com/m1k1o/neko/server/pkg/types/event" "github.com/m1k1o/neko/server/pkg/types/message" @@ -202,6 +203,13 @@ func (h *MessageHandlerCtx) Message(session types.Session, data types.WebSocketM Str("event", data.Event). Str("session_id", session.ID()). Msg("message handler has failed") + + // A failed signal request is how a capture pipeline that cannot start + // reaches the viewer: the encoder is unavailable, so no frames render and + // the client has nothing to report. + if data.Event == event.SIGNAL_REQUEST { + telemetry.ReportLiveViewFailure(session.ID(), err) + } } return true