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) + } + } +}