fix(service): make status-poll vs. push-event ordering per field, not per connection

BeginStatusPoll/ApplySpeakerEvent/CompleteStatusPoll gated an entire
poll's merge (NowPlaying/Volume/Presets/Sources/Bass/IsConnected) behind
one shared speakerEventGeneration counter. Any unrelated push event
during the poll's flight discarded the whole result -- not just the
field that event touched. Sources has no push event at all, so it could
go stale indefinitely under ordinary event traffic, defeating both call
sites that depend on this poll (the 30s fallback poll and the
post-reconnect refresh).

Replace it with StatusField + BeginFieldPoll/CompleteFieldPoll/
ApplyFieldEvent: the same two-counter (issued/applied) pattern already
used for Group, generalized to one instance per independently-racing
field via a small fixed-size array. A poll or event for one field can
now only ever supersede that same field, never a different one. This
also removes the map-based generation bookkeeping the old mechanism
needed (issue/lookup/prune per poll) and the unreachable
"unknown generation" branch it required.

applyGroupUpdatedEvent now shares the same queueBroadcastIfChanged
helper as the other five event types, instead of duplicating the
"broadcast if changed" check inline.

Found in code review of PR #666 (findings #1, #2, #3, #4).

pkg/service/soundtouchweb/websocket.go

@@ -159,18 +159,12 @@ func (app *WebApp) registerGlobalWebSocket(conn *websocket.Conn) error {
 	})
 }
 
-// applySpeakerStatusEvent stores one speaker event and immediately publishes
-// a fresh device projection when its dashboard-visible payload changed.
-func (app *WebApp) applySpeakerStatusEvent(
-	conn *webtypes.DeviceConnection,
-	mut func(*webtypes.DeviceStatus) bool,
-) bool {
-	changed := false
-
-	conn.ApplySpeakerEvent(func(status *webtypes.DeviceStatus) {
-		changed = mut(status)
-	})
-
+// queueBroadcastIfChanged schedules a device-list broadcast when an
+// apply-event/apply-poll call changed something dashboard-visible. Every
+// such call site (including Group's own, separately-ordered path) routes
+// through this so a future change to the shared policy has one place to
+// change.
+func (app *WebApp) queueBroadcastIfChanged(changed bool) bool {
 	if changed {
 		app.QueueDeviceListBroadcast()
 	}
@@ -178,11 +172,29 @@ func (app *WebApp) applySpeakerStatusEvent(
 	return changed
 }
 
+// applySpeakerStatusEvent stores one speaker event for field and immediately
+// publishes a fresh device projection when its dashboard-visible payload
+// changed. Ordering against a concurrent poll of the same field is handled
+// by ApplyFieldEvent; a different field's poll or event is never affected.
+func (app *WebApp) applySpeakerStatusEvent(
+	conn *webtypes.DeviceConnection,
+	field webtypes.StatusField,
+	mut func(*webtypes.DeviceStatus) bool,
+) bool {
+	changed := false
+
+	conn.ApplyFieldEvent(field, func(status *webtypes.DeviceStatus) {
+		changed = mut(status)
+	})
+
+	return app.queueBroadcastIfChanged(changed)
+}
+
 func (app *WebApp) applyNowPlayingEvent(
 	conn *webtypes.DeviceConnection,
 	nowPlaying *models.NowPlaying,
 ) bool {
-	return app.applySpeakerStatusEvent(conn, func(status *webtypes.DeviceStatus) bool {
+	return app.applySpeakerStatusEvent(conn, webtypes.FieldNowPlaying, func(status *webtypes.DeviceStatus) bool {
 		changed := !reflect.DeepEqual(status.NowPlaying, nowPlaying)
 		status.NowPlaying = nowPlaying
 		status.LastActivity = time.Now()
@@ -195,7 +207,7 @@ func (app *WebApp) applyVolumeEvent(
 	conn *webtypes.DeviceConnection,
 	volume *models.Volume,
 ) bool {
-	return app.applySpeakerStatusEvent(conn, func(status *webtypes.DeviceStatus) bool {
+	return app.applySpeakerStatusEvent(conn, webtypes.FieldVolume, func(status *webtypes.DeviceStatus) bool {
 		changed := !reflect.DeepEqual(status.Volume, volume)
 		status.Volume = volume
 		status.LastActivity = time.Now()
@@ -208,7 +220,7 @@ func (app *WebApp) applyConnectionStateEvent(
 	conn *webtypes.DeviceConnection,
 	connected bool,
 ) bool {
-	return app.applySpeakerStatusEvent(conn, func(status *webtypes.DeviceStatus) bool {
+	return app.applySpeakerStatusEvent(conn, webtypes.FieldConnectivity, func(status *webtypes.DeviceStatus) bool {
 		changed := status.IsConnected != connected
 		status.IsConnected = connected
 		status.LastActivity = time.Now()
@@ -221,7 +233,7 @@ func (app *WebApp) applyPresetEvent(
 	conn *webtypes.DeviceConnection,
 	presets *models.Presets,
 ) bool {
-	return app.applySpeakerStatusEvent(conn, func(status *webtypes.DeviceStatus) bool {
+	return app.applySpeakerStatusEvent(conn, webtypes.FieldPresets, func(status *webtypes.DeviceStatus) bool {
 		changed := !reflect.DeepEqual(status.Presets, presets)
 		status.Presets = presets
 		status.LastActivity = time.Now()
@@ -234,7 +246,7 @@ func (app *WebApp) applyBassEvent(
 	conn *webtypes.DeviceConnection,
 	bass *models.Bass,
 ) bool {
-	return app.applySpeakerStatusEvent(conn, func(status *webtypes.DeviceStatus) bool {
+	return app.applySpeakerStatusEvent(conn, webtypes.FieldBass, func(status *webtypes.DeviceStatus) bool {
 		changed := !reflect.DeepEqual(status.Bass, bass)
 		status.Bass = bass
 		status.LastActivity = time.Now()
@@ -496,18 +508,26 @@ func sleepOrDone(conn *webtypes.DeviceConnection, d time.Duration) bool {
 
 // UpdateDeviceStatus fetches current status from the device.
 //
-// Network calls run outside the atomic merge so the CAS loop in
-// UpdateStatus stays fast and doesn't retry slow IO. WebSocket event
-// handlers running concurrently are not lost: their UpdateStatus
-// runs against whichever snapshot they observe, and the merge below
-// sees their changes when it CAS-loops onto the latest status.
+// Network calls run outside any atomic merge so slow I/O never blocks a
+// concurrent CAS retry. Each field (NowPlaying/Volume/Presets/Sources/Bass,
+// plus derived connectivity) is merged and ordered independently via its own
+// StatusField generation (BeginFieldPoll/CompleteFieldPoll/ApplyFieldEvent in
+// webtypes) -- a real-time push event, or a concurrent poll, for one field
+// can supersede only that field. A slow-but-successful fetch for one field
+// is never discarded merely because a DIFFERENT field's event or poll
+// completion happened to land first.
 func (app *WebApp) UpdateDeviceStatus(_ string, conn *webtypes.DeviceConnection) {
 	// Skip status update if client is not available (e.g., in tests)
 	if conn.Client == nil {
 		return
 	}
 
-	pollGeneration := conn.BeginStatusPoll()
+	nowPlayingGen := conn.BeginFieldPoll(webtypes.FieldNowPlaying)
+	volumeGen := conn.BeginFieldPoll(webtypes.FieldVolume)
+	presetsGen := conn.BeginFieldPoll(webtypes.FieldPresets)
+	sourcesGen := conn.BeginFieldPoll(webtypes.FieldSources)
+	bassGen := conn.BeginFieldPoll(webtypes.FieldBass)
+	connectivityGen := conn.BeginFieldPoll(webtypes.FieldConnectivity)
 
 	// /getGroup must be gated to ST10 models -- see Client.GetGroup's doc
 	// comment (verified against real hardware: a ST20 never replies at all,
@@ -537,45 +557,65 @@ func (app *WebApp) UpdateDeviceStatus(_ string, conn *webtypes.DeviceConnection)
 		group, groupErr = conn.Client.GetGroup()
 	}
 
-	// Phase 2: fast merge. Only fields we successfully fetched
-	// overwrite; everything else keeps the value other goroutines may
-	// have just written.
-	conn.CompleteStatusPoll(pollGeneration, func(s *webtypes.DeviceStatus) {
-		statusUpdated := false
+	// Phase 2: fast, independently-ordered merges. Each field applies only
+	// if this round's fetch succeeded AND no newer poll or push event has
+	// already applied for that specific field.
+	anyFetchSucceeded := false
 
-		if nowPlayingErr == nil {
+	if nowPlayingErr == nil {
+		anyFetchSucceeded = true
+
+		conn.CompleteFieldPoll(webtypes.FieldNowPlaying, nowPlayingGen, func(s *webtypes.DeviceStatus) {
 			s.NowPlaying = nowPlaying
-			statusUpdated = true
-		}
+			s.LastActivity = time.Now()
+		})
+	}
 
-		if volumeErr == nil {
+	if volumeErr == nil {
+		anyFetchSucceeded = true
+
+		conn.CompleteFieldPoll(webtypes.FieldVolume, volumeGen, func(s *webtypes.DeviceStatus) {
 			s.Volume = volume
-			statusUpdated = true
-		}
+			s.LastActivity = time.Now()
+		})
+	}
 
-		if presetsErr == nil {
+	if presetsErr == nil {
+		anyFetchSucceeded = true
+
+		conn.CompleteFieldPoll(webtypes.FieldPresets, presetsGen, func(s *webtypes.DeviceStatus) {
 			s.Presets = presets
-			statusUpdated = true
-		}
+			s.LastActivity = time.Now()
+		})
+	}
 
-		if sourcesErr == nil {
+	if sourcesErr == nil {
+		anyFetchSucceeded = true
+
+		conn.CompleteFieldPoll(webtypes.FieldSources, sourcesGen, func(s *webtypes.DeviceStatus) {
 			s.Sources = sources
-			statusUpdated = true
-		}
+			s.LastActivity = time.Now()
+		})
+	}
 
-		if bassErr == nil {
+	if bassErr == nil {
+		anyFetchSucceeded = true
+
+		conn.CompleteFieldPoll(webtypes.FieldBass, bassGen, func(s *webtypes.DeviceStatus) {
 			s.Bass = bass
-			statusUpdated = true
-		}
+			s.LastActivity = time.Now()
+		})
+	}
 
-		// Mark as connected if we successfully got at least one status
-		// from this round. Mirrors prior behaviour: deliberately does NOT
-		// fold groupErr in here. GetGroup is gated to stereo-capable
-		// models and trivially succeeds even when a device is otherwise
-		// struggling (an empty <group/> is a near-guaranteed reply), so
-		// counting it would let a device report connected while every
-		// substantive status fetch above actually failed this round.
-		s.IsConnected = statusUpdated
+	// Mark as connected if we successfully got at least one status from
+	// this round. Mirrors prior behaviour: deliberately does NOT fold
+	// groupErr in here. GetGroup is gated to stereo-capable models and
+	// trivially succeeds even when a device is otherwise struggling (an
+	// empty <group/> is a near-guaranteed reply), so counting it would let
+	// a device report connected while every substantive status fetch above
+	// actually failed this round.
+	conn.CompleteFieldPoll(webtypes.FieldConnectivity, connectivityGen, func(s *webtypes.DeviceStatus) {
+		s.IsConnected = anyFetchSucceeded
 		s.LastActivity = time.Now()
 	})
 
@@ -588,12 +628,7 @@ func (app *WebApp) applyGroupUpdatedEvent(
 	conn *webtypes.DeviceConnection,
 	event *models.GroupUpdatedEvent,
 ) bool {
-	changed := conn.ApplyGroupEvent(&event.Group, time.Now())
-	if changed {
-		app.QueueDeviceListBroadcast()
-	}
-
-	return changed
+	return app.queueBroadcastIfChanged(conn.ApplyGroupEvent(&event.Group, time.Now()))
 }
 
 // HandleDeviceWebSocket handles individual device WebSocket connections for real-time device-specific updates

pkg/service/soundtouchweb/webtypes/status_test.go

@@ -248,40 +248,40 @@ func TestApplyGroupEventIgnoresRoleOrder(t *testing.T) {
 	}
 }
 
-func TestStatusPollCannotOverwriteNewerSpeakerEvent(t *testing.T) {
+func TestFieldPollCannotOverwriteNewerFieldEvent(t *testing.T) {
 	conn := NewDeviceConnection(nil, &models.DeviceInfo{Name: "test"})
-	poll := conn.BeginStatusPoll()
+	poll := conn.BeginFieldPoll(FieldVolume)
 
-	conn.ApplySpeakerEvent(func(status *DeviceStatus) {
+	conn.ApplyFieldEvent(FieldVolume, func(status *DeviceStatus) {
 		status.Volume = &models.Volume{ActualVolume: 99}
 	})
 
-	if conn.CompleteStatusPoll(poll, func(status *DeviceStatus) {
+	if conn.CompleteFieldPoll(FieldVolume, poll, func(status *DeviceStatus) {
 		status.Volume = &models.Volume{ActualVolume: 42}
 	}) {
-		t.Fatal("poll that began before a speaker event was applied")
+		t.Fatal("poll that began before a same-field event was applied")
 	}
 
 	if got := conn.Status().Volume; got == nil || got.ActualVolume != 99 {
-		t.Fatalf("speaker event was overwritten by older poll data: %+v", got)
+		t.Fatalf("field event was overwritten by older poll data: %+v", got)
 	}
 }
 
-func TestNewerStatusPollSupersedesOlderPoll(t *testing.T) {
+func TestNewerFieldPollSupersedesOlderFieldPoll(t *testing.T) {
 	conn := NewDeviceConnection(nil, &models.DeviceInfo{Name: "test"})
-	older := conn.BeginStatusPoll()
-	newer := conn.BeginStatusPoll()
+	older := conn.BeginFieldPoll(FieldVolume)
+	newer := conn.BeginFieldPoll(FieldVolume)
 
-	if !conn.CompleteStatusPoll(newer, func(status *DeviceStatus) {
+	if !conn.CompleteFieldPoll(FieldVolume, newer, func(status *DeviceStatus) {
 		status.Volume = &models.Volume{ActualVolume: 30}
 	}) {
 		t.Fatal("newer poll was not applied")
 	}
 
-	if conn.CompleteStatusPoll(older, func(status *DeviceStatus) {
+	if conn.CompleteFieldPoll(FieldVolume, older, func(status *DeviceStatus) {
 		status.Volume = &models.Volume{ActualVolume: 10}
 	}) {
-		t.Fatal("older poll completed after a newer poll was applied")
+		t.Fatal("older poll completed after a newer same-field poll was applied")
 	}
 
 	if got := conn.Status().Volume; got == nil || got.ActualVolume != 30 {
@@ -289,20 +289,51 @@ func TestNewerStatusPollSupersedesOlderPoll(t *testing.T) {
 	}
 }
 
-func TestDuplicateSpeakerEventStillInvalidatesOlderPoll(t *testing.T) {
+func TestDuplicateFieldEventStillInvalidatesOlderSameFieldPoll(t *testing.T) {
 	conn := NewDeviceConnection(nil, &models.DeviceInfo{Name: "test"})
 	conn.SetStatus(&DeviceStatus{Volume: &models.Volume{ActualVolume: 25}})
-	poll := conn.BeginStatusPoll()
+	poll := conn.BeginFieldPoll(FieldVolume)
 
-	conn.ApplySpeakerEvent(func(status *DeviceStatus) {
+	conn.ApplyFieldEvent(FieldVolume, func(status *DeviceStatus) {
 		status.Volume = &models.Volume{ActualVolume: 25}
 		status.LastActivity = time.Now()
 	})
 
-	if conn.CompleteStatusPoll(poll, func(status *DeviceStatus) {
+	if conn.CompleteFieldPoll(FieldVolume, poll, func(status *DeviceStatus) {
 		status.Volume = &models.Volume{ActualVolume: 10}
 	}) {
-		t.Fatal("poll that preceded duplicate speaker evidence was applied")
+		t.Fatal("poll that preceded duplicate same-field event evidence was applied")
+	}
+}
+
+// TestUnrelatedFieldEventDoesNotInvalidateInFlightPoll guards the actual
+// production bug this per-field design replaces a connection-wide gate to
+// fix: a push event for one field must never discard a DIFFERENT field's
+// still-in-flight, ultimately-successful poll. Sources in particular has no
+// push event at all (see StatusField's doc comment) and would go stale
+// indefinitely under ordinary event traffic if this regressed.
+func TestUnrelatedFieldEventDoesNotInvalidateInFlightPoll(t *testing.T) {
+	conn := NewDeviceConnection(nil, &models.DeviceInfo{Name: "test"})
+	sourcesPoll := conn.BeginFieldPoll(FieldSources)
+
+	// An unrelated field's event fires while the Sources poll is still in
+	// flight.
+	conn.ApplyFieldEvent(FieldVolume, func(status *DeviceStatus) {
+		status.Volume = &models.Volume{ActualVolume: 50}
+	})
+
+	if !conn.CompleteFieldPoll(FieldSources, sourcesPoll, func(status *DeviceStatus) {
+		status.Sources = &models.Sources{}
+	}) {
+		t.Fatal("an unrelated field's event must not invalidate this field's in-flight poll")
+	}
+
+	if got := conn.Status().Sources; got == nil {
+		t.Fatal("Sources poll result was discarded by an unrelated field's event")
+	}
+
+	if got := conn.Status().Volume; got == nil || got.ActualVolume != 50 {
+		t.Fatalf("unrelated field event itself was lost: %+v", got)
 	}
 }
 

pkg/service/soundtouchweb/webtypes/types.go

@@ -47,11 +47,13 @@ type DeviceConnection struct {
 
 	status atomic.Pointer[DeviceStatus]
 
-	statusOrderMu             sync.Mutex
-	nextStatusPollGeneration  uint64
-	lastStatusPollGeneration  uint64
-	speakerEventGeneration    uint64
-	statusPollEventGeneration map[uint64]uint64
+	// fieldGenMu guards fieldGen, the per-field generation ordering used by
+	// BeginFieldPoll/CompleteFieldPoll/ApplyFieldEvent. Each StatusField gets
+	// its own (issued, applied) pair so an event or a poll completion for
+	// one field can never invalidate a different field's in-flight result --
+	// see StatusField's doc comment.
+	fieldGenMu sync.Mutex
+	fieldGen   [numStatusFields]struct{ issued, applied uint64 }
 
 	// groupMu orders polled /getGroup responses against real-time
 	// groupUpdated events. groupGeneration is the highest generation
@@ -86,6 +88,27 @@ type DeviceStatus struct {
 	LastActivity time.Time          `json:"lastActivity"`
 }
 
+// StatusField identifies one independently-racing field of DeviceStatus for
+// BeginFieldPoll/CompleteFieldPoll/ApplyFieldEvent's generation ordering.
+// FieldConnectivity covers IsConnected, which is derived by a poll (from
+// whether any of the other fields' fetches succeeded) but set directly by a
+// real-time connectionStateUpdated event, so it races the same way the other
+// fields do. Group has its own, pre-existing pair-aware ordering
+// (groupMu/groupGeneration/groupAppliedGeneration below) and is not part of
+// this set.
+type StatusField int
+
+// The StatusField values.
+const (
+	FieldNowPlaying StatusField = iota
+	FieldVolume
+	FieldPresets
+	FieldSources
+	FieldBass
+	FieldConnectivity
+	numStatusFields
+)
+
 // NewDeviceConnection creates a fully-initialised connection. The
 // status starts with IsConnected=false and LastActivity set to now;
 // real values arrive via UpdateStatus once the device responds.
@@ -141,66 +164,52 @@ func (c *DeviceConnection) SetStatus(s *DeviceStatus) {
 	c.status.Store(s)
 }
 
-// BeginStatusPoll reserves an ordering generation before a status poll starts
-// network I/O. CompleteStatusPoll uses it to reject an older poll after either
-// a newer poll or a real-time speaker event has supplied fresher state.
-func (c *DeviceConnection) BeginStatusPoll() uint64 {
-	c.statusOrderMu.Lock()
-	defer c.statusOrderMu.Unlock()
+// BeginFieldPoll reserves a generation for an asynchronous fetch of field,
+// before any network I/O starts. Pass the returned value to CompleteFieldPoll
+// once the fetch completes.
+func (c *DeviceConnection) BeginFieldPoll(field StatusField) uint64 {
+	c.fieldGenMu.Lock()
+	defer c.fieldGenMu.Unlock()
 
-	c.nextStatusPollGeneration++
-	if c.statusPollEventGeneration == nil {
-		c.statusPollEventGeneration = make(map[uint64]uint64)
-	}
+	c.fieldGen[field].issued++
 
-	c.statusPollEventGeneration[c.nextStatusPollGeneration] = c.speakerEventGeneration
-
-	return c.nextStatusPollGeneration
+	return c.fieldGen[field].issued
 }
 
-// ApplySpeakerEvent serializes a real-time speaker event against status-poll
-// completion. Even a duplicate event invalidates polls that started before it:
-// the event is newer evidence than their fetched payload.
-func (c *DeviceConnection) ApplySpeakerEvent(mut func(*DeviceStatus)) {
-	c.statusOrderMu.Lock()
-	defer c.statusOrderMu.Unlock()
+// CompleteFieldPoll applies mut only if generation is strictly newer than
+// whatever last actually applied -- poll or event -- for field. A poll for
+// one field losing this race never affects any other field: an unrelated
+// field's event, or an unrelated field's poll completing first, cannot
+// discard this field's fresh, successful data.
+func (c *DeviceConnection) CompleteFieldPoll(field StatusField, generation uint64, mut func(*DeviceStatus)) bool {
+	c.fieldGenMu.Lock()
 
-	c.speakerEventGeneration++
-	c.UpdateStatus(mut)
-}
+	if generation <= c.fieldGen[field].applied {
+		c.fieldGenMu.Unlock()
 
-// CompleteStatusPoll applies a poll result only when no newer poll has already
-// completed and no speaker event arrived after this poll began.
-func (c *DeviceConnection) CompleteStatusPoll(
-	generation uint64,
-	mut func(*DeviceStatus),
-) bool {
-	c.statusOrderMu.Lock()
-	defer c.statusOrderMu.Unlock()
-
-	eventGeneration, knownGeneration := c.statusPollEventGeneration[generation]
-	delete(c.statusPollEventGeneration, generation)
-
-	if generation <= c.lastStatusPollGeneration {
 		return false
 	}
 
-	c.lastStatusPollGeneration = generation
-	for olderGeneration := range c.statusPollEventGeneration {
-		if olderGeneration < generation {
-			delete(c.statusPollEventGeneration, olderGeneration)
-		}
-	}
-
-	if !knownGeneration || eventGeneration != c.speakerEventGeneration {
-		return false
-	}
+	c.fieldGen[field].applied = generation
+	c.fieldGenMu.Unlock()
 
 	c.UpdateStatus(mut)
 
 	return true
 }
 
+// ApplyFieldEvent always applies mut -- a real-time push event is
+// authoritative evidence for field -- and invalidates any poll for field
+// that began before it, without touching any other field's ordering.
+func (c *DeviceConnection) ApplyFieldEvent(field StatusField, mut func(*DeviceStatus)) {
+	c.fieldGenMu.Lock()
+	c.fieldGen[field].issued++
+	c.fieldGen[field].applied = c.fieldGen[field].issued
+	c.fieldGenMu.Unlock()
+
+	c.UpdateStatus(mut)
+}
+
 // UpdateStatus atomically applies mut to a copy of the current status
 // and stores the result. If another goroutine updates the status while
 // mut runs, UpdateStatus retries with the newer status — so concurrent