From 6fc9b23e9fa4b87c110eee3c037590aa31505532 Mon Sep 17 00:00:00 2001 From: Sebastian Norling <1932208+Bazze@users.noreply.github.com> Date: Thu, 21 May 2026 14:55:58 +0200 Subject: [PATCH] feat(event/stream): typed errors and AfterReconnect observability MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces the bare fmt.Errorf wrappers on the Errors channel with three typed errors and adds an Event.AfterReconnect flag so consumers can distinguish post-recreate replay events from live ones. Typed errors ------------ ErrPullFailed, ErrRenewFailed, ErrRecreateFailed all implement Unwrap() and Op() Op. Consumers can branch with errors.As without parsing strings: var pull ErrPullFailed if errors.As(e, &pull) { /* transient; logged */ } var recreate ErrRecreateFailed if errors.As(e, &recreate) { /* alert: camera may be offline */ } Op() returns OpPull / OpRenew / OpRecreate for cases where the caller wants to log the operation name without unwrapping. Both addressed the review's 'highest-leverage v1 change' concern about bare error on the Errors channel. AfterReconnect observability ---------------------------- ONVIF cameras replay each property's current value with PropertyInitialized whenever a new pull-point subscription is established (per the Event Service spec). A consumer doing edge detection on motion = StateActive would otherwise see a phantom 'motion started' for every active property after every reconnect. The pull loop now tracks an afterReconnect flag local to the goroutine: set to true when attemptRecreate returns justRecreated, applied to every emitted event, cleared on the first non-Initialized event we see. This bounds the replay window naturally — once the camera has finished sending current state, the next event tells us we're live. attemptRecreate now returns (justRecreated, cont) so the pull loop knows whether the just-completed recreate succeeded vs. the call returning due to ctx-cancel during backoff. Test coverage ------------- * errors_test.go: typed-error Unwrap/Op assertions plus Stream-level proof that pull and recreate failures arrive on the Errors channel wearing the right type. * AfterReconnect flag: drives the stream through a failure, observes the next event carries the flag and the one after does not. --- event/stream/errors.go | 42 ++++++++++++ event/stream/errors_test.go | 133 ++++++++++++++++++++++++++++++++++++ event/stream/stream.go | 39 ++++++++--- event/stream/types.go | 9 +++ 4 files changed, 212 insertions(+), 11 deletions(-) create mode 100644 event/stream/errors.go create mode 100644 event/stream/errors_test.go diff --git a/event/stream/errors.go b/event/stream/errors.go new file mode 100644 index 0000000..bd80674 --- /dev/null +++ b/event/stream/errors.go @@ -0,0 +1,42 @@ +package stream + +import "fmt" + +// Op identifies which Stream operation failed. Used by ErrPullFailed, +// ErrRenewFailed and ErrRecreateFailed so consumers can branch with +// errors.As without parsing the wrapped message. +type Op string + +const ( + OpPull Op = "pull" + OpRenew Op = "renew" + OpRecreate Op = "recreate" +) + +// ErrPullFailed wraps a transient PullMessages failure. The pull loop +// surfaces it on the Errors channel and continues. Consumers can match +// with errors.As(err, &stream.ErrPullFailed{}) — see +// TestErrors_TypedAssertion. +type ErrPullFailed struct{ Err error } + +func (e ErrPullFailed) Error() string { return fmt.Sprintf("pull messages: %v", e.Err) } +func (e ErrPullFailed) Unwrap() error { return e.Err } +func (ErrPullFailed) Op() Op { return OpPull } + +// ErrRenewFailed wraps a Renew SOAP failure. Renew errors are usually +// recovered implicitly: the subscription dies, pull starts failing, and +// the reconnect logic recreates it. +type ErrRenewFailed struct{ Err error } + +func (e ErrRenewFailed) Error() string { return fmt.Sprintf("renew pull point: %v", e.Err) } +func (e ErrRenewFailed) Unwrap() error { return e.Err } +func (ErrRenewFailed) Op() Op { return OpRenew } + +// ErrRecreateFailed wraps a failed CreatePullPointSubscription during +// the reconnect path. The loop continues with exponential backoff; +// consumers seeing this repeatedly should consider the camera offline. +type ErrRecreateFailed struct{ Err error } + +func (e ErrRecreateFailed) Error() string { return fmt.Sprintf("recreate pull point: %v", e.Err) } +func (e ErrRecreateFailed) Unwrap() error { return e.Err } +func (ErrRecreateFailed) Op() Op { return OpRecreate } diff --git a/event/stream/errors_test.go b/event/stream/errors_test.go new file mode 100644 index 0000000..d68e347 --- /dev/null +++ b/event/stream/errors_test.go @@ -0,0 +1,133 @@ +package stream + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestTypedErrors_UnwrapAndOp(t *testing.T) { + inner := errors.New("boom") + tests := []struct { + name string + err error + op Op + }{ + {"pull", ErrPullFailed{Err: inner}, OpPull}, + {"renew", ErrRenewFailed{Err: inner}, OpRenew}, + {"recreate", ErrRecreateFailed{Err: inner}, OpRecreate}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + assert.True(t, errors.Is(tc.err, inner), "errors.Is should unwrap to inner") + assert.Contains(t, tc.err.Error(), "boom") + + // Each typed error exposes Op() for branch-without-string-parse. + if e, ok := tc.err.(interface{ Op() Op }); ok { + assert.Equal(t, tc.op, e.Op()) + } else { + t.Fatalf("%T does not expose Op()", tc.err) + } + }) + } +} + +func TestStream_PullErrorIsTypedErrPullFailed(t *testing.T) { + fc := newFakeCaller() + fc.queueCallMethod(createPullPointResp, nil) + fc.queueSendSoap("", errors.New("transient")) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + s, err := newStream(ctx, fc, Options{ + PullTimeout: 50 * time.Millisecond, + RetryBackoff: 10 * time.Millisecond, + InitialTermination: 30 * time.Second, + }) + require.NoError(t, err) + defer s.Close() + + select { + case e := <-s.Errors(): + var pullErr ErrPullFailed + require.True(t, errors.As(e, &pullErr), "expected ErrPullFailed, got %T: %v", e, e) + assert.Contains(t, pullErr.Err.Error(), "transient") + case <-time.After(time.Second): + t.Fatal("expected ErrPullFailed on Errors channel") + } +} + +func TestStream_RecreateErrorIsTypedErrRecreateFailed(t *testing.T) { + fc := newFakeCaller() + fc.queueCallMethod(createPullPointResp, nil) + fc.mu.Lock() + fc.defaultCall = fakeResp{err: errors.New("recreate fail")} + fc.defaultSendSoap = fakeResp{err: errors.New("pull fail")} + fc.mu.Unlock() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + s, err := newStream(ctx, fc, Options{ + PullTimeout: 10 * time.Millisecond, + ReconnectAfterFailures: 1, + RetryBackoff: 10 * time.Millisecond, + InitialTermination: 30 * time.Second, + }) + require.NoError(t, err) + defer s.Close() + + deadline := time.Now().Add(time.Second) + var sawRecreate bool + for time.Now().Before(deadline) && !sawRecreate { + select { + case e := <-s.Errors(): + var rec ErrRecreateFailed + if errors.As(e, &rec) { + sawRecreate = true + assert.Contains(t, rec.Err.Error(), "recreate fail") + } + case <-time.After(50 * time.Millisecond): + } + } + assert.True(t, sawRecreate, "expected at least one ErrRecreateFailed on Errors") +} + +func TestStream_AfterReconnectFlagSetOnFirstPostRecreateBatch(t *testing.T) { + fc := newFakeCaller() + fc.queueCallMethod(createPullPointResp, nil) + // Second create is the recreate. + fc.queueCallMethod(createPullPointRespAlt, nil) + + // First pull fails -> triggers recreate with ReconnectAfterFailures=1. + fc.queueSendSoap("", errors.New("transient")) + // First pull after recreate: a Changed motion event. The flag + // should be true, and should clear (because we received a + // non-Initialized event). + fc.queueSendSoap(pullMessagesResp(motionMsg("true")), nil) + // Second pull after recreate: another motion event. Flag should + // now be false. + fc.queueSendSoap(pullMessagesResp(motionMsg("false")), nil) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + s, err := newStream(ctx, fc, Options{ + PullTimeout: 50 * time.Millisecond, + ReconnectAfterFailures: 1, + RetryBackoff: 10 * time.Millisecond, + InitialTermination: 30 * time.Second, + }) + require.NoError(t, err) + defer s.Close() + + ev1 := receive(t, s.Events(), 2*time.Second) + assert.True(t, ev1.AfterReconnect, "first event after recreate must carry AfterReconnect=true") + assert.Equal(t, StateActive, ev1.State) + + ev2 := receive(t, s.Events(), 2*time.Second) + assert.False(t, ev2.AfterReconnect, "subsequent events should not carry AfterReconnect") + assert.Equal(t, StateInactive, ev2.State) +} diff --git a/event/stream/stream.go b/event/stream/stream.go index 9e4fd7a..f10c9c5 100644 --- a/event/stream/stream.go +++ b/event/stream/stream.go @@ -270,6 +270,7 @@ func (s *Stream) run(ctx context.Context) { func (s *Stream) pullLoop(ctx context.Context) { var failures int recreateBackoff := s.opts.RetryBackoff + var afterReconnect bool for { if ctx.Err() != nil { @@ -277,12 +278,16 @@ func (s *Stream) pullLoop(ctx context.Context) { } msgs, err := pullMessages(s.caller, s.getPullPoint(), s.opts) if err != nil { - s.surfaceError(err) + s.surfaceError(ErrPullFailed{Err: err}) failures++ if failures >= s.opts.ReconnectAfterFailures { - if !s.attemptRecreate(ctx, &failures, &recreateBackoff) { + justRecreated, cont := s.attemptRecreate(ctx, &failures, &recreateBackoff) + if !cont { return } + if justRecreated { + afterReconnect = true + } continue } if !sleepCtx(ctx, s.opts.RetryBackoff) { @@ -296,6 +301,17 @@ func (s *Stream) pullLoop(ctx context.Context) { observedAt := s.now() for _, m := range msgs { ev := Decode(m, s.opts.DeviceID, observedAt) + if afterReconnect { + ev.AfterReconnect = true + // ONVIF replays current state with + // PropertyInitialized on a new subscription. + // Clear the flag as soon as we see anything + // other than Initialized — at that point we + // have transitioned to live events. + if ev.Operation != PropertyInitialized { + afterReconnect = false + } + } select { case <-ctx.Done(): return @@ -306,26 +322,27 @@ func (s *Stream) pullLoop(ctx context.Context) { } // attemptRecreate calls CreatePullPointSubscription and on success -// installs the new endpoint atomically. Returns false if ctx was -// cancelled while waiting for backoff (caller should exit the run -// loop). -func (s *Stream) attemptRecreate(ctx context.Context, failures *int, backoff *time.Duration) bool { +// installs the new endpoint atomically. The first return is true when +// recreate succeeded just now (caller flags the next batch with +// AfterReconnect). The second return is false only if ctx was cancelled +// during backoff (caller should exit the run loop). +func (s *Stream) attemptRecreate(ctx context.Context, failures *int, backoff *time.Duration) (justRecreated, cont bool) { addr, err := createPullPoint(s.caller, s.opts) if err != nil { - s.surfaceError(fmt.Errorf("recreate pull point: %w", err)) + s.surfaceError(ErrRecreateFailed{Err: err}) if !sleepCtx(ctx, *backoff) { - return false + return false, false } *backoff *= 2 if *backoff > maxRecreateBackoff { *backoff = maxRecreateBackoff } - return true + return false, true } s.setPullPoint(addr) *failures = 0 *backoff = s.opts.RetryBackoff - return true + return true, true } // renewLoop refreshes the subscription before InitialTermination expires. @@ -349,7 +366,7 @@ func (s *Stream) renewLoop(ctx context.Context) { return case <-ticker.C: if err := renewPullPoint(s.caller, s.getPullPoint(), s.opts); err != nil { - s.surfaceError(fmt.Errorf("renew pull point: %w", err)) + s.surfaceError(ErrRenewFailed{Err: err}) } } } diff --git a/event/stream/types.go b/event/stream/types.go index fcc3612..9f1557d 100644 --- a/event/stream/types.go +++ b/event/stream/types.go @@ -159,4 +159,13 @@ type Event struct { // Timestamp for ordering and DeviceTime only for forensics or // cross-camera correlation when caller manages NTP. DeviceTime time.Time + // AfterReconnect is true for events delivered after the Stream + // silently recreated its pull-point subscription. ONVIF cameras + // replay each property's current value with PropertyInitialized on + // a new subscription, which would otherwise look like a flood of + // new state changes to a consumer doing edge-detection. Watch this + // flag to suppress duplicate handling, or treat it as a normal + // event if you only care about steady-state level. Cleared on the + // first event whose Operation is not PropertyInitialized. + AfterReconnect bool }