diff --git a/core/internal/server/cups/manager.go b/core/internal/server/cups/manager.go index e5b7e8c37..2e0081828 100644 --- a/core/internal/server/cups/manager.go +++ b/core/internal/server/cups/manager.go @@ -229,6 +229,9 @@ func (m *Manager) snapshotState() CUPSState { func (m *Manager) Subscribe(id string) chan CUPSState { ch := make(chan CUPSState, 64) + m.subLifecycleMu.Lock() + defer m.subLifecycleMu.Unlock() + wasEmpty := true m.subscribers.Range(func(key string, ch chan CUPSState) bool { wasEmpty = false @@ -237,19 +240,25 @@ func (m *Manager) Subscribe(id string) chan CUPSState { m.subscribers.Store(id, ch) - if wasEmpty && m.subscription != nil { - if err := m.subscription.Start(); err != nil { - log.Warnf("[CUPS] Failed to start subscription manager: %v", err) - } else { - m.eventWG.Add(1) - go m.eventHandler() - } + if !wasEmpty || m.subscription == nil { + return ch } + if err := m.subscription.Start(); err != nil { + log.Warnf("[CUPS] Failed to start subscription manager: %v", err) + return ch + } + + m.eventWG.Add(1) + go m.eventHandler() + return ch } func (m *Manager) Unsubscribe(id string) { + m.subLifecycleMu.Lock() + defer m.subLifecycleMu.Unlock() + if val, ok := m.subscribers.LoadAndDelete(id); ok { close(val) } @@ -260,18 +269,22 @@ func (m *Manager) Unsubscribe(id string) { return false }) - if isEmpty && m.subscription != nil { - m.subscription.Stop() - m.eventWG.Wait() + if !isEmpty || m.subscription == nil { + return } + + m.subscription.Stop() + m.eventWG.Wait() } func (m *Manager) Close() { close(m.stopChan) + m.subLifecycleMu.Lock() if m.subscription != nil { m.subscription.Stop() } + m.subLifecycleMu.Unlock() m.eventWG.Wait() m.notifierWg.Wait() diff --git a/core/internal/server/cups/manager_test.go b/core/internal/server/cups/manager_test.go index 7f09e6798..59e346a34 100644 --- a/core/internal/server/cups/manager_test.go +++ b/core/internal/server/cups/manager_test.go @@ -1,6 +1,9 @@ package cups import ( + "errors" + "fmt" + "sync" "testing" mocks_cups "github.com/AvengeMedia/DankMaterialShell/core/internal/mocks/cups" @@ -75,6 +78,67 @@ func TestManager_Subscribe(t *testing.T) { assert.Equal(t, 0, count) } +// mirrors the real managers: eventChan guarded by mu, conn/running deliberately +// unsynchronized so overlapping Start/Stop trips the race detector +type stubSubscription struct { + mu sync.Mutex + events chan SubscriptionEvent + conn *int + running bool +} + +func (s *stubSubscription) Start() error { + if s.running { + return errors.New("already running") + } + s.running = true + + s.mu.Lock() + s.events = make(chan SubscriptionEvent) + s.mu.Unlock() + + v := 0 + s.conn = &v + *s.conn++ + + return nil +} + +func (s *stubSubscription) Stop() { + if !s.running { + return + } + s.running = false + s.conn = nil + + s.mu.Lock() + close(s.events) + s.mu.Unlock() +} + +func (s *stubSubscription) Events() <-chan SubscriptionEvent { + s.mu.Lock() + defer s.mu.Unlock() + return s.events +} + +func TestManager_SubscribeUnsubscribeRace(t *testing.T) { + m := NewTestManager(mocks_cups.NewMockCUPSClientInterface(t), nil) + m.subscription = &stubSubscription{} + + var wg sync.WaitGroup + for i := range 8 { + wg.Go(func() { + id := fmt.Sprintf("client-%d", i) + for range 50 { + m.Subscribe(id) + m.Unsubscribe(id) + } + }) + } + wg.Wait() +} + func TestManager_Close(t *testing.T) { mockClient := mocks_cups.NewMockCUPSClientInterface(t) diff --git a/core/internal/server/cups/types.go b/core/internal/server/cups/types.go index 52f0f5b9e..a7f05b68d 100644 --- a/core/internal/server/cups/types.go +++ b/core/internal/server/cups/types.go @@ -79,6 +79,7 @@ type Manager struct { client CUPSClientInterface pkHelper PkHelper subscription SubscriptionManagerInterface + subLifecycleMu sync.Mutex stateMutex sync.RWMutex subscribers syncmap.Map[string, chan CUPSState] stopChan chan struct{}