diff --git a/pkg/service/soundtouchweb/websocket.go b/pkg/service/soundtouchweb/websocket.go
index 7fedd7e0..94dee3de 100644
--- a/pkg/service/soundtouchweb/websocket.go
+++ b/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 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 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
diff --git a/pkg/service/soundtouchweb/webtypes/status_test.go b/pkg/service/soundtouchweb/webtypes/status_test.go
index 1ee38f84..083f920d 100644
--- a/pkg/service/soundtouchweb/webtypes/status_test.go
+++ b/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)
}
}
diff --git a/pkg/service/soundtouchweb/webtypes/types.go b/pkg/service/soundtouchweb/webtypes/types.go
index 01a74620..1647f601 100644
--- a/pkg/service/soundtouchweb/webtypes/types.go
+++ b/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