fix(protocol): guard connection state access

pkg/messaging/core.go

@@ -46,7 +46,8 @@ type Core struct {
 	wg   sync.WaitGroup
 	quit chan struct{}
 
-	connectionState connection.State
+	connectionState   connection.State
+	connectionStateMu sync.Mutex
 
 	wakumetrics *wakumetrics2.Client
 }
@@ -374,16 +375,19 @@ func (c *Core) decryptMessage(myIdentityKey *ecdsa.PrivateKey, theirPublicKey *e
 func (c *Core) connectionChanged(state connection.State) {
 	c.stack.Transport.ConnectionChanged(state)
 
-	if !c.connectionState.Offline && state.Offline {
+	c.connectionStateMu.Lock()
+	wasOffline := c.connectionState.Offline
+	c.connectionState = state
+	c.connectionStateMu.Unlock()
+
+	if !wasOffline && state.Offline {
 		c.controller.StopReliability()
 	}
 
-	if c.connectionState.Offline && !state.Offline {
+	if wasOffline && !state.Offline {
 		err := c.controller.StartReliability()
 		if err != nil {
 			c.logger.Error("failed to start datasync", zap.Error(err))
 		}
 	}
-
-	c.connectionState = state
 }

protocol/messenger.go

@@ -151,6 +151,7 @@ type Messenger struct {
 	}
 
 	connectionState       connection.State
+	connectionStateMu     sync.RWMutex
 	contractMaker         *contracts.ContractMaker
 	verificationDatabase  *verification.Persistence
 	savedAddressesManager *wallet.SavedAddressesManager

protocol/messenger_connection_state_test.go

@@ -0,0 +1,29 @@
+package protocol
+
+import (
+	"sync"
+	"testing"
+
+	"github.com/status-im/status-go/internal/connection"
+)
+
+func TestMessengerConnectionStateAccessorsAreConcurrentSafe(t *testing.T) {
+	m := &Messenger{}
+
+	var wg sync.WaitGroup
+	for i := 0; i < 100; i++ {
+		wg.Add(2)
+
+		go func(i int) {
+			defer wg.Done()
+			m.setConnectionState(connection.State{Expensive: i%2 == 0})
+		}(i)
+
+		go func() {
+			defer wg.Done()
+			_ = m.isConnectionExpensive()
+		}()
+	}
+
+	wg.Wait()
+}

protocol/messenger_mailserver.go

@@ -593,17 +593,31 @@ func (m *Messenger) calculateGapForChat(chat *Chat, from uint32) (*common.Messag
 }
 
 func (m *Messenger) canSyncWithStoreNodes() (bool, error) {
-	if m.connectionState.IsExpensive() {
+	if m.isConnectionExpensive() {
 		return m.settings.CanSyncOnMobileNetwork()
 	}
 	return true, nil
 }
 
 func (m *Messenger) ConnectionChanged(state connection.State) {
 	m.messaging.ConnectionChanged(state)
+	m.setConnectionState(state)
+}
+
+func (m *Messenger) setConnectionState(state connection.State) {
+	m.connectionStateMu.Lock()
+	defer m.connectionStateMu.Unlock()
+
 	m.connectionState = state
 }
 
+func (m *Messenger) isConnectionExpensive() bool {
+	m.connectionStateMu.RLock()
+	defer m.connectionStateMu.RUnlock()
+
+	return m.connectionState.IsExpensive()
+}
+
 func (m *Messenger) processMailserverBatch(peerInfo peer.AddrInfo, batch types2.StoreNodeBatch) error {
 	canSync, err := m.canSyncWithStoreNodes()
 	if err != nil {