Fix data race b/c of lacking lock protection for "data" slice

ci/test/test.go

@@ -30,7 +30,10 @@ func TestWithCoverage() error {
 		return err
 	}
 	args := append([]string{"test", "-race", "-timeout", "5m", "-cover", "-coverprofile", "coverage.txt", "-covermode", "atomic"}, packages...)
-	return sh.RunV("go", args...)
+
+	env := make(map[string]string)
+	env["GORACE"] = "halt_on_error=1"
+	return sh.RunWith(env, "go", args...)
 }
 
 // Test runs unit tests
@@ -39,8 +42,11 @@ func Test() error {
 	if err != nil {
 		return err
 	}
-	args := append([]string{"test", "-race", "-count=1", "-timeout", "5m"}, packages...)
-	return sh.RunV("go", args...)
+	args := append([]string{"test", "-race", "-timeout", "5m"}, packages...)
+
+	env := make(map[string]string)
+	env["GORACE"] = "halt_on_error=1"
+	return sh.RunWith(env, "go", args...)
 }
 
 func unitTestPackages() ([]string, error) {

session/pingpong/consumer_balance_tracker_test.go

@@ -364,34 +364,37 @@ func TestConsumerBalanceTracker_InprogressUnregisteredBalanceReturnedWhenNoBount
 }
 
 func TestConsumerBalanceTracker_RecoverGrandTotalPromisedSettledIsBiggerThanPromissedNotOffChain(t *testing.T) {
-	id1 := identity.FromAddress("0x000000001")
-	grandTotalPromised := big.NewInt(10)
-	settledAmount := big.NewInt(11)
-	bus := eventbus.New()
-	mcts := NewConsumerTotalsStorage(bus)
-	bc := mockConsumerBalanceChecker{}
-	cfg := defaultCfg
-	cfg.LongSync.Interval = time.Millisecond * 300
-	calc := newMockAddressProvider()
-	calc.addrToReturn = id1.ToCommonAddress()
-	mockBlockchainProvider := &mockBlockchainInfoProvider{}
-
-	mockBlockchainProvider.AddConsumerChannelsHermes(1, id1.ToCommonAddress(), client.ConsumersHermes{
-		Settled: big.NewInt(6),
-	})
+	// make data race more likely to happen
+	for i := 0; i < 10; i++ {
+		id1 := identity.FromAddress("0x000000001")
+		grandTotalPromised := big.NewInt(10)
+		settledAmount := big.NewInt(11)
+		bus := eventbus.New()
+		mcts := NewConsumerTotalsStorage(bus)
+		bc := mockConsumerBalanceChecker{}
+		cfg := defaultCfg
+		cfg.LongSync.Interval = time.Millisecond * 300
+		calc := newMockAddressProvider()
+		calc.addrToReturn = id1.ToCommonAddress()
+		mockBlockchainProvider := &mockBlockchainInfoProvider{}
+
+		mockBlockchainProvider.AddConsumerChannelsHermes(1, id1.ToCommonAddress(), client.ConsumersHermes{
+			Settled: big.NewInt(6),
+		})
 
-	cbt := NewConsumerBalanceTracker(bus, &bc, mcts, &mockconsumerInfoGetter{grandTotalPromised, settledAmount}, &mockTransactor{}, &mockRegistrationStatusProvider{}, calc, mockBlockchainProvider, defaultCfg)
+		cbt := NewConsumerBalanceTracker(bus, &bc, mcts, &mockconsumerInfoGetter{grandTotalPromised, settledAmount}, &mockTransactor{}, &mockRegistrationStatusProvider{}, calc, mockBlockchainProvider, defaultCfg)
 
-	err := cbt.Subscribe(bus)
-	assert.NoError(t, err)
-	bus.Publish(identity.AppTopicIdentityUnlock, identity.AppEventIdentityUnlock{
-		ChainID: 1,
-		ID:      id1,
-	})
-	assert.Eventually(t, func() bool {
-		savedBalance, _ := mcts.Get(1, id1, common.BigToAddress(big.NewInt(0)))
-		return savedBalance.Cmp(big.NewInt(6)) == 0
-	}, defaultWaitTime, defaultWaitInterval)
+		err := cbt.Subscribe(bus)
+		assert.NoError(t, err)
+		bus.Publish(identity.AppTopicIdentityUnlock, identity.AppEventIdentityUnlock{
+			ChainID: 1,
+			ID:      id1,
+		})
+		assert.Eventually(t, func() bool {
+			savedBalance, _ := mcts.Get(1, id1, common.BigToAddress(big.NewInt(0)))
+			return savedBalance.Cmp(big.NewInt(6)) == 0
+		}, defaultWaitTime, defaultWaitInterval)
+	}
 }
 
 type mockConsumerBalanceChecker struct {

session/pingpong/consumer_totals_storage.go

@@ -94,6 +94,9 @@ func (cts *ConsumerTotalsStorage) Store(chainID int64, id identity.Identity, her
 // Get fetches the amount as promised for the given channel.
 func (cts *ConsumerTotalsStorage) Get(chainID int64, id identity.Identity, hermesID common.Address) (*big.Int, error) {
 	key := cts.makeKey(chainID, id, hermesID)
+	cts.createLock.Lock()
+	defer cts.createLock.Unlock()
+
 	element, ok := cts.data[key]
 	if !ok {
 		return nil, ErrNotFound