Merge pull request #405 fix flaky test TestWriterReconnector_Write_QueueLimit

Author: rekby

internal/topic/topicwriterinternal/writer_reconnector_test.go

@@ -263,47 +263,48 @@ func TestWriterImpl_WriteCodecs(t *testing.T) {
 }
 
 func TestWriterReconnector_Write_QueueLimit(t *testing.T) {
-	ctx := xtest.Context(t)
-	w := newWriterReconnectorStopped(newWriterReconnectorConfig(
-		WithAutoSetSeqNo(false),
-		WithMaxQueueLen(2),
-	))
-	w.firstConnectionHandled.Store(true)
-
-	waitStartQueueWait := func() {
-		xtest.SpinWaitCondition(t, nil, func() bool {
-			if w.semaphore.TryAcquire(1) {
-				w.semaphore.Release(1)
-				return false
-			}
-			return true
-		})
-	}
+	xtest.TestManyTimes(t, func(t testing.TB) {
+		ctx := xtest.Context(t)
+		w := newWriterReconnectorStopped(newWriterReconnectorConfig(
+			WithAutoSetSeqNo(false),
+			WithMaxQueueLen(2),
+		))
+		w.firstConnectionHandled.Store(true)
 
-	err := w.Write(ctx, newTestMessages(1, 2))
-	require.NoError(t, err)
+		waitStartQueueWait := func(targetWaiters int) {
+			xtest.SpinWaitCondition(t, nil, func() bool {
+				res := getWaitersCount(w.semaphore) == targetWaiters
+				return res
+			})
+		}
 
-	ctxNoQueueSpace, ctxNoQueueSpaceCancel := context.WithCancel(ctx)
+		err := w.Write(ctx, newTestMessages(1, 2))
+		require.NoError(t, err)
 
-	go func() {
-		waitStartQueueWait()
-		ctxNoQueueSpaceCancel()
-	}()
-	err = w.Write(ctxNoQueueSpace, newTestMessages(3))
-	require.ErrorIs(t, err, PublicErrQueueIsFull)
+		ctxNoQueueSpace, ctxNoQueueSpaceCancel := context.WithCancel(ctx)
 
-	go func() {
-		waitStartQueueWait()
-		ackErr := w.queue.AcksReceived([]rawtopicwriter.WriteAck{
-			{
-				SeqNo: 1,
-			},
-		})
-		require.NoError(t, ackErr)
-	}()
+		go func() {
+			waitStartQueueWait(1)
+			ctxNoQueueSpaceCancel()
+		}()
+		err = w.Write(ctxNoQueueSpace, newTestMessages(3))
+		if !errors.Is(err, PublicErrQueueIsFull) {
+			require.ErrorIs(t, err, PublicErrQueueIsFull)
+		}
+
+		go func() {
+			waitStartQueueWait(1)
+			ackErr := w.queue.AcksReceived([]rawtopicwriter.WriteAck{
+				{
+					SeqNo: 1,
+				},
+			})
+			require.NoError(t, ackErr)
+		}()
 
-	err = w.Write(ctx, newTestMessages(3))
-	require.NoError(t, err)
+		err = w.Write(ctx, newTestMessages(3))
+		require.NoError(t, err)
+	})
 }
 
 func TestEnv(t *testing.T) {

internal/topic/topicwriterinternal/writer_reconnector_unsafe_test.go

@@ -0,0 +1,28 @@
+package topicwriterinternal
+
+import (
+	"container/list"
+	"reflect"
+	"runtime"
+	"sync"
+	"unsafe"
+
+	"golang.org/x/sync/semaphore"
+)
+
+func getWaitersCount(sem *semaphore.Weighted) int {
+	defer runtime.KeepAlive(sem)
+
+	semVal := reflect.ValueOf(sem).Elem()
+	mutexField := semVal.FieldByName("mu")
+
+	mutexAddr := unsafe.Pointer(mutexField.UnsafeAddr())
+	mutex := (*sync.Mutex)(mutexAddr)
+	mutex.Lock()
+	defer mutex.Unlock()
+
+	waitersField := semVal.FieldByName("waiters")
+	waitersPointer := unsafe.Pointer(waitersField.UnsafeAddr())
+	waiters := (*list.List)(waitersPointer)
+	return waiters.Len()
+}