Merge pull request #12 from kerberos-io/feature/enhance-ptz-failure-logging

feature/enhance-ptz-failure-logging

event/stream/doc.go

@@ -18,7 +18,8 @@
 //	}
 //
 // NewStream performs network I/O so auth and reachability failures
-// surface synchronously. Events and Errors close when the Stream stops;
+// surface synchronously, and rejects a client timeout that cannot
+// outlast PullTimeout with ErrInvalidOptions. Events and Errors close when the Stream stops;
 // Errors sends are non-blocking so a stalled consumer drops older
 // errors rather than blocking the pull loop. After a silent reconnect,
 // the next batch's events carry Event.AfterReconnect=true.

event/stream/pulltimeout.go

@@ -0,0 +1,51 @@
+package stream
+
+import (
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/kerberos-io/onvif"
+)
+
+// ErrInvalidOptions marks a configuration that cannot succeed. Callers
+// retry the pull/renew/recreate errors; retrying this one never helps,
+// so it is a distinct sentinel they can short-circuit on.
+var ErrInvalidOptions = errors.New("stream: invalid options")
+
+// minClientHeadroom is how far http.Client.Timeout must exceed
+// PullTimeout. The client ceiling covers dial, TLS and the response
+// transfer on top of the poll it has to outlast, and starts before the
+// camera has parsed the request; on a cellular bearer that overhead
+// runs to hundreds of milliseconds.
+const minClientHeadroom = 5 * time.Second
+
+// validateClientTimeout rejects a client ceiling that cannot outlast
+// the PullMessages long-poll plus minClientHeadroom.
+//
+// Zero means unbounded and is accepted: it is the SDK's default when a
+// caller passes no client, so rejecting it would break every default
+// consumer. Note it is not risk-free — the caller interface documents
+// that ctx cannot interrupt an in-flight SOAP call, so only the client
+// timeout can unwedge a stalled camera.
+func validateClientTimeout(clientTimeout, pullTimeout time.Duration) error {
+	if clientTimeout == 0 || clientTimeout >= pullTimeout+minClientHeadroom {
+		return nil
+	}
+	return fmt.Errorf(
+		"%w: http.Client.Timeout (%s) must exceed PullTimeout (%s) by at least %s; PullMessages is a long-poll and the client would abort every quiet pull",
+		ErrInvalidOptions, clientTimeout, pullTimeout, minClientHeadroom)
+}
+
+// clientTimeoutOf reports the device's HTTP client ceiling, or 0 when
+// the SDK is using its own default (unbounded) client.
+func clientTimeoutOf(dev *onvif.Device) time.Duration {
+	if dev == nil {
+		return 0
+	}
+	c := dev.GetDeviceParams().HttpClient
+	if c == nil {
+		return 0
+	}
+	return c.Timeout
+}

event/stream/pulltimeout_test.go

@@ -0,0 +1,68 @@
+package stream
+
+import (
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestValidateClientTimeout — PullMessages is a long-poll: the camera
+// holds the connection open for PullTimeout waiting for an event. An
+// http.Client.Timeout covers the whole exchange and starts before the
+// camera has parsed the request, so a client ceiling at or below
+// PullTimeout loses the race on every quiet interval and the pull can
+// only ever fail. This shipped once (both were 5s) and presented as a
+// slow camera rather than a misconfiguration.
+//
+// Strict inequality is not enough: the client also has to cover dial,
+// TLS and the response transfer, which on a cellular bearer runs to
+// hundreds of milliseconds. Hence a real headroom floor.
+func TestValidateClientTimeout(t *testing.T) {
+	tests := []struct {
+		name    string
+		client  time.Duration
+		pull    time.Duration
+		wantErr bool
+	}{
+		{"unbounded client is the caller's risk, not an error", 0, 30 * time.Second, false},
+		{"comfortable headroom", 40 * time.Second, 30 * time.Second, false},
+		{"exactly the minimum headroom", 30*time.Second + minClientHeadroom, 30 * time.Second, false},
+		{"a hair under the minimum headroom", 30*time.Second + minClientHeadroom - time.Millisecond, 30 * time.Second, true},
+		{"strictly greater but no headroom", 30*time.Second + time.Millisecond, 30 * time.Second, true},
+		{"equal timeouts always lose", 5 * time.Second, 5 * time.Second, true},
+		{"client below pull", 4 * time.Second, 30 * time.Second, true},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := validateClientTimeout(tt.client, tt.pull)
+			if tt.wantErr {
+				require.Error(t, err, "client=%s pull=%s must be rejected", tt.client, tt.pull)
+				assert.ErrorIs(t, err, ErrInvalidOptions,
+					"callers need a sentinel to tell a permanent misconfiguration from a transient failure")
+				assert.Contains(t, err.Error(), "PullTimeout",
+					"the error must name the option the caller has to change")
+				return
+			}
+			assert.NoError(t, err, "client=%s pull=%s must be accepted", tt.client, tt.pull)
+		})
+	}
+}
+
+// TestErrInvalidOptions_IsDistinctFromStreamErrors — the pull/renew/
+// recreate errors are transient and callers retry them. A bad Options
+// never becomes valid by retrying, so it must not be mistaken for one.
+func TestErrInvalidOptions_IsDistinctFromStreamErrors(t *testing.T) {
+	err := validateClientTimeout(5*time.Second, 5*time.Second)
+	require.Error(t, err)
+
+	var pull ErrPullFailed
+	var renew ErrRenewFailed
+	var recreate ErrRecreateFailed
+	assert.False(t, errors.As(err, &pull))
+	assert.False(t, errors.As(err, &renew))
+	assert.False(t, errors.As(err, &recreate))
+}

event/stream/stream.go

@@ -39,7 +39,9 @@ type Options struct {
 	// server-side filtering is fragile across vendors and empty is
 	// required for AXIS.
 	RawTopicFilter string
-	// PullTimeout — zero means default (5s).
+	// PullTimeout — zero means default (5s). The device's
+	// http.Client.Timeout must exceed this by minClientHeadroom or
+	// NewStream returns ErrInvalidOptions.
 	PullTimeout time.Duration
 	// MessageLimit — zero means default (32). Busy AXIS cameras with
 	// many configured rules can burst beyond 10 per pull.
@@ -234,6 +236,12 @@ func (s *Stream) updateGrantedTerminationIfGen(gen uint64, t time.Time) {
 //
 // The returned Stream stops when ctx is cancelled or Close is called.
 func NewStream(ctx context.Context, dev *onvif.Device, opts Options) (*Stream, error) {
+	// Checked before the subscription call: this config can only fail,
+	// so surfacing it here beats a stream that appears to work and
+	// silently survives on reconnects alone.
+	if err := validateClientTimeout(clientTimeoutOf(dev), opts.withDefaults().PullTimeout); err != nil {
+		return nil, err
+	}
 	return newStream(ctx, deviceCaller{dev: dev}, opts)
 }
 

event/stream/topics.go

@@ -79,10 +79,19 @@ var topicRules = []struct {
 	// https://developer.axis.com/vapix/network-video/event-and-action-services/
 	{"MotionRegionDetector/Motion", KindMotion},
 
-	// AXIS Guard suite — vendor analytics apps with Camera<N>Profile<ID>
-	// suffixes. Treated as motion so they can drive motion-triggered
-	// recording on cameras using these apps instead of basic VMD.
+	// tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_<N> — AXIS VMD 3, the
+	// firmware-builtin predecessor of the VMD 4 ACAP. It lives under
+	// RuleEngine, not CameraApplicationPlatform, so the VMD rule below
+	// does not cover it. Still shipping on deployed cameras.
+	// https://developer.axis.com/vapix/network-video/event-and-action-services/
+	{"RuleEngine/VMD3/", KindMotion},
+
+	// AXIS ACAP motion apps with Camera<N>Profile<ID> suffixes. VMD 4 is
+	// the stock app shipped on the camera; the Guard suite are the
+	// paid analytics products. Prefix-match because of the suffix.
+	// https://developer.axis.com/vapix/applications/vmd4
 	// https://developer.axis.com/vapix/applications/motion-guard
+	{"CameraApplicationPlatform/VMD/", KindMotion},
 	{"CameraApplicationPlatform/MotionGuard/", KindMotion},
 	{"CameraApplicationPlatform/FenceGuard/", KindMotion},
 	{"CameraApplicationPlatform/LoiteringGuard/", KindMotion},

event/stream/topics_test.go

@@ -25,6 +25,18 @@ func TestClassifyTopic(t *testing.T) {
 		{"axis_fence_guard", "tnsaxis:CameraApplicationPlatform/FenceGuard/Camera1ProfileANY", KindMotion},
 		{"axis_loitering_guard", "tnsaxis:CameraApplicationPlatform/LoiteringGuard/Camera1ProfileANY", KindMotion},
 
+		// AXIS VMD 4 — the stock motion app, and the one an installer
+		// reaches for before any Guard product. The profile suffix
+		// varies with the configured VMD profile.
+		{"axis_vmd4_profile_any", "tnsaxis:CameraApplicationPlatform/VMD/Camera1ProfileANY", KindMotion},
+		{"axis_vmd4_profile_numbered", "tnsaxis:CameraApplicationPlatform/VMD/Camera1Profile1", KindMotion},
+
+		// AXIS VMD 3 — the firmware-builtin predecessor, published under
+		// RuleEngine rather than CameraApplicationPlatform. Still
+		// shipping on deployed cameras.
+		{"axis_vmd3_video_1", "tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_1", KindMotion},
+		{"axis_vmd3_video_2", "tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_2", KindMotion},
+
 		// --- Tampering --------------------------------------------------
 
 		{"tamper_detector", "tns1:RuleEngine/TamperDetector/Tamper", KindTampering},
@@ -88,6 +100,16 @@ func TestClassifyTopic(t *testing.T) {
 		{"relay_failure_not_digital_output", "tns1:Device/HardwareFailure/RelayFailure", KindUnknown},
 		{"digital_input_config_not_digital_input", "tns1:Device/IO/DigitalInputConfiguration", KindUnknown},
 		{"tamper_detector_log_not_tampering", "tns1:Device/Diagnostics/TamperDetectorLog", KindUnknown},
+
+		// The two AXIS VMD needles carry a trailing slash so they match a
+		// whole path segment. Without it, any sibling app or rule whose
+		// name merely starts with VMD / VMD3 would classify as motion and
+		// drive recording.
+		{"vmd_statistics_app_not_motion", "tnsaxis:CameraApplicationPlatform/VMDStatistics/Camera1", KindUnknown},
+		{"vmd3_config_rule_not_motion", "tns1:RuleEngine/tnsaxis:VMD3Config/Changed", KindUnknown},
+		// VMD3 is scoped to RuleEngine; the same name under another
+		// container is a different thing.
+		{"vmd3_outside_rule_engine_not_motion", "tnsaxis:Storage/VMD3/Status", KindUnknown},
 	}
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {

networking/digest_test.go

@@ -1,10 +1,57 @@
 package networking
 
 import (
+	"net/http"
+	"net/http/httptest"
 	"strings"
 	"testing"
 )
 
+func TestSendSoapWithDigestReturnsServerError(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
+		writer.WriteHeader(http.StatusInternalServerError)
+	}))
+	defer server.Close()
+
+	response, err := SendSoapWithDigest(server.Client(), server.URL, "<Envelope/>", "user", "password")
+	if response == nil {
+		t.Fatal("SendSoapWithDigest returned a nil response")
+	}
+	defer response.Body.Close()
+	if err == nil {
+		t.Fatal("SendSoapWithDigest returned nil error for HTTP 500")
+	}
+}
+
+func TestSendSoapWithDigestReturnsServerErrorAfterAuthentication(t *testing.T) {
+	requestCount := 0
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		requestCount++
+		if requestCount == 1 {
+			writer.Header().Set("WWW-Authenticate", `Digest realm="AXIS", nonce="nonce", qop="auth", algorithm=MD5`)
+			writer.WriteHeader(http.StatusUnauthorized)
+			return
+		}
+		if !strings.HasPrefix(request.Header.Get("Authorization"), "Digest ") {
+			t.Error("authenticated retry is missing Digest Authorization header")
+		}
+		writer.WriteHeader(http.StatusInternalServerError)
+	}))
+	defer server.Close()
+
+	response, err := SendSoapWithDigest(server.Client(), server.URL, "<Envelope/>", "user", "password")
+	if response == nil {
+		t.Fatal("SendSoapWithDigest returned a nil response")
+	}
+	defer response.Body.Close()
+	if err == nil {
+		t.Fatal("SendSoapWithDigest returned nil error for authenticated HTTP 500")
+	}
+	if requestCount != 2 {
+		t.Fatalf("request count = %d, want 2", requestCount)
+	}
+}
+
 func TestParseDigestChallenge(t *testing.T) {
 	challenge := `Digest realm="testrealm@host.com", qop="auth,auth-int", nonce="dcd98b7102dd2f0e8b11d0f600bfb0c093", opaque="5ccc069c403ebaf9f0171e9517f40e41", algorithm=MD5`
 	parts := parseDigestChallenge(challenge)

networking/networking.go

@@ -24,12 +24,14 @@ func SendSoap(httpClient *http.Client, endpoint, message string) (*http.Response
 		return resp, errors.Annotate(err, "Post")
 	}
 
-	// if resp.StatusCode is 4xx,5xx, return error
-	if resp.StatusCode >= 400 && resp.StatusCode < 600 {
-		return resp, errors.Errorf("Server error: %d: %s", resp.StatusCode, resp.Status)
-	}
+	return resp, responseError(resp)
+}
 
-	return resp, nil
+func responseError(resp *http.Response) error {
+	if resp.StatusCode >= 400 && resp.StatusCode < 600 {
+		return errors.Errorf("Server error: %d: %s", resp.StatusCode, resp.Status)
+	}
+	return nil
 }
 
 // SendSoapWithDigest sends a soap message and, when the device answers with an
@@ -57,13 +59,13 @@ func SendSoapWithDigest(httpClient *http.Client, endpoint, message, username, pa
 
 	// Only escalate to HTTP digest when the device explicitly asks for it.
 	if resp.StatusCode != http.StatusUnauthorized {
-		return resp, nil
+		return resp, responseError(resp)
 	}
 
 	challenge := resp.Header.Get("WWW-Authenticate")
 	if !strings.HasPrefix(strings.ToLower(strings.TrimSpace(challenge)), "digest") {
 		// Not a digest challenge (e.g. Basic) - nothing more we can do here.
-		return resp, nil
+		return resp, responseError(resp)
 	}
 
 	authorization := newDigestAuthorization(challenge, http.MethodPost, endpoint, username, password)
@@ -86,7 +88,7 @@ func SendSoapWithDigest(httpClient *http.Client, endpoint, message, username, pa
 		return resp, errors.Annotate(err, "Post with digest")
 	}
 
-	return resp, nil
+	return resp, responseError(resp)
 }
 
 // stripWSSecurityHeader removes the wsse:Security header block from a SOAP

ptz/types_test.go

@@ -0,0 +1,25 @@
+package ptz
+
+import (
+	"encoding/xml"
+	"strings"
+	"testing"
+
+	"github.com/kerberos-io/onvif/xsd/onvif"
+)
+
+func TestContinuousMoveIncludesZeroPanTiltCoordinates(t *testing.T) {
+	request := ContinuousMove{
+		Velocity: onvif.PTZSpeedPanTilt{
+			PanTilt: onvif.Vector2D{X: 0.5, Y: 0},
+		},
+	}
+
+	encoded, err := xml.Marshal(request)
+	if err != nil {
+		t.Fatalf("xml.Marshal() error = %v", err)
+	}
+	if !strings.Contains(string(encoded), `x="0.5" y="0"`) {
+		t.Fatalf("ContinuousMove PanTilt = %s, want explicit x and y attributes", encoded)
+	}
+}

xsd/onvif/onvif.go

@@ -666,13 +666,13 @@ type PTZSpeedPanTilt struct {
 }
 
 type Vector2D struct {
-	X     float64     `xml:"x,attr,omitempty"`
-	Y     float64     `xml:"y,attr,omitempty"`
+	X     float64     `xml:"x,attr"`
+	Y     float64     `xml:"y,attr"`
 	Space *xsd.AnyURI `xml:"space,attr,omitempty"`
 }
 
 type Vector1D struct {
-	X     float64     `xml:"x,attr,omitempty"`
+	X     float64     `xml:"x,attr"`
 	Space *xsd.AnyURI `xml:"space,attr,omitempty"`
 }
 
@@ -1176,7 +1176,7 @@ type PresetTour struct {
 	Status            PTZPresetTourStatus            `xml:"Status"`
 	AutoStart         xsd.Boolean                    `xml:"AutoStart"`
 	StartingCondition PTZPresetTourStartingCondition `xml:"StartingCondition"`
-	TourSpot          []PTZPresetTourSpot           	 `xml:"TourSpot"`
+	TourSpot          []PTZPresetTourSpot                    `xml:"TourSpot"`
 	Extension         PTZPresetTourExtension         `xml:"Extension"`
 }