fix(machinery/onvif): harden event-stream dispatch and add TDD coverage

Addresses the critical and important findings from the second review of the agent integration. TDD followed locally: tests were written first and confirmed RED against the previous implementation before the fix turned them GREEN. Critical fixes -------------- * Shutdown-race panic (concurrency P0): the 3s gap between the agent's ctx cancel and close(HandleMotion) was reachable by a buffered event delivered after cancel, where dispatchEvent's send-with-default select would panic on the closed channel. dispatchEvent now takes ctx, has a pre-check after the kind/state/recording filters, and the send select includes a <-ctx.Done() arm. Pinned by TestDispatchEvent_CtxCancelledAndHandleMotionClosed_DoesNotPanic (asserts NotPanics; current code without the fix panics). * No retry on initial connect (Go P0 + ops P1): previously the goroutine exited permanently if ConnectToOnvifDevice or stream.NewStream failed at agent start — a brief boot-time DNS or network blip silently disabled ONVIF until restart. Construction is now wrapped in a retry loop with exponential backoff (1s -> 5min), matching what cloud.HandleHeartBeat does for its ONVIF connection attempts. The library handles in-stream recovery already; this covers the gap the library cannot see. * Strict 'true' match (Go P0): isONVIFMotionEnabled now normalises case and trims whitespace, so 'True', 'TRUE', ' true ' all enable the feature. Pinned by TestIsONVIFMotionEnabled_CaseAndWhitespace. Important fixes --------------- * Empty DeviceID fallback (Go P1): resolveDeviceID falls back from configuration.Name to camera.ONVIFXAddr to 'unknown' so log lines and metrics always have a useful identifier. Pinned by TestResolveDeviceID_FallbackChain. * Recovery log (ops P1): the run loop tracks a 'recovering' flag set when an ErrPullFailed/ErrRecreateFailed lands on Errors and cleared on the first successful Event. Logs an Info 'event stream recovered' line so on-call operators can see error streaks clear, instead of waking up to ERROR with no closure. * Misconfig log bumped Info -> Warning so the 'ONVIFXAddr is empty' line stands out from the heartbeat noise. Tests ----- events_test.go covers the dispatch contract end-to-end: * Motion+Active -> HandleMotion (happy path). * Motion+Inactive ignored (motion-stop is a documented follow-up). * Non-motion kinds ignored. * Recording='false' gates the send. * Full HandleMotion drops rather than blocks. * Ctx-cancelled + closed HandleMotion does not panic. * isONVIFMotionEnabled handles case and whitespace. * resolveDeviceID fallback chain. go.mod / go.sum: testify moved from indirect to direct dependency. Deferred (out of scope for this commit, tracked as follow-ups): * Heartbeat surface for ONVIF state ('disabled|running|failed') — requires a Cloud.go change beyond this integration's scope. * OTel span/metric for stream lifecycle. * Runtime toggle without restart (config-reload). * Replace-directive layout documentation — separate docs commit.
2026-08-23 15:08:32 +00:00 · 2026-05-21 17:40:24 +02:00
parent ed85261c8e
commit 4f2a96b5e1
3 changed files with 317 additions and 30 deletions
--- a/machinery/go.mod
+++ b/machinery/go.mod
@@ -32,6 +32,7 @@ require (
 	github.com/pion/rtp v1.8.19
 	github.com/pion/webrtc/v4 v4.1.2
 	github.com/sirupsen/logrus v1.9.3
+	github.com/stretchr/testify v1.10.0
 	github.com/swaggo/files v1.0.1
 	github.com/swaggo/gin-swagger v1.6.0
 	github.com/swaggo/swag v1.16.4
@@ -58,6 +59,7 @@ require (
 	github.com/clbanning/mxj v1.8.4 // indirect
 	github.com/clbanning/mxj/v2 v2.7.0 // indirect
 	github.com/cloudwego/base64x v0.1.5 // indirect
+	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/elastic/go-windows v1.0.2 // indirect
 	github.com/elgs/gostrgen v0.0.0-20161222160715-9d61ae07eeae // indirect
 	github.com/erikstmartin/go-testdb v0.0.0-20160219214506-8d10e4a1bae5 // indirect
@@ -108,6 +110,7 @@ require (
 	github.com/pion/stun/v3 v3.0.0 // indirect
 	github.com/pion/transport/v3 v3.0.7 // indirect
 	github.com/pion/turn/v4 v4.0.0 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/procfs v0.15.1 // indirect
 	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
 	github.com/ugorji/go/codec v1.2.12 // indirect
--- a/machinery/src/onvif/events.go
+++ b/machinery/src/onvif/events.go
@@ -3,6 +3,7 @@ package onvif
 import (
 	"context"
 	"errors"
+	"strings"
 	"time"

 	"github.com/kerberos-io/agent/machinery/src/log"
@@ -10,43 +11,78 @@ import (
 	"github.com/kerberos-io/onvif/event/stream"
 )

+// initialBackoff and maxBackoff bound the wait between successive
+// attempts to (re)open the event stream after a transient construction
+// failure (camera reachable but ONVIF not yet ready, brief network blip
+// at agent boot, etc.). The library itself handles in-stream reconnect;
+// these guards cover the initial-connect path the library cannot see.
+const (
+	initialBackoff = time.Second
+	maxBackoff     = 5 * time.Minute
+)
+
 // HandleONVIFEventStream opens an event/stream against the configured
 // ONVIF camera and routes Motion events into communication.HandleMotion
 // so they trigger the existing recording pipeline.
 //
-// The goroutine returns when the stream's context is cancelled (the
-// caller closes ctx when shutting the agent down) or when the camera
-// is not configured for ONVIF.
+// The goroutine retries construction with exponential backoff on
+// transient failure (camera reachable later, credentials reloaded,
+// network restored). It exits cleanly when ctx is cancelled.
 //
-// This is a feature behind Capture.ONVIFMotion. When the flag is empty
-// or "false" the goroutine returns immediately, preserving the
-// pixel-diff motion detector as the only source.
+// This is a feature behind Capture.ONVIFMotion. When the flag is not
+// enabled the goroutine returns immediately, preserving the pixel-diff
+// motion detector as the only source. Toggling the flag at runtime
+// requires an agent restart (Capture.ONVIFMotion is read once at
+// goroutine start).
 func HandleONVIFEventStream(ctx context.Context, configuration *models.Configuration, communication *models.Communication) {
 	log.Log.Debug("onvif.HandleONVIFEventStream(): started")
 	defer log.Log.Debug("onvif.HandleONVIFEventStream(): finished")

-	cfg := configuration.Config.Capture
-	if cfg.ONVIFMotion != "true" {
+	if !isONVIFMotionEnabled(configuration.Config.Capture.ONVIFMotion) {
 		return
 	}
-	camera := cfg.IPCamera
-	if camera.ONVIFXAddr == "" {
-		log.Log.Info("onvif.HandleONVIFEventStream(): ONVIFMotion enabled but ONVIFXAddr is empty; nothing to do")
+	if configuration.Config.Capture.IPCamera.ONVIFXAddr == "" {
+		log.Log.Warning("onvif.HandleONVIFEventStream(): ONVIFMotion enabled but ONVIFXAddr is empty; nothing to do")
 		return
 	}

+	backoff := initialBackoff
+	for {
+		if ctx.Err() != nil {
+			return
+		}
+		recoverable := runStreamOnce(ctx, configuration, communication)
+		if !recoverable {
+			return
+		}
+		if !sleepCtx(ctx, backoff) {
+			return
+		}
+		backoff *= 2
+		if backoff > maxBackoff {
+			backoff = maxBackoff
+		}
+	}
+}
+
+// runStreamOnce opens one event stream and consumes from it until ctx
+// is cancelled or the stream exits. Returns true when the caller
+// should retry construction (transient failure), false on clean
+// ctx-driven shutdown.
+func runStreamOnce(ctx context.Context, configuration *models.Configuration, communication *models.Communication) (retry bool) {
+	camera := configuration.Config.Capture.IPCamera
+
 	device, _, err := ConnectToOnvifDevice(&camera)
 	if err != nil {
 		log.Log.Error("onvif.HandleONVIFEventStream(): connect: " + err.Error())
-		return
+		return true
 	}

-	s, err := stream.NewStream(ctx, device, stream.Options{
-		DeviceID: configuration.Name,
-	})
+	deviceID := resolveDeviceID(configuration.Name, camera.ONVIFXAddr)
+	s, err := stream.NewStream(ctx, device, stream.Options{DeviceID: deviceID})
 	if err != nil {
 		log.Log.Error("onvif.HandleONVIFEventStream(): open stream: " + err.Error())
-		return
+		return true
 	}
 	defer func() {
 		if err := s.Close(); err != nil {
@@ -54,21 +90,33 @@ func HandleONVIFEventStream(ctx context.Context, configuration *models.Configura
 		}
 	}()

-	log.Log.Info("onvif.HandleONVIFEventStream(): consuming events for " + configuration.Name)
+	log.Log.Info("onvif.HandleONVIFEventStream(): consuming events for " + deviceID)

+	// recovering tracks whether we're in a degraded period so the
+	// first successful event after an error streak can log a recovery
+	// line for on-call operators.
+	var recovering bool
 	for {
 		select {
 		case <-ctx.Done():
-			return
+			return false
 		case ev, ok := <-s.Events():
 			if !ok {
-				return
+				// Lib's run goroutine exited — happens only on ctx
+				// cancel today (library handles its own reconnect),
+				// so treat as clean shutdown.
+				return false
 			}
-			dispatchEvent(ev, configuration, communication)
+			if recovering {
+				log.Log.Info("onvif.HandleONVIFEventStream(): event stream recovered for " + deviceID)
+				recovering = false
+			}
+			dispatchEvent(ctx, ev, configuration, communication)
 		case e, ok := <-s.Errors():
 			if !ok {
-				return
+				return false
 			}
+			recovering = true
 			logStreamError(e)
 		}
 	}
@@ -76,17 +124,20 @@ func HandleONVIFEventStream(ctx context.Context, configuration *models.Configura

 // dispatchEvent routes a single decoded ONVIF Event into the agent's
 // existing channels. Motion-active events become MotionDataPartial on
-// HandleMotion, matching what the pixel-diff detector emits. Other
-// event kinds are logged at debug for now; a follow-up will wire
-// DigitalInput/Output into the existing inputOutputDeviceMap.
-func dispatchEvent(ev stream.Event, configuration *models.Configuration, communication *models.Communication) {
+// HandleMotion, matching what the pixel-diff detector emits.
+//
+// The ctx pre-check is the shutdown-race guard: between cancel() and
+// close(communication.HandleMotion) the agent leaves a ~3s window in
+// which a stale event could otherwise attempt to send on a closed
+// channel and panic. If ctx is done we drop the event silently — the
+// recording pipeline is already winding down.
+func dispatchEvent(ctx context.Context, ev stream.Event, configuration *models.Configuration, communication *models.Communication) {
 	if ev.Kind != stream.KindMotion {
 		log.Log.Debug("onvif.dispatchEvent(): non-motion event " + ev.Kind.String() + " topic=" + ev.Topic)
 		return
 	}
-	// We only fire on the leading edge — StateActive. Motion-stop
-	// handling is a follow-up that needs the recorder state machine
-	// to accept an explicit stop signal; today the recorder uses a
+	// Leading-edge only. Motion-stop wiring into the recorder state
+	// machine is tracked as a follow-up; today the recorder uses a
 	// fixed PostRecording timeout.
 	if ev.State != stream.StateActive {
 		return
@@ -94,11 +145,20 @@ func dispatchEvent(ev stream.Event, configuration *models.Configuration, communi
 	if configuration.Config.Capture.Recording == "false" {
 		return
 	}
+	if ctx.Err() != nil {
+		return
+	}
+	// Timestamp in seconds matches what computervision/main.go emits;
+	// downstream consumers (capture/main.go) tolerate either second-
+	// or millisecond-precision.
 	dataToPass := models.MotionDataPartial{
 		Timestamp:       time.Now().Unix(),
-		NumberOfChanges: 0, // ONVIF doesn't quantify motion area.
+		NumberOfChanges: 0, // ONVIF does not quantify motion area.
 	}
 	select {
+	case <-ctx.Done():
+		// Closes the residual race: ctx cancelled between the
+		// pre-check above and reaching this select.
 	case communication.HandleMotion <- dataToPass:
 	default:
 		log.Log.Debug("onvif.dispatchEvent(): HandleMotion full, dropping ONVIF motion event")
@@ -106,7 +166,9 @@ func dispatchEvent(ev stream.Event, configuration *models.Configuration, communi
 }

 // logStreamError logs at a level matching the severity. Recreate
-// failures are louder because they usually mean the camera is offline.
+// failures are loud because they usually mean the camera is offline;
+// pull/renew failures are debug because the library recovers from them
+// automatically.
 func logStreamError(e error) {
 	var recreate stream.ErrRecreateFailed
 	var pull stream.ErrPullFailed
@@ -122,3 +184,39 @@ func logStreamError(e error) {
 		log.Log.Info("onvif.HandleONVIFEventStream(): stream error: " + e.Error())
 	}
 }
+
+// isONVIFMotionEnabled returns true when the Capture.ONVIFMotion flag
+// is set to "true" with case and whitespace tolerance. The rest of the
+// Capture struct uses string flags so we keep the same shape; the
+// difference is that ONVIFMotion defaults to disabled (opt-in), unlike
+// Recording / Motion / Snapshots which default to enabled.
+func isONVIFMotionEnabled(v string) bool {
+	return strings.EqualFold(strings.TrimSpace(v), "true")
+}
+
+// resolveDeviceID returns the most useful identifier for the camera
+// in stream events, logs and metrics. Falls back from configuration
+// name (the operator-supplied label) to the ONVIF endpoint to a
+// constant placeholder so log lines always have something to grep.
+func resolveDeviceID(configName, xaddr string) string {
+	if n := strings.TrimSpace(configName); n != "" {
+		return n
+	}
+	if x := strings.TrimSpace(xaddr); x != "" {
+		return x
+	}
+	return "unknown"
+}
+
+// sleepCtx blocks for d or until ctx is cancelled. Returns false if
+// ctx was cancelled, true if the full duration elapsed.
+func sleepCtx(ctx context.Context, d time.Duration) bool {
+	t := time.NewTimer(d)
+	defer t.Stop()
+	select {
+	case <-ctx.Done():
+		return false
+	case <-t.C:
+		return true
+	}
+}
--- a/machinery/src/onvif/events_test.go
+++ b/machinery/src/onvif/events_test.go
@@ -0,0 +1,186 @@
+package onvif
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/kerberos-io/agent/machinery/src/models"
+	"github.com/kerberos-io/onvif/event/stream"
+	"github.com/stretchr/testify/assert"
+)
+
+func makeConfig(recording, onvifMotion, name string) *models.Configuration {
+	return &models.Configuration{
+		Name: name,
+		Config: models.Config{
+			Capture: models.Capture{
+				Recording:   recording,
+				ONVIFMotion: onvifMotion,
+			},
+		},
+	}
+}
+
+func makeCommunication(buffer int) *models.Communication {
+	return &models.Communication{
+		HandleMotion: make(chan models.MotionDataPartial, buffer),
+	}
+}
+
+// --- dispatchEvent ---------------------------------------------------
+
+func TestDispatchEvent_MotionActive_SendsToHandleMotion(t *testing.T) {
+	cfg := makeConfig("true", "true", "cam-1")
+	comm := makeCommunication(1)
+	ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	dispatchEvent(ctx, ev, cfg, comm)
+
+	select {
+	case m := <-comm.HandleMotion:
+		assert.NotZero(t, m.Timestamp)
+	case <-time.After(time.Second):
+		t.Fatal("expected motion data on HandleMotion")
+	}
+}
+
+func TestDispatchEvent_MotionInactive_DoesNotSend(t *testing.T) {
+	cfg := makeConfig("true", "true", "cam-1")
+	comm := makeCommunication(1)
+	ev := stream.Event{Kind: stream.KindMotion, State: stream.StateInactive}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	dispatchEvent(ctx, ev, cfg, comm)
+
+	select {
+	case <-comm.HandleMotion:
+		t.Fatal("inactive motion must not reach HandleMotion (motion-stop is a follow-up)")
+	case <-time.After(100 * time.Millisecond):
+	}
+}
+
+func TestDispatchEvent_NonMotionKindIgnored(t *testing.T) {
+	cfg := makeConfig("true", "true", "cam-1")
+	comm := makeCommunication(1)
+	ev := stream.Event{Kind: stream.KindDigitalInput, State: stream.StateActive}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	dispatchEvent(ctx, ev, cfg, comm)
+
+	select {
+	case <-comm.HandleMotion:
+		t.Fatal("non-motion kinds must not reach HandleMotion")
+	case <-time.After(100 * time.Millisecond):
+	}
+}
+
+func TestDispatchEvent_RecordingDisabled_DoesNotSend(t *testing.T) {
+	cfg := makeConfig("false", "true", "cam-1")
+	comm := makeCommunication(1)
+	ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	dispatchEvent(ctx, ev, cfg, comm)
+
+	select {
+	case <-comm.HandleMotion:
+		t.Fatal("Recording=false must gate the send (matches computervision behaviour)")
+	case <-time.After(100 * time.Millisecond):
+	}
+}
+
+func TestDispatchEvent_HandleMotionFull_DropsRatherThanBlocks(t *testing.T) {
+	cfg := makeConfig("true", "true", "cam-1")
+	// Pre-fill the buffer so the next send would block.
+	comm := &models.Communication{HandleMotion: make(chan models.MotionDataPartial, 1)}
+	comm.HandleMotion <- models.MotionDataPartial{}
+	ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	done := make(chan struct{})
+	go func() {
+		dispatchEvent(ctx, ev, cfg, comm)
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("dispatchEvent must drop when HandleMotion is full, not block")
+	}
+}
+
+func TestDispatchEvent_CtxCancelledAndHandleMotionClosed_DoesNotPanic(t *testing.T) {
+	// Regression for the shutdown race: between cancel() and
+	// close(HandleMotion) the agent leaves a 3s window. If dispatchEvent
+	// runs in that window AFTER the channel is closed, a non-protected
+	// send would panic. The ctx pre-check must short-circuit before the
+	// send is attempted.
+	cfg := makeConfig("true", "true", "cam-1")
+	comm := &models.Communication{HandleMotion: make(chan models.MotionDataPartial, 1)}
+	close(comm.HandleMotion)
+	ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel() // already cancelled, matching the shutdown sequence
+
+	assert.NotPanics(t, func() {
+		dispatchEvent(ctx, ev, cfg, comm)
+	})
+}
+
+// --- isONVIFMotionEnabled --------------------------------------------
+
+func TestIsONVIFMotionEnabled_CaseAndWhitespace(t *testing.T) {
+	tests := []struct {
+		in   string
+		want bool
+	}{
+		{"true", true},
+		{"True", true},
+		{"TRUE", true},
+		{" true", true},
+		{"true ", true},
+		{"  true  ", true},
+		{"false", false},
+		{"False", false},
+		{"", false},
+		{"yes", false},
+		{"1", false},
+	}
+	for _, tc := range tests {
+		t.Run(tc.in, func(t *testing.T) {
+			assert.Equal(t, tc.want, isONVIFMotionEnabled(tc.in))
+		})
+	}
+}
+
+// --- resolveDeviceID -------------------------------------------------
+
+func TestResolveDeviceID_FallbackChain(t *testing.T) {
+	tests := []struct {
+		name    string
+		cfgName string
+		xaddr   string
+		want    string
+	}{
+		{"name_set", "front-door", "192.168.1.10", "front-door"},
+		{"name_empty_xaddr_set", "", "192.168.1.10", "192.168.1.10"},
+		{"name_whitespace_only_xaddr_set", "  ", "192.168.1.10", "192.168.1.10"},
+		{"both_empty", "", "", "unknown"},
+		{"name_with_trailing_whitespace", "cam-2  ", "192.168.1.10", "cam-2"},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			assert.Equal(t, tc.want, resolveDeviceID(tc.cfgName, tc.xaddr))
+		})
+	}
+}