Merge pull request #426 from dolonet/fix/flaky-ci-race-and-bloom

antireplay/stable_bloom_filter_test.go

@@ -12,7 +12,7 @@ type StableBloomFilterTestSuite struct {
 }
 
 func (suite *StableBloomFilterTestSuite) TestOp() {
-	filter := antireplay.NewStableBloomFilter(500, 0.001)
+	filter := antireplay.NewStableBloomFilter(100000, 0.001)
 
 	suite.False(filter.SeenBefore([]byte{1, 2, 3}))
 	suite.False(filter.SeenBefore([]byte{4, 5, 6}))

mtglib/internal/doppel/scout.go

@@ -61,23 +61,29 @@ func (s Scout) learn(ctx context.Context, url string) (ScoutResult, error) {
 		client.CloseIdleConnections()
 	}
 
-	if err != nil || len(results.data) == 0 {
+	if err != nil {
 		return ScoutResult{}, err
 	}
 
+	data, writeIndex := results.Snapshot()
+
+	if len(data) == 0 {
+		return ScoutResult{}, nil
+	}
+
 	var result ScoutResult
 
 	// Compute inter-record durations (existing logic).
 	lastTimestamp := time.Time{}
 
-	for i, v := range results.data {
+	for i, v := range data {
 		if v.recordType != tls.TypeApplicationData {
 			continue
 		}
 
 		if lastTimestamp.IsZero() {
 			if i > 0 {
-				lastTimestamp = results.data[i-1].timestamp
+				lastTimestamp = data[i-1].timestamp
 			} else {
 				lastTimestamp = v.timestamp
 			}
@@ -90,12 +96,12 @@ func (s Scout) learn(ctx context.Context, url string) (ScoutResult, error) {
 	// Compute cert size: sum of ApplicationData payload between CCS and
 	// the first client Write (which marks the end of server handshake).
 	seenCCS := false
-	boundary := results.writeIndex
+	boundary := writeIndex
 	if boundary < 0 {
-		boundary = len(results.data)
+		boundary = len(data)
 	}
 
-	for i, v := range results.data {
+	for i, v := range data {
 		if i >= boundary {
 			break
 		}

mtglib/internal/doppel/scout_conn_collected.go

@@ -1,6 +1,10 @@
 package doppel
 
-import "time"
+import (
+	"slices"
+	"sync"
+	"time"
+)
 
 const (
 	ScoutConnCollectedPreallocSize = 100
@@ -13,23 +17,38 @@ type ScoutConnResult struct {
 }
 
 type ScoutConnCollected struct {
+	mu         sync.Mutex
 	data       []ScoutConnResult
 	writeIndex int // index at which client first wrote post-handshake data; -1 if not set
 }
 
 func (s *ScoutConnCollected) Add(record byte, payloadLen int) {
+	s.mu.Lock()
 	s.data = append(s.data, ScoutConnResult{
 		timestamp:  time.Now(),
 		recordType: record,
 		payloadLen: payloadLen,
 	})
+	s.mu.Unlock()
 }
 
 // MarkWrite records the current data length as the handshake boundary.
 func (s *ScoutConnCollected) MarkWrite() {
+	s.mu.Lock()
 	if s.writeIndex < 0 {
 		s.writeIndex = len(s.data)
 	}
+	s.mu.Unlock()
+}
+
+// Snapshot returns a copy of the collected data and the write index.
+func (s *ScoutConnCollected) Snapshot() ([]ScoutConnResult, int) {
+	s.mu.Lock()
+	snapshot := slices.Clone(s.data)
+	writeIndex := s.writeIndex
+	s.mu.Unlock()
+
+	return snapshot, writeIndex
 }
 
 func NewScoutConnCollected() *ScoutConnCollected {

mtglib/internal/doppel/scout_conn_collected_test.go

@@ -1,6 +1,7 @@
 package doppel
 
 import (
+	"sync"
 	"testing"
 	"time"
 
@@ -16,8 +17,10 @@ func (suite *ScoutConnCollectedTestSuite) TestAddSingle() {
 	collected := NewScoutConnCollected()
 	collected.Add(tls.TypeApplicationData, 100)
 
-	suite.Len(collected.data, 1)
-	suite.Equal(byte(tls.TypeApplicationData), collected.data[0].recordType)
+	data, _ := collected.Snapshot()
+
+	suite.Len(data, 1)
+	suite.Equal(byte(tls.TypeApplicationData), data[0].recordType)
 }
 
 func (suite *ScoutConnCollectedTestSuite) TestAddTimestampsAreMonotonic() {
@@ -31,11 +34,52 @@ func (suite *ScoutConnCollectedTestSuite) TestAddTimestampsAreMonotonic() {
 	time.Sleep(time.Microsecond)
 	collected.Add(tls.TypeApplicationData, 100)
 
-	for i := 1; i < len(collected.data); i++ {
-		suite.True(collected.data[i].timestamp.After(collected.data[i-1].timestamp))
+	data, _ := collected.Snapshot()
+
+	for i := 1; i < len(data); i++ {
+		suite.True(data[i].timestamp.After(data[i-1].timestamp))
 	}
 }
 
+func (suite *ScoutConnCollectedTestSuite) TestConcurrentAddSnapshot() {
+	collected := NewScoutConnCollected()
+
+	var wg sync.WaitGroup
+
+	wg.Add(3)
+
+	go func() {
+		defer wg.Done()
+
+		for i := 0; i < 1000; i++ {
+			collected.Add(tls.TypeApplicationData, i)
+		}
+	}()
+
+	go func() {
+		defer wg.Done()
+
+		for i := 0; i < 100; i++ {
+			collected.MarkWrite()
+		}
+	}()
+
+	go func() {
+		defer wg.Done()
+
+		for i := 0; i < 1000; i++ {
+			// call Snapshot concurrently to exercise the lock under -race
+			collected.Snapshot() //nolint:errcheck
+		}
+	}()
+
+	wg.Wait()
+
+	data, writeIndex := collected.Snapshot()
+	suite.Len(data, 1000)
+	suite.GreaterOrEqual(writeIndex, 0)
+}
+
 func TestScoutConnCollected(t *testing.T) {
 	t.Parallel()
 	suite.Run(t, &ScoutConnCollectedTestSuite{})

mtglib/proxy_test.go

@@ -175,7 +175,7 @@ func (suite *ProxyTestSuite) TestHTTPSRequest() {
 	addr := fmt.Sprintf("https://%s/headers", suite.ProxyAddress())
 
 	resp, err := client.Get(addr) //nolint: noctx
-	suite.NoError(err)
+	suite.Require().NoError(err)
 
 	defer resp.Body.Close() //nolint: errcheck