From 09a3992bac7bb329a059360c9ff5e3d186630ddc Mon Sep 17 00:00:00 2001 From: David Newhall II Date: Sun, 9 Aug 2026 14:31:13 -0700 Subject: [PATCH 1/2] Add locked accessors for the fields Refresh() replaces. Refresh() swaps Cameras, Groups and Info under the server's mutex, but that mutex is unexported, so the doc telling callers to "use Rlock() on this struct" was impossible to follow: every reader raced the refresh. Apps hit this whenever Refresh() runs from a retry loop or event handler while requests read the camera list. GetCameras(), GetInfo() and GetGroups() read those fields under the read lock and return a snapshot. The library's own event stream and file listing now use them too, since both run outside the refreshing goroutine. Co-authored-by: Cursor --- events.go | 13 ++++---- files.go | 9 ++++-- refresh_concurrent_test.go | 62 ++++++++++++++++++++++++++++++++++++++ securityspy.go | 34 +++++++++++++++++++-- securityspy_types.go | 22 ++++++++------ 5 files changed, 120 insertions(+), 20 deletions(-) create mode 100644 refresh_concurrent_test.go diff --git a/events.go b/events.go index 42139ea..756d88c 100644 --- a/events.go +++ b/events.go @@ -180,8 +180,8 @@ func (e *Events) custom(eventType EventType, eventID, cam int, msg string) { now := time.Now().Round(time.Second) var camera *Camera - if e.server.Cameras != nil { - camera = e.server.Cameras.ByNum(cam) + if cams := e.server.GetCameras(); cams != nil { + camera = cams.ByNum(cam) } e.enqueue(&Event{ @@ -373,8 +373,8 @@ func (e *Events) UnmarshalEvent(text string) *Event { newEvent.Msg = parts[3] gmtOffset := 0.0 - if e.server.Info != nil { - gmtOffset = e.server.Info.GmtOffset.Hours() + if info := e.server.GetInfo(); info != nil { + gmtOffset = info.GmtOffset.Hours() } eventTime = fmt.Sprintf("%v%+03.0f", parts[0], gmtOffset) @@ -393,10 +393,11 @@ func (e *Events) UnmarshalEvent(text string) *Event { // Parse the camera number. parts[2] = strings.TrimPrefix(parts[2], "CAM") - if parts[2] != "X" && e.server.Cameras != nil { + + if cams := e.server.GetCameras(); parts[2] != "X" && cams != nil { if cameraNum, err := strconv.Atoi(parts[2]); err != nil { newEvent.Errors = append(newEvent.Errors, ErrCAMParseFail) - } else if newEvent.Camera = e.server.Cameras.ByNum(cameraNum); newEvent.Camera == nil { + } else if newEvent.Camera = cams.ByNum(cameraNum); newEvent.Camera == nil { newEvent.Errors = append(newEvent.Errors, ErrCAMMissing) } } diff --git a/files.go b/files.go index d52b8fa..e85d296 100644 --- a/files.go +++ b/files.go @@ -106,8 +106,9 @@ func (f *Files) GetFile(name string) (*File, error) { file := &File{ Title: name, server: f.server, - GmtOffset: f.server.Info.GmtOffset.Duration, + GmtOffset: f.server.GetInfo().GmtOffset.Duration, } + cams := f.server.GetCameras() if fileExtSplit := strings.Split(name, "."); len(fileExtSplit) != fileParts { return file, ErrInvalidName @@ -115,7 +116,7 @@ func (f *Files) GetFile(name string) (*File, error) { return file, ErrInvalidName } else if file.Updated, err = time.Parse(FileDateFormat, nameDateSplit[0]); err != nil { return file, ErrInvalidName - } else if file.Camera = f.server.Cameras.ByName(nameDateSplit[len(nameDateSplit)-1]); file.Camera == nil { + } else if file.Camera = cams.ByName(nameDateSplit[len(nameDateSplit)-1]); file.Camera == nil { return file, ErrCAMMissing } else if file.Link.Type = "video/quicktime"; fileExtSplit[1] == "jpg" { file.Link.Type = "image/jpeg" @@ -190,9 +191,11 @@ func (f *Files) getFiles(cameraNums []int, start, end time.Time, fileTypes, cont return nil, fmt.Errorf("getting download: %w", err) } + cams := f.server.GetCameras() + for i := range feed.Entries { // Add the camera, server and file interfaces to every file entry. - feed.Entries[i].Camera = f.server.Cameras.ByNum(feed.Entries[i].CameraNum) + feed.Entries[i].Camera = cams.ByNum(feed.Entries[i].CameraNum) feed.Entries[i].server = f.server feed.Entries[i].GmtOffset = feed.GmtOffset.Duration entries = append(entries, feed.Entries[i]) diff --git a/refresh_concurrent_test.go b/refresh_concurrent_test.go new file mode 100644 index 0000000..f9ec394 --- /dev/null +++ b/refresh_concurrent_test.go @@ -0,0 +1,62 @@ +package securityspy_test + +import ( + "sync" + "testing" + + "github.com/stretchr/testify/require" +) + +// TestRefreshConcurrentReaders is a race-detector test. Refresh() replaces +// Cameras, Groups and Info, and apps commonly call it from a background retry +// loop or an event handler while requests read those same fields. The Get +// accessors are the only safe way to read them, so the readers below run for +// as long as the refresher does. +func TestRefreshConcurrentReaders(t *testing.T) { + t.Parallel() + + serverObj, _, _ := testServerWithCamera(t) + done := make(chan struct{}) + + var wait sync.WaitGroup + + wait.Go(func() { + defer close(done) + + for range 25 { + require.NoError(t, serverObj.Refresh()) + } + }) + + wait.Go(func() { + for { + select { + case <-done: + return + default: + } + + cams := serverObj.GetCameras() + require.NotNil(t, cams) + require.NotNil(t, cams.ByNum(3)) + require.NotEmpty(t, cams.All()) + } + }) + + wait.Go(func() { + for { + select { + case <-done: + return + default: + } + + info := serverObj.GetInfo() + require.NotNil(t, info) + _ = info.Version + _ = serverObj.GetGroups() + } + }) + + wait.Wait() +} diff --git a/securityspy.go b/securityspy.go index f1bf57b..79c4d01 100644 --- a/securityspy.go +++ b/securityspy.go @@ -49,12 +49,42 @@ func NewMust(config *server.Config) *Server { // Refresh gets fresh camera and serverInfo data from SecuritySpy, // run this after every action to keep the data pool up to date. -// This is not at all thread safe. Do not run this if other methods -// may run in a different go routine. +// It replaces the Cameras, Groups and Info fields, so other goroutines must +// read those through GetCameras(), GetGroups() and GetInfo() while this can run. func (s *Server) Refresh() error { return s.RefreshContext(context.Background()) } +// GetCameras returns the camera list. Use this instead of the Cameras field +// when another goroutine may call Refresh(), which replaces it. +// The returned *Cameras is a snapshot: a later refresh builds a new one. +func (s *Server) GetCameras() *Cameras { + s.mu.RLock() + defer s.mu.RUnlock() + + return s.Cameras +} + +// GetInfo returns the server info. Use this instead of the Info field when +// another goroutine may call Refresh(), which replaces it. +// The returned *ServerInfo is a snapshot: a later refresh builds a new one. +func (s *Server) GetInfo() *ServerInfo { + s.mu.RLock() + defer s.mu.RUnlock() + + return s.Info +} + +// GetGroups returns the camera groups. Use this instead of the Groups field +// when another goroutine may call Refresh(), which replaces it. +// The returned slice is a snapshot: a later refresh builds a new one. +func (s *Server) GetGroups() []*Group { + s.mu.RLock() + defer s.mu.RUnlock() + + return s.Groups +} + // RefreshContext gets fresh camera and serverInfo data from SecuritySpy with context support. func (s *Server) RefreshContext(ctx context.Context) error { //nolint:cyclop // schedule name wiring s.mu.Lock() diff --git a/securityspy_types.go b/securityspy_types.go index 7205d4b..26a73b9 100644 --- a/securityspy_types.go +++ b/securityspy_types.go @@ -13,9 +13,10 @@ import ( // Server is the main interface for this library. // Contains sub-interfaces for cameras, ptz, files & events // This is provided in exchange for a url, username and password. -// If your app calls Refresh(), it is your duty to use Rlock() on -// this struct if there's a chance you may call methods while -// Refresh() is running. +// +// Refresh() replaces Cameras, Groups and Info, so an app that calls it from +// one goroutine must read those three through GetCameras(), GetGroups() and +// GetInfo() everywhere else. Reading the fields directly races the refresh. type Server struct { *server.Config @@ -23,12 +24,15 @@ type Server struct { // // Deprecated: unused; video capture is pure Go and does not shell out to ffmpeg. Encoder string - Files *Files // Files interface. - Events *Events // Events interface. - Cameras *Cameras // Cameras & PTZ interfaces. - Groups []*Group // Camera groups from systemInfo (v6+). - Info *ServerInfo // ServerInfo struct (no methods). - mu sync.RWMutex // Lock for Refresh(). + Files *Files // Files interface. + Events *Events // Events interface. + // Cameras is replaced by Refresh(); use GetCameras() when a refresh can race the read. + Cameras *Cameras + // Groups is replaced by Refresh(); use GetGroups() when a refresh can race the read. + Groups []*Group // Camera groups from systemInfo (v6+). + // Info is replaced by Refresh(); use GetInfo() when a refresh can race the read. + Info *ServerInfo // ServerInfo struct (no methods). + mu sync.RWMutex // Guards the three fields Refresh() replaces. } // Group is a named camera group from ++systemInfo (v6+). From a3ee38f91d094000f9ee149737f08d026d98f964 Mon Sep 17 00:00:00 2001 From: David Newhall II Date: Sun, 9 Aug 2026 15:41:24 -0700 Subject: [PATCH 2/2] Hold the refresh lock only for the swap, not the request. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit RefreshContext held the write lock across the systemInfo round trip, so every accessor added in the previous commit would block for the length of that request — up to the client timeout when SecuritySpy is unreachable, which is exactly when an app's retry loop refreshes most often. Readers would have stalled behind a refresh that had nothing new to offer, while a perfectly good snapshot sat in memory. The request and camera wiring now build into locals and a separate mutex serializes refreshes, so the write lock covers only the three assignments. Co-authored-by: Cursor --- refresh_concurrent_test.go | 106 +++++++++++++++++++++++++++++++------ securityspy.go | 85 +++++++++++++++++------------ securityspy_types.go | 5 +- 3 files changed, 146 insertions(+), 50 deletions(-) diff --git a/refresh_concurrent_test.go b/refresh_concurrent_test.go index f9ec394..49b428e 100644 --- a/refresh_concurrent_test.go +++ b/refresh_concurrent_test.go @@ -1,7 +1,9 @@ package securityspy_test import ( + "net/http" "sync" + "sync/atomic" "testing" "github.com/stretchr/testify/require" @@ -12,6 +14,9 @@ import ( // loop or an event handler while requests read those same fields. The Get // accessors are the only safe way to read them, so the readers below run for // as long as the refresher does. +// +// The workers report with t.Errorf rather than require: FailNow may only be +// called from the goroutine running the test. func TestRefreshConcurrentReaders(t *testing.T) { t.Parallel() @@ -24,35 +29,47 @@ func TestRefreshConcurrentReaders(t *testing.T) { defer close(done) for range 25 { - require.NoError(t, serverObj.Refresh()) + err := serverObj.Refresh() + if err != nil { + t.Errorf("Refresh: %v", err) + + return + } } }) wait.Go(func() { - for { - select { - case <-done: + for !isDone(done) { + cams := serverObj.GetCameras() + if cams == nil { + t.Error("GetCameras returned nil during a refresh") + return - default: } - cams := serverObj.GetCameras() - require.NotNil(t, cams) - require.NotNil(t, cams.ByNum(3)) - require.NotEmpty(t, cams.All()) + if cams.ByNum(3) == nil { + t.Error("camera 3 went missing during a refresh") + + return + } + + if len(cams.All()) == 0 { + t.Error("camera list emptied during a refresh") + + return + } } }) wait.Go(func() { - for { - select { - case <-done: + for !isDone(done) { + info := serverObj.GetInfo() + if info == nil { + t.Error("GetInfo returned nil during a refresh") + return - default: } - info := serverObj.GetInfo() - require.NotNil(t, info) _ = info.Version _ = serverObj.GetGroups() } @@ -60,3 +77,62 @@ func TestRefreshConcurrentReaders(t *testing.T) { wait.Wait() } + +// TestRefreshDoesNotBlockReaders: a refresh holds the write lock only for the +// swap, so a slow (or hung) systemInfo request must not stall readers. The +// handler parks the second refresh mid-request; if the refresh held the lock +// across the round trip, the reads below would block until the test timed out. +func TestRefreshDoesNotBlockReaders(t *testing.T) { + t.Parallel() + + var ( + requests atomic.Int64 + parked = make(chan struct{}) + release = make(chan struct{}) + ) + + serverObj := newTestServer(t, func(resp http.ResponseWriter, req *http.Request) { + if req.URL.Path != systemInfoPath { + http.NotFound(resp, req) + + return + } + + if requests.Add(1) > 1 { // let the first refresh load the snapshot + close(parked) + <-release + } + + resp.Header().Set("Content-Type", "application/xml") + _, _ = resp.Write([]byte(testSystemInfoV6)) + }) + + require.NoError(t, serverObj.Refresh()) + + refreshed := make(chan error, 1) + go func() { refreshed <- serverObj.Refresh() }() + + <-parked + + // The refresh is parked mid-request; reads still come from the old snapshot. + cams := serverObj.GetCameras() + if cams == nil || cams.ByNum(3) == nil { + t.Error("readers were blocked by an in-flight refresh") + } + + if serverObj.GetInfo() == nil { + t.Error("GetInfo was blocked by an in-flight refresh") + } + + close(release) + require.NoError(t, <-refreshed) +} + +func isDone(done <-chan struct{}) bool { + select { + case <-done: + return true + default: + return false + } +} diff --git a/securityspy.go b/securityspy.go index 79c4d01..69626bd 100644 --- a/securityspy.go +++ b/securityspy.go @@ -86,61 +86,80 @@ func (s *Server) GetGroups() []*Group { } // RefreshContext gets fresh camera and serverInfo data from SecuritySpy with context support. -func (s *Server) RefreshContext(ctx context.Context) error { //nolint:cyclop // schedule name wiring - s.mu.Lock() - defer s.mu.Unlock() +// +// The systemInfo request and all the wiring happen off to the side, so readers +// keep serving the previous snapshot for the length of the round trip and only +// block for the swap at the end. A refresh that times out never stalls them. +func (s *Server) RefreshContext(ctx context.Context) error { + // refreshMu serializes refreshes with each other, mu only guards the swap. + s.refreshMu.Lock() + defer s.refreshMu.Unlock() var sysInfo systemInfo - if err := s.GetXMLContext(ctx, "++systemInfo", nil, &sysInfo); err != nil { + err := s.GetXMLContext(ctx, "++systemInfo", nil, &sysInfo) + if err != nil { return fmt.Errorf("getting systemInfo: %w", err) } - s.Info = sysInfo.Server - if s.Info == nil { - s.Info = &ServerInfo{} + info := sysInfo.Server + if info == nil { + info = &ServerInfo{} } - s.Cameras = &Cameras{cameras: sysInfo.cameras(), server: s} - s.Groups = sysInfo.GroupList.Groups - s.Info.Refreshed = time.Now() + info.Refreshed = time.Now() // Point all the unmarshalled data into an exported struct. Better-formatted data. - s.Info.ServerSchedules = sysInfo.schedules() - s.Info.SchedulePresets = sysInfo.schedulePresets() - s.Info.ScheduleOverrides = sysInfo.scheduleOverrides() - - for idx, cam := range s.Cameras.cameras { - s.Cameras.cameras[idx].server = s - if s.Cameras.cameras[idx].PTZ != nil { - s.Cameras.cameras[idx].PTZ.camera = s.Cameras.cameras[idx] + info.ServerSchedules = sysInfo.schedules() + info.SchedulePresets = sysInfo.schedulePresets() + info.ScheduleOverrides = sysInfo.scheduleOverrides() + + cameras := &Cameras{cameras: sysInfo.cameras(), server: s} + s.wireCameras(cameras.cameras, info) + + s.mu.Lock() + defer s.mu.Unlock() + + s.Info = info + s.Cameras = cameras + s.Groups = sysInfo.GroupList.Groups + + return nil +} + +// wireCameras points every camera back at the server and fills in the schedule +// names, which systemInfo only provides as IDs. Runs on a new camera list before +// it is published, so no lock is needed. +func (s *Server) wireCameras(cameras []*Camera, info *ServerInfo) { + for idx, cam := range cameras { + cameras[idx].server = s + if cameras[idx].PTZ != nil { + cameras[idx].PTZ.camera = cameras[idx] } - // Fill in the missing schedule names (all we have are IDs, so fetch the names from systemInfo) - if name, ok := s.Info.ServerSchedules[cam.ScheduleIDA.ID]; ok { - s.Cameras.cameras[idx].ScheduleIDA.Name = name + + if name, ok := info.ServerSchedules[cam.ScheduleIDA.ID]; ok { + cameras[idx].ScheduleIDA.Name = name } - if name, ok := s.Info.ServerSchedules[cam.ScheduleIDCC.ID]; ok { - s.Cameras.cameras[idx].ScheduleIDCC.Name = name + if name, ok := info.ServerSchedules[cam.ScheduleIDCC.ID]; ok { + cameras[idx].ScheduleIDCC.Name = name } - if name, ok := s.Info.ServerSchedules[cam.ScheduleIDMC.ID]; ok { - s.Cameras.cameras[idx].ScheduleIDMC.Name = name + if name, ok := info.ServerSchedules[cam.ScheduleIDMC.ID]; ok { + cameras[idx].ScheduleIDMC.Name = name } - if name, ok := s.Info.ScheduleOverrides[cam.ScheduleOverrideA.ID]; ok { - s.Cameras.cameras[idx].ScheduleOverrideA.Name = name + if name, ok := info.ScheduleOverrides[cam.ScheduleOverrideA.ID]; ok { + cameras[idx].ScheduleOverrideA.Name = name } - if name, ok := s.Info.ScheduleOverrides[cam.ScheduleOverrideCC.ID]; ok { - s.Cameras.cameras[idx].ScheduleOverrideCC.Name = name + if name, ok := info.ScheduleOverrides[cam.ScheduleOverrideCC.ID]; ok { + cameras[idx].ScheduleOverrideCC.Name = name } - if name, ok := s.Info.ScheduleOverrides[cam.ScheduleOverrideMC.ID]; ok { - s.Cameras.cameras[idx].ScheduleOverrideMC.Name = name + if name, ok := info.ScheduleOverrides[cam.ScheduleOverrideMC.ID]; ok { + cameras[idx].ScheduleOverrideMC.Name = name } } - - return nil } // GetScripts fetches and returns the list of script files. diff --git a/securityspy_types.go b/securityspy_types.go index 26a73b9..a493ed4 100644 --- a/securityspy_types.go +++ b/securityspy_types.go @@ -31,8 +31,9 @@ type Server struct { // Groups is replaced by Refresh(); use GetGroups() when a refresh can race the read. Groups []*Group // Camera groups from systemInfo (v6+). // Info is replaced by Refresh(); use GetInfo() when a refresh can race the read. - Info *ServerInfo // ServerInfo struct (no methods). - mu sync.RWMutex // Guards the three fields Refresh() replaces. + Info *ServerInfo // ServerInfo struct (no methods). + mu sync.RWMutex // Guards the three fields Refresh() replaces. + refreshMu sync.Mutex // Serializes refreshes; held across the systemInfo request. } // Group is a named camera group from ++systemInfo (v6+).