fix(mqtt): ensure proper locking when subscribing and unsubscribing topics (#3864)

Signed-off-by: Anton Troshin <[email protected]>
Co-authored-by: Yaron Schneider <[email protected]>

Author: antontroshin

pubsub/mqtt3/mqtt.go

@@ -46,6 +46,7 @@ type mqttPubSub struct {
 	logger          logger.Logger
 	topics          map[string]mqttPubSubSubscription
 	subscribingLock sync.RWMutex
+	deletionLock    sync.RWMutex
 	reconnectCh     chan struct{}
 	closeCh         chan struct{}
 	closed          atomic.Bool
@@ -143,11 +144,11 @@ func (m *mqttPubSub) Subscribe(ctx context.Context, req pubsub.SubscribeRequest,
 
 	m.subscribingLock.Lock()
 	defer m.subscribingLock.Unlock()
-
 	// Add the topic then start the subscription
 	m.addTopic(topic, handler)
 
 	token := m.conn.Subscribe(topic, m.metadata.Qos, m.onMessage(ctx))
+
 	var err error
 	select {
 	case <-token.Done():
@@ -180,7 +181,9 @@ func (m *mqttPubSub) Subscribe(ctx context.Context, req pubsub.SubscribeRequest,
 		}
 
 		// Delete the topic from the map first, which stops routing messages to handlers
+		m.deletionLock.Lock()
 		delete(m.topics, topic)
+		m.deletionLock.Unlock()
 
 		// We will call Unsubscribe only if cleanSession is true or if "unsubscribeOnClose" in the request metadata is true
 		// Otherwise, calling this will make the broker lose the position of our subscription, which is not what we want if we are going to reconnect later