enable race unit-tests (#708)

Author: adecaro

.github/workflows/tests.yml

@@ -33,7 +33,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        tests: [unit-tests]
+        tests: [unit-tests-race]
 
     steps:
       - name: Checkout code

Makefile

@@ -49,9 +49,9 @@ run-optl:
 
 
 .PHONY: unit-tests-race
-unit-tests-race:
+unit-tests-race: testing-docker-images
 	@export GORACE=history_size=7; export FAB_BINS=$(FAB_BINS); go test -race -cover $(shell go list ./... | grep -v '/integration/')
-	cd integration/nwo/; export FAB_BINS=$(FAB_BINS); go test -cover ./...
+	cd integration/nwo/; export FAB_BINS=$(FAB_BINS); go test -race -cover ./...
 
 .PHONY: docker-images
 docker-images: fabric-docker-images weaver-docker-images orion-server-images monitoring-docker-images testing-docker-images

platform/common/utils/lazy/holder.go

@@ -37,7 +37,7 @@ type lazyHolder[V any] struct {
 func (h *lazyHolder[V]) Get() (V, error) {
 	h.mu.RLock()
 	if h.set {
-		h.mu.RUnlock()
+		defer h.mu.RUnlock()
 		return h.v, nil
 	}
 	h.mu.RUnlock()

platform/common/utils/lazy/holder_test.go

@@ -0,0 +1,69 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package lazy
+
+import (
+	"errors"
+	"io"
+	"sync"
+	"testing"
+	"time"
+)
+
+type dummyCloser struct {
+	mu sync.Mutex
+	v  int
+}
+
+func (d *dummyCloser) Close() error {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	d.v = 0
+	return nil
+}
+
+func TestLazyHolderRaceCondition(t *testing.T) {
+	// Provider function for the lazyHolder.
+	provider := func() (*dummyCloser, error) {
+		time.Sleep(10 * time.Millisecond) // Simulate some delay
+		return &dummyCloser{v: 42}, nil
+	}
+
+	// Test for race conditions
+	holder := NewCloserHolder(provider)
+
+	var wg sync.WaitGroup
+
+	// Concurrently call Get.
+	for i := 0; i < 10; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			v, err := holder.Get()
+			if err != nil {
+				t.Errorf("Get() error: %v", err)
+			}
+			if v == nil {
+				t.Errorf("Get() returned nil value")
+			}
+		}()
+	}
+
+	// Concurrently call Reset.
+	for i := 0; i < 5; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			err := holder.Reset()
+			if err != nil && !errors.Is(err, io.EOF) {
+				t.Errorf("Reset() error: %v", err)
+			}
+		}()
+	}
+
+	wg.Wait()
+}

platform/fabric/core/generic/chaincode/discovery.go

@@ -132,11 +132,10 @@ func (d *Discovery) Response() (discovery.Response, error) {
 	// Do we have a response already?
 	d.chaincode.discoveryResultsCacheLock.RLock()
 	responseBoxed, err := d.chaincode.discoveryResultsCache.Get(key)
+	d.chaincode.discoveryResultsCacheLock.RUnlock()
 	if responseBoxed != nil && err == nil {
-		d.chaincode.discoveryResultsCacheLock.RUnlock()
 		return responseBoxed.(discovery.Response), nil
 	}
-	d.chaincode.discoveryResultsCacheLock.RUnlock()
 
 	d.chaincode.discoveryResultsCacheLock.Lock()
 	defer d.chaincode.discoveryResultsCacheLock.Unlock()

platform/fabric/core/generic/chaincode/manager.go

@@ -67,11 +67,10 @@ func NewManager(
 func (c *Manager) Chaincode(name string) driver.Chaincode {
 	c.ChaincodesLock.RLock()
 	ch, ok := c.Chaincodes[name]
+	c.ChaincodesLock.RUnlock()
 	if ok {
-		c.ChaincodesLock.RUnlock()
 		return ch
 	}
-	c.ChaincodesLock.RUnlock()
 
 	c.ChaincodesLock.Lock()
 	defer c.ChaincodesLock.Unlock()

platform/fabric/events_test.go

@@ -20,18 +20,24 @@ import (
 
 type mockSubscriber struct {
 	listener events.Listener
-	sync.RWMutex
+	m        sync.RWMutex
 }
 
 func (m *mockSubscriber) Subscribe(chaincodeName string, listener events.Listener) {
+	m.m.Lock()
+	defer m.m.Unlock()
 	m.listener = listener
 }
 
 func (m *mockSubscriber) Unsubscribe(chaincodeName string, listener events.Listener) {
+	m.m.Lock()
+	defer m.m.Unlock()
 	m.listener = nil
 }
 
 func (m *mockSubscriber) Publish(chaincodeName string, event *committer.ChaincodeEvent) {
+	m.m.RLock()
+	defer m.m.RUnlock()
 	if m.listener != nil {
 		m.listener.OnReceive(event)
 	}

platform/view/services/comm/host/rest/websocket/stream_test.go

@@ -77,13 +77,18 @@ func TestWriter(t *testing.T) {
 	assert.NoError(t, stream.Close())
 
 	output := make([][]byte, 0, len(input))
+	m := sync.RWMutex{}
 	go func() {
 		for written := range conn.WrittenValues() {
+			m.Lock()
 			output = append(output, written)
+			m.Unlock()
 		}
 	}()
 
 	assert.Eventually(t, func() bool {
+		m.RLock()
+		defer m.RUnlock()
 		return len(input) == len(output)
 	}, 5*time.Second, time.Second)
 }

platform/view/services/comm/session.go

@@ -28,11 +28,12 @@ type NetworkStreamSession struct {
 	incoming        chan *view.Message
 	streams         map[*streamHandler]struct{}
 	closed          bool
-	mutex           sync.Mutex
+	mutex           sync.RWMutex
 }
 
 func (n *NetworkStreamSession) Info() view.SessionInfo {
-	n.mutex.Lock()
+	n.mutex.RLock()
+	defer n.mutex.RUnlock()
 	ret := view.SessionInfo{
 		ID:           n.sessionID,
 		Caller:       n.caller,
@@ -41,7 +42,6 @@ func (n *NetworkStreamSession) Info() view.SessionInfo {
 		EndpointPKID: n.endpointID,
 		Closed:       n.closed,
 	}
-	n.mutex.Unlock()
 	return ret
 }
 
@@ -111,20 +111,25 @@ func (n *NetworkStreamSession) closeInternal() {
 }
 
 func (n *NetworkStreamSession) sendWithStatus(ctx context.Context, payload []byte, status int32) error {
+	n.mutex.RLock()
 	info := host.StreamInfo{
 		RemotePeerID:      string(n.endpointID),
 		RemotePeerAddress: n.endpointAddress,
 		ContextID:         n.contextID,
 		SessionID:         n.sessionID,
 	}
-	err := n.node.sendTo(ctx, info, &ViewPacket{
+	packet := &ViewPacket{
 		ContextID: n.contextID,
 		SessionID: n.sessionID,
 		Caller:    n.callerViewID,
 		Status:    status,
 		Payload:   payload,
-	})
+	}
+	n.mutex.RUnlock()
+
+	err := n.node.sendTo(ctx, info, packet)
 	if logger.IsEnabledFor(zapcore.DebugLevel) {
+		n.mutex.RLock()
 		logger.Debugf(
 			"sent message [len:%d] to [%s:%s] from [%s] with err [%s]",
 			len(payload),
@@ -144,6 +149,7 @@ func (n *NetworkStreamSession) sendWithStatus(ctx context.Context, payload []byt
 				debug.Stack(),
 			)
 		}
+		n.mutex.RUnlock()
 	}
 	return err
 }

platform/view/services/kvs/kvs_test.go

@@ -134,7 +134,7 @@ func testParallelWrites(t *testing.T, driver driver.Driver, cp kvs.ConfigProvide
 	assert.NoError(t, err)
 	for i := 0; i < n; i++ {
 		go func(i int) {
-			err = kvstore.Put(k1, &stuff{"santa", 1})
+			err := kvstore.Put(k1, &stuff{"santa", 1})
 			assert.NoError(t, err)
 			defer wg.Done()
 		}(i)