fix(player): publish speaker events immediately

pkg/service/soundtouchweb/handler.go

@@ -56,6 +56,10 @@ type WebApp struct {
 	// client cannot indefinitely block updates for healthy clients.
 	webSocketWriteTimeout time.Duration
 
+	deviceBroadcastMu      sync.Mutex
+	deviceBroadcastPending bool
+	deviceBroadcastRunning bool
+
 	Version    string
 	Commit     string
 	Date       string
@@ -773,6 +777,43 @@ func (app *WebApp) BroadcastDeviceList() {
 	}
 }
 
+// QueueDeviceListBroadcast schedules one device projection without blocking a
+// speaker's event read loop on browser I/O. At most one worker runs per app;
+// events during a slow write are coalesced into one follow-up snapshot rather
+// than expanding into an unbounded queue or set of goroutines.
+func (app *WebApp) QueueDeviceListBroadcast() {
+	app.deviceBroadcastMu.Lock()
+
+	app.deviceBroadcastPending = true
+	if app.deviceBroadcastRunning {
+		app.deviceBroadcastMu.Unlock()
+
+		return
+	}
+
+	app.deviceBroadcastRunning = true
+	app.deviceBroadcastMu.Unlock()
+
+	go app.runDeviceListBroadcasts()
+}
+
+func (app *WebApp) runDeviceListBroadcasts() {
+	for {
+		app.deviceBroadcastMu.Lock()
+		if !app.deviceBroadcastPending {
+			app.deviceBroadcastRunning = false
+			app.deviceBroadcastMu.Unlock()
+
+			return
+		}
+
+		app.deviceBroadcastPending = false
+		app.deviceBroadcastMu.Unlock()
+
+		app.BroadcastDeviceList()
+	}
+}
+
 // BroadcastDiscoveryStatus sends discovery progress updates to all connected WebSocket clients
 func (app *WebApp) BroadcastDiscoveryStatus(status string, deviceCount int) {
 	discoveryStatus := &webtypes.DiscoveryStatus{

pkg/service/soundtouchweb/websocket.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"log"
 	"net/http"
+	"reflect"
 	"sync"
 	"time"
 
@@ -158,6 +159,90 @@ 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)
+	})
+
+	if changed {
+		app.QueueDeviceListBroadcast()
+	}
+
+	return changed
+}
+
+func (app *WebApp) applyNowPlayingEvent(
+	conn *webtypes.DeviceConnection,
+	nowPlaying *models.NowPlaying,
+) bool {
+	return app.applySpeakerStatusEvent(conn, func(status *webtypes.DeviceStatus) bool {
+		changed := !reflect.DeepEqual(status.NowPlaying, nowPlaying)
+		status.NowPlaying = nowPlaying
+		status.LastActivity = time.Now()
+
+		return changed
+	})
+}
+
+func (app *WebApp) applyVolumeEvent(
+	conn *webtypes.DeviceConnection,
+	volume *models.Volume,
+) bool {
+	return app.applySpeakerStatusEvent(conn, func(status *webtypes.DeviceStatus) bool {
+		changed := !reflect.DeepEqual(status.Volume, volume)
+		status.Volume = volume
+		status.LastActivity = time.Now()
+
+		return changed
+	})
+}
+
+func (app *WebApp) applyConnectionStateEvent(
+	conn *webtypes.DeviceConnection,
+	connected bool,
+) bool {
+	return app.applySpeakerStatusEvent(conn, func(status *webtypes.DeviceStatus) bool {
+		changed := status.IsConnected != connected
+		status.IsConnected = connected
+		status.LastActivity = time.Now()
+
+		return changed
+	})
+}
+
+func (app *WebApp) applyPresetEvent(
+	conn *webtypes.DeviceConnection,
+	presets *models.Presets,
+) bool {
+	return app.applySpeakerStatusEvent(conn, func(status *webtypes.DeviceStatus) bool {
+		changed := !reflect.DeepEqual(status.Presets, presets)
+		status.Presets = presets
+		status.LastActivity = time.Now()
+
+		return changed
+	})
+}
+
+func (app *WebApp) applyBassEvent(
+	conn *webtypes.DeviceConnection,
+	bass *models.Bass,
+) bool {
+	return app.applySpeakerStatusEvent(conn, func(status *webtypes.DeviceStatus) bool {
+		changed := !reflect.DeepEqual(status.Bass, bass)
+		status.Bass = bass
+		status.LastActivity = time.Now()
+
+		return changed
+	})
+}
+
 // HandleWebSocket handles WebSocket connections for real-time updates
 func (app *WebApp) HandleWebSocket(w http.ResponseWriter, r *http.Request) {
 	conn, err := app.Upgrader.Upgrade(w, r, nil)
@@ -323,35 +408,27 @@ func (app *WebApp) ConnectDeviceWebSocket(deviceID string, conn *webtypes.Device
 
 			prevSource = np.Source
 
-			conn.UpdateStatus(func(s *webtypes.DeviceStatus) {
-				s.NowPlaying = np
-				s.LastActivity = time.Now()
-			})
+			app.applyNowPlayingEvent(conn, np)
 		})
 
 		wsClient.OnVolumeUpdated(func(event *models.VolumeUpdatedEvent) {
-			conn.UpdateStatus(func(s *webtypes.DeviceStatus) {
-				s.Volume = &event.Volume
-				s.LastActivity = time.Now()
-			})
+			app.applyVolumeEvent(conn, &event.Volume)
 		})
 
 		wsClient.OnConnectionState(func(event *models.ConnectionStateUpdatedEvent) {
-			conn.UpdateStatus(func(s *webtypes.DeviceStatus) {
-				s.IsConnected = event.ConnectionState.IsConnected()
-				s.LastActivity = time.Now()
-			})
+			app.applyConnectionStateEvent(conn, event.ConnectionState.IsConnected())
 		})
 
 		wsClient.OnPresetUpdated(func(event *models.PresetUpdatedEvent) {
-			conn.UpdateStatus(func(s *webtypes.DeviceStatus) {
-				s.Presets = &event.Presets
-				s.LastActivity = time.Now()
-			})
+			app.applyPresetEvent(conn, &event.Presets)
+		})
+
+		wsClient.OnBassUpdated(func(event *models.BassUpdatedEvent) {
+			app.applyBassEvent(conn, &event.Bass)
 		})
 
 		wsClient.OnGroupUpdated(func(event *models.GroupUpdatedEvent) {
-			applyGroupUpdatedEvent(conn, event)
+			app.applyGroupUpdatedEvent(conn, event)
 		})
 
 		if err := wsClient.Connect(); err != nil {
@@ -371,9 +448,7 @@ func (app *WebApp) ConnectDeviceWebSocket(deviceID string, conn *webtypes.Device
 
 		conn.WebSocket = wsClient
 
-		conn.UpdateStatus(func(s *webtypes.DeviceStatus) {
-			s.IsConnected = true
-		})
+		app.applyConnectionStateEvent(conn, true)
 
 		log.Printf("WebSocket connected for device %s", sanitizeLog(deviceID))
 
@@ -389,9 +464,7 @@ func (app *WebApp) ConnectDeviceWebSocket(deviceID string, conn *webtypes.Device
 		// Block until the device-side WebSocket disconnects.
 		wsClient.Wait()
 
-		conn.UpdateStatus(func(s *webtypes.DeviceStatus) {
-			s.IsConnected = false
-		})
+		app.applyConnectionStateEvent(conn, false)
 
 		log.Printf("WebSocket disconnected for device %s — reconnecting in %s", sanitizeLog(deviceID), backoff)
 
@@ -434,6 +507,8 @@ func (app *WebApp) UpdateDeviceStatus(_ string, conn *webtypes.DeviceConnection)
 		return
 	}
 
+	pollGeneration := conn.BeginStatusPoll()
+
 	// /getGroup must be gated to ST10 models -- see Client.GetGroup's doc
 	// comment (verified against real hardware: a ST20 never replies at all,
 	// hanging until the client's timeout instead of returning quickly).
@@ -465,7 +540,7 @@ func (app *WebApp) UpdateDeviceStatus(_ string, conn *webtypes.DeviceConnection)
 	// Phase 2: fast merge. Only fields we successfully fetched
 	// overwrite; everything else keeps the value other goroutines may
 	// have just written.
-	conn.UpdateStatus(func(s *webtypes.DeviceStatus) {
+	conn.CompleteStatusPoll(pollGeneration, func(s *webtypes.DeviceStatus) {
 		statusUpdated := false
 
 		if nowPlayingErr == nil {
@@ -509,8 +584,16 @@ func (app *WebApp) UpdateDeviceStatus(_ string, conn *webtypes.DeviceConnection)
 	}
 }
 
-func applyGroupUpdatedEvent(conn *webtypes.DeviceConnection, event *models.GroupUpdatedEvent) {
-	conn.ApplyGroupEvent(&event.Group, time.Now())
+func (app *WebApp) applyGroupUpdatedEvent(
+	conn *webtypes.DeviceConnection,
+	event *models.GroupUpdatedEvent,
+) bool {
+	changed := conn.ApplyGroupEvent(&event.Group, time.Now())
+	if changed {
+		app.QueueDeviceListBroadcast()
+	}
+
+	return changed
 }
 
 // HandleDeviceWebSocket handles individual device WebSocket connections for real-time device-specific updates

pkg/service/soundtouchweb/websocket_test.go

@@ -2,6 +2,10 @@ package soundtouchweb
 
 import (
 	"errors"
+	"github.com/gesellix/bose-soundtouch/pkg/client"
+	"github.com/gesellix/bose-soundtouch/pkg/models"
+	"github.com/gesellix/bose-soundtouch/pkg/service/soundtouchweb/webtypes"
+	"github.com/gorilla/websocket"
 	"net/http"
 	"net/http/httptest"
 	"strings"
@@ -9,11 +13,6 @@ import (
 	"sync/atomic"
 	"testing"
 	"time"
-
-	"github.com/gesellix/bose-soundtouch/pkg/client"
-	"github.com/gesellix/bose-soundtouch/pkg/models"
-	"github.com/gesellix/bose-soundtouch/pkg/service/soundtouchweb/webtypes"
-	"github.com/gorilla/websocket"
 )
 
 func TestUpdateDeviceStatusRefreshesGroup(t *testing.T) {
@@ -142,6 +141,7 @@ func TestUpdateDeviceStatusNotConnectedWhenOnlyGroupSucceeds(t *testing.T) {
 }
 
 func TestApplyGroupUpdatedEventReplacesGroup(t *testing.T) {
+	app := NewWebApp()
 	conn := webtypes.NewDeviceConnection(nil, nil)
 	previousActivity := time.Unix(1, 0)
 	conn.SetStatus(&webtypes.DeviceStatus{
@@ -154,7 +154,9 @@ func TestApplyGroupUpdatedEventReplacesGroup(t *testing.T) {
 	event := &models.GroupUpdatedEvent{
 		Group: models.Group{ID: "pair-new", Name: "Renamed Pair"},
 	}
-	applyGroupUpdatedEvent(conn, event)
+	if !app.applyGroupUpdatedEvent(conn, event) {
+		t.Fatal("new group event should publish a changed projection")
+	}
 
 	status := conn.Status()
 	if status.Group != &event.Group || status.Group.ID != "pair-new" {
@@ -170,7 +172,9 @@ func TestApplyGroupUpdatedEventReplacesGroup(t *testing.T) {
 	}
 
 	teardown := &models.GroupUpdatedEvent{Group: models.Group{}}
-	applyGroupUpdatedEvent(conn, teardown)
+	if !app.applyGroupUpdatedEvent(conn, teardown) {
+		t.Fatal("teardown event should publish a changed projection")
+	}
 
 	if conn.Status().Group != nil {
 		t.Errorf("teardown event did not clear the group: %+v", conn.Status().Group)
@@ -684,3 +688,211 @@ func TestWebSocketWriteBatchRefreshesDeadlineForHealthyWriter(t *testing.T) {
 		t.Fatalf("healthy writer deadline = %v, want a fresh deadline after %v", deadline, started)
 	}
 }
+
+func TestSpeakerEventHelpersPublishOnlyChangedPayloads(t *testing.T) {
+	app := NewWebApp()
+	conn := webtypes.NewDeviceConnection(nil, &models.DeviceInfo{Name: "test"})
+
+	nowPlaying := &models.NowPlaying{Source: "LOCAL_INTERNET_RADIO", Track: "Test station"}
+	if !app.applyNowPlayingEvent(conn, nowPlaying) {
+		t.Fatal("new now-playing payload was not reported as changed")
+	}
+	if app.applyNowPlayingEvent(conn, nowPlaying) {
+		t.Fatal("duplicate now-playing payload was reported as changed")
+	}
+
+	volume := &models.Volume{ActualVolume: 25, TargetVolume: 25}
+	if !app.applyVolumeEvent(conn, volume) {
+		t.Fatal("new volume payload was not reported as changed")
+	}
+	if app.applyVolumeEvent(conn, volume) {
+		t.Fatal("duplicate volume payload was reported as changed")
+	}
+
+	if !app.applyConnectionStateEvent(conn, true) {
+		t.Fatal("new connection state was not reported as changed")
+	}
+	if app.applyConnectionStateEvent(conn, true) {
+		t.Fatal("duplicate connection state was reported as changed")
+	}
+
+	presets := &models.Presets{Preset: []models.Preset{{ID: 1}}}
+	if !app.applyPresetEvent(conn, presets) {
+		t.Fatal("new presets payload was not reported as changed")
+	}
+	if app.applyPresetEvent(conn, presets) {
+		t.Fatal("duplicate presets payload was reported as changed")
+	}
+
+	bass := &models.Bass{ActualBass: -2}
+	if !app.applyBassEvent(conn, bass) {
+		t.Fatal("new bass payload was not reported as changed")
+	}
+	if app.applyBassEvent(conn, bass) {
+		t.Fatal("duplicate bass payload was reported as changed")
+	}
+}
+
+func TestSpeakerEventPublishesImmediateDeviceProjection(t *testing.T) {
+	app := NewWebApp()
+	device := webtypes.NewDeviceConnection(nil, &models.DeviceInfo{Name: "test"})
+	device.SetStatus(&webtypes.DeviceStatus{
+		Volume:      &models.Volume{ActualVolume: 10, TargetVolume: 10},
+		IsConnected: true,
+	})
+	app.AddDevice("speaker", device)
+
+	serverConnection := make(chan *websocket.Conn, 1)
+	releaseServer := make(chan struct{})
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		conn, err := app.Upgrader.Upgrade(w, r, nil)
+		if err != nil {
+			t.Errorf("upgrade test WebSocket: %v", err)
+
+			return
+		}
+		serverConnection <- conn
+		<-releaseServer
+		_ = conn.Close()
+	}))
+
+	client, response, err := websocket.DefaultDialer.Dial(
+		"ws"+strings.TrimPrefix(server.URL, "http"), nil,
+	)
+	if response != nil {
+		_ = response.Body.Close()
+	}
+	if err != nil {
+		close(releaseServer)
+		server.Close()
+		t.Fatalf("dial test WebSocket: %v", err)
+	}
+	remote := <-serverConnection
+	t.Cleanup(func() {
+		app.removeGlobalWebSocketClient(remote)
+		_ = client.Close()
+		close(releaseServer)
+		server.Close()
+	})
+
+	if err := app.registerGlobalWebSocket(remote); err != nil {
+		t.Fatalf("register browser WebSocket: %v", err)
+	}
+	if err := client.SetReadDeadline(time.Now().Add(time.Second)); err != nil {
+		t.Fatalf("set initial read deadline: %v", err)
+	}
+	var initial webtypes.WebSocketMessage
+	if err := client.ReadJSON(&initial); err != nil {
+		t.Fatalf("read initial device projection: %v", err)
+	}
+
+	app.webSocketWriteMu.Lock()
+	writerLocked := true
+	defer func() {
+		if writerLocked {
+			app.webSocketWriteMu.Unlock()
+		}
+	}()
+
+	applied := make(chan bool, 1)
+	go func() {
+		applied <- app.applyVolumeEvent(device, &models.Volume{ActualVolume: 25, TargetVolume: 25})
+	}()
+
+	select {
+	case changed := <-applied:
+		if !changed {
+			t.Fatal("volume event was not applied")
+		}
+	case <-time.After(250 * time.Millisecond):
+		t.Fatal("speaker event blocked on browser WebSocket I/O")
+	}
+
+	app.webSocketWriteMu.Unlock()
+	writerLocked = false
+
+	if err := client.SetReadDeadline(time.Now().Add(time.Second)); err != nil {
+		t.Fatalf("set event read deadline: %v", err)
+	}
+	var update webtypes.WebSocketMessage
+	if err := client.ReadJSON(&update); err != nil {
+		t.Fatalf("read immediate device projection: %v", err)
+	}
+	if update.Type != "devices" {
+		t.Fatalf("event frame type = %q, want devices", update.Type)
+	}
+
+	devices, ok := update.Data.(map[string]interface{})
+	if !ok {
+		t.Fatalf("event projection = %#v, want device map", update.Data)
+	}
+	speaker, ok := devices["speaker"].(map[string]interface{})
+	if !ok {
+		t.Fatalf("speaker projection = %#v, want object", devices["speaker"])
+	}
+	status, ok := speaker["status"].(map[string]interface{})
+	if !ok {
+		t.Fatalf("speaker status = %#v, want object", speaker["status"])
+	}
+	volume, ok := status["volume"].(map[string]interface{})
+	if !ok || volume["ActualVolume"] != float64(25) {
+		t.Fatalf("projected volume = %#v, want 25", status["volume"])
+	}
+}
+
+func TestQueuedDeviceBroadcastCoalescesBurst(t *testing.T) {
+	app := NewWebApp()
+
+	app.webSocketWriteMu.Lock()
+	writerLocked := true
+	defer func() {
+		if writerLocked {
+			app.webSocketWriteMu.Unlock()
+		}
+	}()
+
+	app.QueueDeviceListBroadcast()
+	deadline := time.Now().Add(time.Second)
+	for {
+		app.deviceBroadcastMu.Lock()
+		running := app.deviceBroadcastRunning
+		pending := app.deviceBroadcastPending
+		app.deviceBroadcastMu.Unlock()
+		if running && !pending {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatal("broadcast worker did not begin its blocked write")
+		}
+		time.Sleep(time.Millisecond)
+	}
+
+	for range 100 {
+		app.QueueDeviceListBroadcast()
+	}
+
+	app.deviceBroadcastMu.Lock()
+	if !app.deviceBroadcastPending {
+		app.deviceBroadcastMu.Unlock()
+		t.Fatal("burst did not retain one coalesced follow-up")
+	}
+	app.deviceBroadcastMu.Unlock()
+
+	app.webSocketWriteMu.Unlock()
+	writerLocked = false
+
+	deadline = time.Now().Add(time.Second)
+	for {
+		app.deviceBroadcastMu.Lock()
+		running := app.deviceBroadcastRunning
+		pending := app.deviceBroadcastPending
+		app.deviceBroadcastMu.Unlock()
+		if !running && !pending {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("coalesced worker did not drain: running=%v pending=%v", running, pending)
+		}
+		time.Sleep(time.Millisecond)
+	}
+}

pkg/service/soundtouchweb/webtypes/status_test.go

@@ -248,6 +248,64 @@ func TestApplyGroupEventIgnoresRoleOrder(t *testing.T) {
 	}
 }
 
+func TestStatusPollCannotOverwriteNewerSpeakerEvent(t *testing.T) {
+	conn := NewDeviceConnection(nil, &models.DeviceInfo{Name: "test"})
+	poll := conn.BeginStatusPoll()
+
+	conn.ApplySpeakerEvent(func(status *DeviceStatus) {
+		status.Volume = &models.Volume{ActualVolume: 99}
+	})
+
+	if conn.CompleteStatusPoll(poll, func(status *DeviceStatus) {
+		status.Volume = &models.Volume{ActualVolume: 42}
+	}) {
+		t.Fatal("poll that began before a speaker 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)
+	}
+}
+
+func TestNewerStatusPollSupersedesOlderPoll(t *testing.T) {
+	conn := NewDeviceConnection(nil, &models.DeviceInfo{Name: "test"})
+	older := conn.BeginStatusPoll()
+	newer := conn.BeginStatusPoll()
+
+	if !conn.CompleteStatusPoll(newer, func(status *DeviceStatus) {
+		status.Volume = &models.Volume{ActualVolume: 30}
+	}) {
+		t.Fatal("newer poll was not applied")
+	}
+
+	if conn.CompleteStatusPoll(older, func(status *DeviceStatus) {
+		status.Volume = &models.Volume{ActualVolume: 10}
+	}) {
+		t.Fatal("older poll completed after a newer poll was applied")
+	}
+
+	if got := conn.Status().Volume; got == nil || got.ActualVolume != 30 {
+		t.Fatalf("newer poll state was overwritten: %+v", got)
+	}
+}
+
+func TestDuplicateSpeakerEventStillInvalidatesOlderPoll(t *testing.T) {
+	conn := NewDeviceConnection(nil, &models.DeviceInfo{Name: "test"})
+	conn.SetStatus(&DeviceStatus{Volume: &models.Volume{ActualVolume: 25}})
+	poll := conn.BeginStatusPoll()
+
+	conn.ApplySpeakerEvent(func(status *DeviceStatus) {
+		status.Volume = &models.Volume{ActualVolume: 25}
+		status.LastActivity = time.Now()
+	})
+
+	if conn.CompleteStatusPoll(poll, func(status *DeviceStatus) {
+		status.Volume = &models.Volume{ActualVolume: 10}
+	}) {
+		t.Fatal("poll that preceded duplicate speaker evidence was applied")
+	}
+}
+
 func TestStatusSnapshotIsolation(t *testing.T) {
 	// A snapshot returned by Status() must NOT change when a later
 	// UpdateStatus replaces a pointer field. This proves the atomic

pkg/service/soundtouchweb/webtypes/types.go

@@ -47,6 +47,12 @@ type DeviceConnection struct {
 
 	status atomic.Pointer[DeviceStatus]
 
+	statusOrderMu             sync.Mutex
+	nextStatusPollGeneration  uint64
+	lastStatusPollGeneration  uint64
+	speakerEventGeneration    uint64
+	statusPollEventGeneration map[uint64]uint64
+
 	// groupMu orders polled /getGroup responses against real-time
 	// groupUpdated events. groupGeneration is the highest generation
 	// issued (by BeginGroupRefresh or ApplyGroupEvent);
@@ -135,6 +141,66 @@ 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()
+
+	c.nextStatusPollGeneration++
+	if c.statusPollEventGeneration == nil {
+		c.statusPollEventGeneration = make(map[uint64]uint64)
+	}
+
+	c.statusPollEventGeneration[c.nextStatusPollGeneration] = c.speakerEventGeneration
+
+	return c.nextStatusPollGeneration
+}
+
+// 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()
+
+	c.speakerEventGeneration++
+	c.UpdateStatus(mut)
+}
+
+// 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.UpdateStatus(mut)
+
+	return true
+}
+
 // 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