evmos · GAtom22 · Jul 5, 2023 · Jul 5, 2023 · Jul 5, 2023 · Jul 5, 2023
@@ -49,6 +49,10 @@ Ref: https://keepachangelog.com/en/1.0.0/
 
 - (evm) [#1625](https://github.com/evmos/evmos/pull/1625) Migrate updated EVM extensions
 
+### Bug Fixes
+
+- (rpc) [#1646](https://github.com/evmos/evmos/pull/1646) Fix racing conditions on RPC PubSub logic
+
 ## [v13.0.0-rc3] - 2023-06-15
 
 ### Bug Fixes

@@ -304,14 +304,14 @@ TEST_TARGETS := test-unit test-unit-cover test-race
 # Test runs-specific rules. To add a new test target, just add
 # a new rule, customise ARGS or TEST_PACKAGES ad libitum, and
 # append the new rule to the TEST_TARGETS list.
-test-unit: ARGS=-timeout=15m -race
+test-unit: ARGS=-timeout=15m
 test-unit: TEST_PACKAGES=$(PACKAGES_UNIT)
 
 test-race: ARGS=-race
 test-race: TEST_PACKAGES=$(PACKAGES_NOSIMULATION)
 $(TEST_TARGETS): run-tests
 
-test-unit-cover: ARGS=-timeout=15m -race -coverprofile=coverage.txt -covermode=atomic
+test-unit-cover: ARGS=-timeout=15m -coverprofile=coverage.txt -covermode=atomic
 test-unit-cover: TEST_PACKAGES=$(PACKAGES_UNIT)
 
 test-e2e:

@@ -20,20 +20,26 @@ type EventBus interface {
 	Topics() []string
 }
 
+type subscriptionsContainer struct {
+	mu            sync.Mutex
+	subscriptions map[string]map[uint64]chan<- coretypes.ResultEvent
+}
+
+type topicsContainer struct {
+	mu     sync.RWMutex
+	topics map[string]<-chan coretypes.ResultEvent
+}
+
 type memEventBus struct {
-	topics          map[string]<-chan coretypes.ResultEvent
-	topicsMux       *sync.RWMutex
-	subscribers     map[string]map[uint64]chan<- coretypes.ResultEvent
-	subscribersMux  *sync.RWMutex
+	t               topicsContainer
+	s               subscriptionsContainer
 	currentUniqueID uint64
 }
 
 func NewEventBus() EventBus {
 	return &memEventBus{
-		topics:         make(map[string]<-chan coretypes.ResultEvent),
-		topicsMux:      new(sync.RWMutex),
-		subscribers:    make(map[string]map[uint64]chan<- coretypes.ResultEvent),
-		subscribersMux: new(sync.RWMutex),
+		t: topicsContainer{topics: make(map[string]<-chan coretypes.ResultEvent)},
+		s: subscriptionsContainer{subscriptions: make(map[string]map[uint64]chan<- coretypes.ResultEvent)},
 	}
 }
 
@@ -42,64 +48,64 @@ func (m *memEventBus) GenUniqueID() uint64 {
 }
 
 func (m *memEventBus) Topics() (topics []string) {
-	m.topicsMux.RLock()
-	defer m.topicsMux.RUnlock()
+	m.t.mu.RLock()
+	defer m.t.mu.RUnlock()
 
-	topics = make([]string, 0, len(m.topics))
-	for topicName := range m.topics {
+	topics = make([]string, 0, len(m.t.topics))
+	for topicName := range m.t.topics {
 		topics = append(topics, topicName)
 	}
 
 	return topics
 }
 
 func (m *memEventBus) AddTopic(name string, src <-chan coretypes.ResultEvent) error {
-	m.topicsMux.RLock()
-	_, ok := m.topics[name]
-	m.topicsMux.RUnlock()
+	m.t.mu.RLock()
+	_, ok := m.t.topics[name]
+	m.t.mu.RUnlock()
 
 	if ok {
 		return errors.New("topic already registered")
 	}
 
-	m.topicsMux.Lock()
-	m.topics[name] = src
-	m.topicsMux.Unlock()
+	m.t.mu.Lock()
+	m.t.topics[name] = src
+	m.t.mu.Unlock()
 
 	go m.publishTopic(name, src)
 
 	return nil
 }
 
 func (m *memEventBus) RemoveTopic(name string) {
-	m.topicsMux.Lock()
-	delete(m.topics, name)
-	m.topicsMux.Unlock()
+	m.t.mu.Lock()
+	defer m.t.mu.Unlock()
+	delete(m.t.topics, name)
 }
 
 func (m *memEventBus) Subscribe(name string) (<-chan coretypes.ResultEvent, UnsubscribeFunc, error) {
-	m.topicsMux.RLock()
-	_, ok := m.topics[name]
-	m.topicsMux.RUnlock()
+	m.t.mu.RLock()
+	_, ok := m.t.topics[name]
+	m.t.mu.RUnlock()
 
 	if !ok {
 		return nil, nil, errors.Errorf("topic not found: %s", name)
 	}
 
 	ch := make(chan coretypes.ResultEvent)
-	m.subscribersMux.Lock()
-	defer m.subscribersMux.Unlock()
+	m.s.mu.Lock()
+	defer m.s.mu.Unlock()
 
 	id := m.GenUniqueID()
-	if _, ok := m.subscribers[name]; !ok {
-		m.subscribers[name] = make(map[uint64]chan<- coretypes.ResultEvent)
+	if _, ok := m.s.subscriptions[name]; !ok {
+		m.s.subscriptions[name] = make(map[uint64]chan<- coretypes.ResultEvent)
 	}
-	m.subscribers[name][id] = ch
+	m.s.subscriptions[name][id] = ch
 
 	unsubscribe := func() {
-		m.subscribersMux.Lock()
-		defer m.subscribersMux.Unlock()
-		delete(m.subscribers[name], id)
+		m.s.mu.Lock()
+		defer m.s.mu.Unlock()
+		delete(m.s.subscriptions[name], id)
 	}
 
 	return ch, unsubscribe, nil
@@ -110,31 +116,31 @@ func (m *memEventBus) publishTopic(name string, src <-chan coretypes.ResultEvent
 		msg, ok := <-src
 		if !ok {
 			m.closeAllSubscribers(name)
-			m.topicsMux.Lock()
-			delete(m.topics, name)
-			m.topicsMux.Unlock()
+			m.t.mu.Lock()
+			delete(m.t.topics, name)
+			m.t.mu.Unlock()
 			return
 		}
 		m.publishAllSubscribers(name, msg)
 	}
 }
 
 func (m *memEventBus) closeAllSubscribers(name string) {
-	m.subscribersMux.Lock()
-	defer m.subscribersMux.Unlock()
+	m.s.mu.Lock()
+	defer m.s.mu.Unlock()
 
-	subsribers := m.subscribers[name]
-	delete(m.subscribers, name)
+	subsribers := m.s.subscriptions[name]
+	delete(m.s.subscriptions, name)
 	// #nosec G705
 	for _, sub := range subsribers {
 		close(sub)
 	}
 }
 
 func (m *memEventBus) publishAllSubscribers(name string, msg coretypes.ResultEvent) {
-	m.subscribersMux.RLock()
-	subsribers := m.subscribers[name]
-	m.subscribersMux.RUnlock()
+	m.s.mu.Lock()
+	defer m.s.mu.Unlock()
+	subsribers := m.s.subscriptions[name]
 	// #nosec G705
 	for _, sub := range subsribers {
 		select {

@@ -85,3 +85,48 @@ func TestSubscribe(t *testing.T) {
 	wg.Wait()
 	time.Sleep(time.Second)
 }
+
+func TestConcurrentSubscribeAndPublish(t *testing.T) {
+	var wg sync.WaitGroup
+
+	q := NewEventBus()
+
+	lolSrc := make(chan coretypes.ResultEvent)
+	topicName := "lol"
+
+	err := q.AddTopic(topicName, lolSrc)
+	require.NoError(t, err)
+
+	subscribersCount := 50
+
+	emptyMsg := coretypes.ResultEvent{}
+
+	for i := 0; i < subscribersCount; i++ {
+		wg.Add(1)
+		// concurrently subscribe to the topic
+		go func() {
+			defer wg.Done()
+			_, _, err := q.Subscribe(topicName)
+			require.NoError(t, err)
+		}()
+
+		// send events to the topic
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			lolSrc <- emptyMsg
+		}()
+	}
+
+	// close channel to make test end
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+
+		time.Sleep(2 * time.Second)
+
+		close(lolSrc)
+	}()
+
+	wg.Wait()
+}