From 90bb0fd00ecf111bcc134e8fc990488df3082b0c Mon Sep 17 00:00:00 2001 From: Gary Date: Fri, 25 Sep 2026 13:40:57 +0200 Subject: [PATCH] fix(service): stop an unsubscribe from deadlocking the bus Unsubscribe took the lock once per subscriber it looked at, each unlock deferred to its return: a websocket closing while it was not the first subscriber of an event locked the bus's mutex twice and never came back. Every subscription after it hung once its websocket was open, so the subscriber believed it was listening and received nothing. The core's alerts, which resubscribe each time the core restarts, hit it on every install check: three connections closed together, the third deadlocked. The lock is now taken once per call, in both services. The dispatch reads the subscriber lists under it too, as a map read while Subscribe writes it can bring the bus down. And two silent drops go: a publish waits for the dispatch instead of dropping an event it is busy with, and a subscriber may be 64 events behind instead of one. --- service/action_service_websocket.go | 12 ++- service/event_service_websocket.go | 84 +++++++++---------- service/event_service_websocket_test.go | 104 ++++++++++++++++++++++++ 3 files changed, 151 insertions(+), 49 deletions(-) create mode 100644 service/event_service_websocket_test.go diff --git a/service/action_service_websocket.go b/service/action_service_websocket.go index 6c8f5cc..f934329 100644 --- a/service/action_service_websocket.go +++ b/service/action_service_websocket.go @@ -90,6 +90,11 @@ func (s *ActionServiceWS) Subscribe(sourceID string, names []string) (chan model } func (s *ActionServiceWS) Unsubscribe(sourceID string, name string, c chan model.Action) error { + // once, for the whole call: taken per subscriber looked at, its unlock + // deferred, it deadlocked on the second (see EventServiceWS.Unsubscribe) + s.mutex.Lock() + defer s.mutex.Unlock() + if s.subscriberChannels == nil { return ErrSubscriberChannelsNotFound } @@ -103,14 +108,7 @@ func (s *ActionServiceWS) Unsubscribe(sourceID string, name string, c chan model } for i, subscriber := range s.subscriberChannels[sourceID][name] { - s.mutex.Lock() - defer s.mutex.Unlock() - if subscriber == c { - if i >= len(s.subscriberChannels[sourceID][name]) { - logger.Error("the i-th subscriber is removed before we get here - concurrency issue?", zap.Int("subscriber", i), zap.Int("total", len(s.subscriberChannels[sourceID][name]))) - return ErrAlreadySubscribed - } s.subscriberChannels[sourceID][name] = append(s.subscriberChannels[sourceID][name][:i], s.subscriberChannels[sourceID][name][i+1:]...) return nil } diff --git a/service/event_service_websocket.go b/service/event_service_websocket.go index 16f3db3..0c0d913 100644 --- a/service/event_service_websocket.go +++ b/service/event_service_websocket.go @@ -12,6 +12,11 @@ import ( "go.uber.org/zap" ) +// subscriberBuffer is how many events a subscriber may be behind before it +// misses one: a backup's error right after its begin, or two containers dying +// together, is more than one. +const subscriberBuffer = 64 + type EventServiceWS struct { typeService *EventTypeService @@ -35,6 +40,9 @@ func (s *EventServiceWS) Publish(event model.Event) { // TODO - ensure properties are valid for event type + // Waits for the dispatch to take it, never long: the route publishes from a + // goroutine of its own, and an event dropped here because the dispatch was + // busy with the one before was an alert never sent. select { case s.inboundChannel <- event: @@ -43,8 +51,6 @@ func (s *EventServiceWS) Publish(event model.Event) { logger.Info(err.Error()) } return - - default: // drop event if no one is listening } } @@ -83,7 +89,7 @@ func (s *EventServiceWS) Subscribe(sourceID string, names []string) (chan model. s.subscriberChannels[sourceID] = make(map[string][]chan model.Event) } - c := make(chan model.Event, 1) + c := make(chan model.Event, subscriberBuffer) for _, name := range names { if s.subscriberChannels[sourceID][name] == nil { @@ -98,6 +104,12 @@ func (s *EventServiceWS) Subscribe(sourceID string, names []string) (chan model. } func (s *EventServiceWS) Unsubscribe(sourceID string, name string, c chan model.Event) error { + // Once, for the whole call. Taken per subscriber looked at, its unlock + // deferred, it deadlocked the bus on the second: every subscription after + // that hung once its websocket was open, and its events never came. + mutex.Lock() + defer mutex.Unlock() + if s.subscriberChannels == nil { return ErrSubscriberChannelsNotFound } @@ -111,15 +123,8 @@ func (s *EventServiceWS) Unsubscribe(sourceID string, name string, c chan model. } for i, subscriber := range s.subscriberChannels[sourceID][name] { - mutex.Lock() - defer mutex.Unlock() - if subscriber == c { logger.Info("unsubscribing from event type", zap.String("sourceID", sourceID), zap.String("name", name), zap.Int("subscriber", i)) - if i >= len(s.subscriberChannels[sourceID][name]) { - logger.Error("the i-th subscriber is removed before we get here - concurrency issue?", zap.Int("subscriber", i), zap.Int("total", len(s.subscriberChannels[sourceID][name]))) - return ErrAlreadySubscribed - } s.subscriberChannels[sourceID][name] = append(s.subscriberChannels[sourceID][name][:i], s.subscriberChannels[sourceID][name][i+1:]...) return nil } @@ -182,52 +187,47 @@ func (s *EventServiceWS) Start(ctx *context.Context) { return } - if s.subscriberChannels == nil { - continue - } - - if s.subscriberChannels[event.SourceID] == nil { - continue - } - - if s.subscriberChannels[event.SourceID][event.Name] == nil { - continue - } + // Under the lock Subscribe and Unsubscribe change the lists with: read + // while they change, a map can bring the bus down, and a subscriber + // just gone has its channel closed. + func() { + mutex.Lock() + defer mutex.Unlock() - for _, c := range s.subscriberChannels[event.SourceID][event.Name] { - select { - case c <- event: - case <-(*s.ctx).Done(): + if s.subscriberChannels == nil || s.subscriberChannels[event.SourceID] == nil { return - default: // drop event if no one is listening - continue } - } - case <-ticker.C: - if s.subscriberChannels == nil { - continue - } + for _, c := range s.subscriberChannels[event.SourceID][event.Name] { + select { + case c <- event: + default: // a subscriber this far behind misses it: the bus waits for none + } + } + }() + case <-ticker.C: heartbeat := model.Event{ SourceID: common.MessageBusSourceID, Name: common.MessageBusHeartbeatName, Timestamp: time.Now().Unix(), } - for _, source := range s.subscriberChannels { - for _, subscribers := range source { - for _, subscriber := range subscribers { - select { - case subscriber <- heartbeat: - case <-(*s.ctx).Done(): - return - default: // drop event if no one is listening - continue + func() { + mutex.Lock() + defer mutex.Unlock() + + for _, source := range s.subscriberChannels { + for _, subscribers := range source { + for _, subscriber := range subscribers { + select { + case subscriber <- heartbeat: + default: // one already waiting does as well + } } } } - } + }() } } } diff --git a/service/event_service_websocket_test.go b/service/event_service_websocket_test.go new file mode 100644 index 0000000..7c29a39 --- /dev/null +++ b/service/event_service_websocket_test.go @@ -0,0 +1,104 @@ +package service + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/ReCasaOS/CasaOS-Common/utils/logger" + "github.com/ReCasaOS/CasaOS-MessageBus/model" + "github.com/ReCasaOS/CasaOS-MessageBus/repository" + "gotest.tools/assert" +) + +// within fails the test when f has not returned after five seconds. +func within(t *testing.T, what string, f func()) { + t.Helper() + done := make(chan struct{}) + go func() { + f() + close(done) + }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatalf("%s did not return: the bus is deadlocked", what) + } +} + +func newEventService(t *testing.T) *EventServiceWS { + t.Helper() + logger.LogInitConsoleOnly() + repository, err := repository.NewDatabaseRepositoryInMemory() + assert.NilError(t, err) + t.Cleanup(func() { repository.Close() }) + typeService := NewEventTypeService(&repository) + _, err = typeService.RegisterEventType(model.EventType{SourceID: "app-management", Name: "backup:error"}) + assert.NilError(t, err) + return NewEventServiceWS(typeService) +} + +// Three websockets on one event, the middle one closing: Unsubscribe took the +// lock once per subscriber it looked at, and deadlocked on the second. Every +// subscription after that hung, the core's alerts among them. +func TestUnsubscribeASubscriberAfterTheFirst(t *testing.T) { + s := newEventService(t) + channels := make([]chan model.Event, 3) + for i := range channels { + c, err := s.Subscribe("app-management", []string{"backup:error"}) + assert.NilError(t, err) + channels[i] = c + } + + within(t, "Unsubscribe", func() { + assert.NilError(t, s.Unsubscribe("app-management", "backup:error", channels[1])) + }) + within(t, "Subscribe", func() { + _, err := s.Subscribe("app-management", []string{"backup:error"}) + assert.NilError(t, err) + }) + assert.Equal(t, len(s.subscriberChannels["app-management"]["backup:error"]), 3) +} + +// Events published together all reach a subscriber: none is dropped because +// the dispatch was busy with another, or the subscriber had one waiting. +func TestEventsPublishedTogetherAllArrive(t *testing.T) { + s := newEventService(t) + ctx, cancel := context.WithCancel(context.Background()) + stopped := make(chan struct{}) + go func() { + s.Start(&ctx) + close(stopped) + }() + t.Cleanup(func() { + cancel() + <-stopped + }) + for started := false; !started; { + mutex.Lock() + started = s.stop != nil + mutex.Unlock() + time.Sleep(time.Millisecond) + } + + c, err := s.Subscribe("app-management", []string{"backup:error"}) + assert.NilError(t, err) + var publishing sync.WaitGroup + for range 10 { + publishing.Add(1) + go func() { + defer publishing.Done() + s.Publish(model.Event{SourceID: "app-management", Name: "backup:error"}) + }() + } + publishing.Wait() + + for i := range 10 { + select { + case <-c: + case <-time.After(5 * time.Second): + t.Fatalf("%d of 10 events arrived", i) + } + } +}