Matching fairness: standing backlog test (#8028)

## What changed?
Add unit test for backlog manager that sends tasks through with standing
backlogs.

## Why?
Small framework to test concurrent operations on backlog managers.

## How did you test it?
- [x] added new unit test(s)

---------

Co-authored-by: Stephan Behnke <stephan.behnke@temporal.io>

.github/.golangci.yml

@@ -22,15 +22,15 @@ issues:
       text: "time.Sleep"
       linters:
         - forbidigo
-    - path: _test\.go|tests/.+\.go
+    - path: _test\.go|tests/.+\.go|common/testing/
       text: "panic"
       linters:
         - forbidigo
-    - path: _test\.go|tests/.+\.go
+    - path: _test\.go|tests/.+\.go|common/testing/
       text: "(cyclomatic|cognitive)"  # false positives when using subtests
       linters:
         - revive
-    - path: _test\.go|tests/.+\.go
+    - path: _test\.go|tests/.+\.go|common/testing/
       text: "(dot-imports|unchecked-type-assertion)"  # helpful in tests
       linters:
         - revive

common/testing/long.go

@@ -0,0 +1,23 @@
+package testing
+
+import (
+	"os"
+	"strconv"
+	"testing"
+)
+
+// LongTest calls Skip() on the testing.T or suite unless the environment variable
+// TEMPORAL_TEST_LONG is set to true.
+func LongTest(t any) {
+	if long, _ := strconv.ParseBool(os.Getenv("TEMPORAL_TEST_LONG")); long {
+		return
+	}
+
+	if s, ok := t.(interface{ T() *testing.T }); ok {
+		t = s.T()
+	}
+	if s, ok := t.(interface{ Skip(...any) }); ok {
+		s.Skip("skipping long test, use TEMPORAL_TEST_LONG=1 to run")
+	}
+	panic("skipping long test, use TEMPORAL_TEST_LONG=1 to run")
+}

service/matching/backlog_manager_test.go

@@ -1,25 +1,38 @@
 package matching
 
 import (
+	"container/list"
 	"context"
+	"fmt"
+	"maps"
+	"math"
+	"math/rand"
+	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/suite"
+	commonpb "go.temporal.io/api/common/v1"
 	enumspb "go.temporal.io/api/enums/v1"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
 	"go.temporal.io/server/common/dynamicconfig"
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/primitives/timestamp"
+	testutil "go.temporal.io/server/common/testing"
 	"go.temporal.io/server/common/testing/testlogger"
 	"go.temporal.io/server/common/tqid"
+	"go.temporal.io/server/common/util"
 	"go.temporal.io/server/service/matching/counter"
 	"go.uber.org/mock/gomock"
+	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
 type BacklogManagerTestSuite struct {
 	suite.Suite
 
+	cfgcli     *dynamicconfig.MemoryClient
+	cfgcol     *dynamicconfig.Collection
 	newMatcher bool
 	fairness   bool
 	logger     *testlogger.TestLogger
@@ -28,6 +41,8 @@ type BacklogManagerTestSuite struct {
 	cancelCtx  context.CancelFunc
 	taskMgr    *testTaskManager
 	ptqMgr     *MockphysicalTaskQueueManager
+
+	addSpooledTask func(*internalTask) error
 }
 
 func TestBacklogManager_Classic_Suite(t *testing.T) {
@@ -40,7 +55,7 @@ func TestBacklogManager_Pri_Suite(t *testing.T) {
 	suite.Run(t, &BacklogManagerTestSuite{newMatcher: true})
 }
 
-func TestBacklogManager_Fair_TestSuite(t *testing.T) {
+func TestBacklogManager_Fair_Suite(t *testing.T) {
 	t.Parallel()
 	suite.Run(t, &BacklogManagerTestSuite{newMatcher: true, fairness: true})
 }
@@ -54,15 +69,24 @@ func (s *BacklogManagerTestSuite) SetupTest() {
 		s.taskMgr = newTestTaskManager(s.logger)
 	}
 
-	cfg := NewConfig(dynamicconfig.NewNoopCollection())
+	s.cfgcli = dynamicconfig.NewMemoryClient()
+	s.cfgcol = dynamicconfig.NewCollection(s.cfgcli, s.logger)
+
 	f, _ := tqid.NewTaskQueueFamily("", "test-queue")
 	prtn := f.TaskQueue(enumspb.TASK_QUEUE_TYPE_WORKFLOW).NormalPartition(0)
 	queue := UnversionedQueueKey(prtn)
-	tlCfg := newTaskQueueConfig(prtn.TaskQueue(), cfg, "test-namespace")
+	tlCfg := newTaskQueueConfig(prtn.TaskQueue(), NewConfig(s.cfgcol), "test-namespace")
 
 	s.ptqMgr = NewMockphysicalTaskQueueManager(s.controller)
 	s.ptqMgr.EXPECT().QueueKey().Return(queue).AnyTimes()
 	s.ptqMgr.EXPECT().ProcessSpooledTask(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+	s.ptqMgr.EXPECT().AddSpooledTask(gomock.Any()).DoAndReturn(func(t *internalTask) error {
+		if s.addSpooledTask != nil {
+			return s.addSpooledTask(t)
+		}
+		return nil
+	}).AnyTimes()
+	s.addSpooledTask = nil
 
 	var ctx context.Context
 	ctx, s.cancelCtx = context.WithCancel(context.Background())
@@ -343,3 +367,240 @@ func totalApproximateBacklogCount(c backlogManager) (total int64) {
 	}
 	return total
 }
+
+type standingBacklogParams struct {
+	lower, upper   int64         // range of standing backlog
+	gap            int64         // add/finish tasks as long as we're within gap of the target
+	period         time.Duration // interval between peaks/troughs
+	duration       time.Duration // total duration
+	keys           int           // unique fairness keys
+	zipfS, zipfV   float64       // parameters for fairness key distribution
+	cfg            map[dynamicconfig.Key]any
+	delayInjection time.Duration
+	faultInjection float32
+}
+
+var defaultStandingBacklogParams = standingBacklogParams{
+	lower:    20,
+	upper:    200,
+	gap:      2,
+	period:   3 * time.Second,
+	duration: 5 * time.Second,
+	keys:     30,
+	zipfS:    3,
+	zipfV:    1,
+	cfg: map[dynamicconfig.Key]any{
+		// reduce these for better coverage
+		dynamicconfig.MatchingGetTasksBatchSize.Key(): 100,
+		dynamicconfig.MatchingGetTasksReloadAt.Key():  40,
+		dynamicconfig.MatchingMaxTaskBatchSize.Key():  50,
+	},
+	delayInjection: 1 * time.Millisecond,
+	faultInjection: 0.015,
+}
+
+func (s *BacklogManagerTestSuite) TestStandingBacklog_Short() {
+	s.testStandingBacklog(defaultStandingBacklogParams)
+}
+
+func (s *BacklogManagerTestSuite) TestStandingBacklog_ManyKeysUniform() {
+	testutil.LongTest(s)
+	p := defaultStandingBacklogParams
+	p.zipfS = 1.01 // not exactly uniform but closer
+	p.zipfV = 10000
+	p.keys = 10000
+	p.period = 5 * time.Second
+	p.duration = 15 * time.Second
+	s.testStandingBacklog(p)
+}
+
+func (s *BacklogManagerTestSuite) TestStandingBacklog_FullyDrain() {
+	testutil.LongTest(s)
+	p := defaultStandingBacklogParams
+	p.lower = -20
+	p.period = 3 * time.Second
+	p.duration = 15 * time.Second
+	s.testStandingBacklog(p)
+}
+
+func (s *BacklogManagerTestSuite) TestStandingBacklog_WideRange() {
+	testutil.LongTest(s)
+	p := defaultStandingBacklogParams
+	p.lower = 3
+	p.upper = 1000
+	p.period = 15 * time.Second
+	p.duration = 15 * time.Second
+	s.testStandingBacklog(p)
+}
+
+func (s *BacklogManagerTestSuite) TestStandingBacklog_FiveMin() {
+	testutil.LongTest(s)
+	p := defaultStandingBacklogParams
+	p.lower = -10
+	p.upper = 400
+	p.period = time.Minute
+	p.duration = 5 * time.Minute
+	p.cfg = maps.Clone(p.cfg)
+	p.cfg[dynamicconfig.MatchingGetTasksBatchSize.Key()] = 300
+	p.cfg[dynamicconfig.MatchingGetTasksReloadAt.Key()] = 60
+	p.delayInjection = 3 * time.Millisecond
+	s.testStandingBacklog(p)
+}
+
+func (s *BacklogManagerTestSuite) testStandingBacklog(p standingBacklogParams) {
+	if !s.newMatcher && !s.fairness {
+		s.T().Skip("TestStandingBacklogs is for priority + fairness backlog manager only")
+	}
+
+	zipf := rand.NewZipf(rand.New(rand.NewSource(time.Now().UnixNano())), p.zipfS, p.zipfV, uint64(p.keys-1))
+
+	for k, v := range p.cfg {
+		s.cfgcli.OverrideValue(k, v)
+	}
+
+	// add delays and fault injection
+	s.taskMgr.delayInjection = p.delayInjection
+	if p.faultInjection > 0 {
+		s.taskMgr.addFault("GetTasks", "Unavailable", p.faultInjection)
+		s.taskMgr.addFault("CreateTasks", "Unavailable", p.faultInjection)
+		s.logger.Expect(testlogger.Error, "Persistent store operation failure")
+	}
+
+	log := func(string, ...any) {}
+	// uncomment this for verbose logs:
+	// log = func(f string, a ...any) { fmt.Printf(f, a...) }
+
+	ctx, cancel := context.WithTimeout(context.Background(), p.duration+15*time.Second)
+	defer cancel()
+
+	var wg sync.WaitGroup
+	var lock sync.Mutex
+	var tasks list.List // this is the in-memory buffer (mock for the matcher)
+	var target, inflight, processed, index atomic.Int64
+	var tracker sync.Map // tracks tasks so we can find missing ones
+	target.Store((p.lower + p.upper) / 2)
+	const testIsOver = int64(-1000000)
+
+	s.addSpooledTask = func(t *internalTask) error {
+		lock.Lock()
+		defer lock.Unlock()
+		e := tasks.PushBack(t)
+		t.removeFromMatcher = func() {
+			lock.Lock()
+			defer lock.Unlock()
+			tasks.Remove(e)
+			log("buf evict %s -> %d\n", t.fairLevel(), tasks.Len())
+		}
+		log("buf add %s -> %d\n", t.fairLevel(), tasks.Len())
+		return nil
+	}
+	getTask := func() *internalTask {
+		lock.Lock()
+		defer lock.Unlock()
+		e := tasks.Front()
+		if e == nil {
+			return nil
+		}
+		t := tasks.Remove(e).(*internalTask)
+		log("buf remove %s -> %d\n", t.fairLevel(), tasks.Len())
+		return t
+	}
+	makeNewTask := func() *persistencespb.TaskInfo {
+		return &persistencespb.TaskInfo{
+			CreateTime:       timestamppb.Now(),
+			ScheduledEventId: index.Add(1),
+			Priority: &commonpb.Priority{
+				// TODO: add priority key option too
+				FairnessKey: fmt.Sprintf("fkey-%02d", zipf.Uint64()),
+			},
+		}
+	}
+	delta := func() int64 {
+		return inflight.Load() - target.Load()
+	}
+	sleep := func() {
+		d := time.Millisecond + time.Duration(rand.Float32()*float32(3*time.Millisecond))
+		_ = util.InterruptibleSleep(ctx, d)
+	}
+	finished := func() bool { return ctx.Err() != nil || target.Load() == testIsOver && inflight.Load() == 0 }
+	sleepUntil := func(cond func() bool) bool {
+		for !finished() && !cond() {
+			sleep()
+		}
+		return !finished()
+	}
+
+	start := time.Now()
+	s.blm.Start()
+	defer s.blm.Stop()
+	s.NoError(s.blm.WaitUntilInitialized(context.Background()))
+
+	// writer
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for sleepUntil(func() bool { return delta() <= p.gap }) {
+			info := makeNewTask()
+			tracker.Store(info.ScheduledEventId, info.Priority.FairnessKey)
+			inflight.Add(1)
+			if s.blm.SpoolTask(info) == nil {
+				log("spool %5d -> %3d\n", info.ScheduledEventId, inflight.Load())
+			} else {
+				log("spool %5d failed\n", info.ScheduledEventId, inflight.Load())
+				tracker.Delete(info.ScheduledEventId)
+				inflight.Add(-1)
+				sleep()
+			}
+		}
+	}()
+
+	// poller
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for sleepUntil(func() bool { return delta() >= -p.gap }) {
+			if t := getTask(); t != nil {
+				// TODO: error sometimes?
+				t.finish(nil, true)
+
+				tindex := t.event.Data.ScheduledEventId
+				if _, loaded := tracker.LoadAndDelete(tindex); loaded {
+					inflight.Add(-1)
+				} else {
+					// this is a duplicate task (as if matching called RecordTaskStarted twice)
+					log("finished task was not in tracker: %d\n", tindex)
+				}
+				log("finish %s -> %3d  %v  lag %5d\n", t.fairLevel(), inflight.Load(), t.getPriority().GetFairnessKey(), index.Load()-tindex)
+				processed.Add(1)
+			} else {
+				sleep()
+			}
+		}
+	}()
+
+	// adjust target over time
+	for t := target.Load(); time.Since(start) < p.duration; sleep() {
+		factor := (math.Sin(2*math.Pi*time.Since(start).Seconds()/p.period.Seconds()) + 1.0) / 2
+		next := p.lower + int64(factor*float64(p.upper-p.lower+1))
+		if t != next {
+			t = next
+			target.Store(t)
+			log("target %d", t)
+		}
+	}
+
+	// drain and wait until exited
+	s.T().Log("draining")
+	target.Store(testIsOver)
+	wg.Wait()
+
+	if !s.Zero(inflight.Load(), "did not drain all tasks!") {
+		tracker.Range(func(k, v any) bool {
+			s.T().Logf("  outstanding task: %d %s", k.(int64), v.(string))
+			return true
+		})
+	}
+
+	elapsed := time.Since(start)
+	s.T().Logf("processed %d tasks, %.3f/s", processed.Load(), float64(processed.Load())/elapsed.Seconds())
+}

service/matching/matching_engine_test.go

@@ -2839,7 +2839,6 @@ func (s *matchingEngineSuite) TestUnloadOnMembershipChange() {
 
 	config := s.newConfig()
 	config.MembershipUnloadDelay = dynamicconfig.GetDurationPropertyFn(10 * time.Millisecond)
-	// TODO(fairness): why is this calling s.newMatchingEngine instead of using s.matchingEngine?
 	e := s.newMatchingEngine(config, s.classicTaskManager, s.fairTaskManager)
 	e.Start()
 	defer e.Stop()
@@ -3615,6 +3614,7 @@ type testTaskManager struct {
 	updateMetadataOnCreateTasks bool
 
 	faultInjection map[string]float32 // "op:error" -> fraction of time
+	delayInjection time.Duration
 }
 
 type dbTaskQueueKey struct {
@@ -3716,6 +3716,10 @@ func (m *testTaskManager) CreateTaskQueue(
 ) (*persistence.CreateTaskQueueResponse, error) {
 	tli := request.TaskQueueInfo
 	tlm := m.getQueueData(tli.Name, tli.NamespaceId, tli.TaskType)
+
+	m.delay()
+	defer m.delay()
+
 	tlm.Lock()
 	defer tlm.Unlock()
 
@@ -3735,6 +3739,10 @@ func (m *testTaskManager) UpdateTaskQueue(
 ) (*persistence.UpdateTaskQueueResponse, error) {
 	tli := request.TaskQueueInfo
 	tlm := m.getQueueData(tli.Name, tli.NamespaceId, tli.TaskType)
+
+	m.delay()
+	defer m.delay()
+
 	tlm.Lock()
 	defer tlm.Unlock()
 
@@ -3796,6 +3804,10 @@ func (m *testTaskManager) CompleteTasksLessThan(
 	} else if !m.fairness && request.ExclusiveMaxPass != 0 {
 		return 0, serviceerror.NewInternal("invalid CompleteTasksLessThan request on queue")
 	}
+
+	m.delay()
+	defer m.delay()
+
 	tlm := m.getQueueData(request.TaskQueueName, request.NamespaceID, request.TaskType)
 	tlm.Lock()
 	defer tlm.Unlock()
@@ -3833,6 +3845,12 @@ func (m *testTaskManager) DeleteTaskQueue(
 	return nil
 }
 
+func (m *testTaskManager) delay() {
+	if m.delayInjection > 0 && rand.Int31n(128) >= 13 {
+		time.Sleep(time.Duration(rand.Float32() * float32(m.delayInjection))) // nolint:forbidigo
+	}
+}
+
 // all calls to addFault should be done before starting to call methods on testTaskManager
 func (m *testTaskManager) addFault(method, err string, fraction float32) {
 	if m.faultInjection == nil {
@@ -3854,6 +3872,9 @@ func (m *testTaskManager) CreateTasks(
 	taskType := request.TaskQueueInfo.Data.TaskType
 	rangeID := request.TaskQueueInfo.RangeID
 
+	m.delay()
+	defer m.delay()
+
 	if m.fault("CreateTasks", "ConditionFailed") {
 		return nil, &persistence.ConditionFailedError{Msg: "Fake ConditionFailedError"}
 	} else if m.fault("CreateTasks", "Unavailable") {
@@ -3918,6 +3939,9 @@ func (m *testTaskManager) GetTasks(
 		return nil, serviceerror.NewInternal("invalid GetTasks request on queue")
 	}
 
+	m.delay()
+	defer m.delay()
+
 	if m.fault("GetTasks", "Unavailable") {
 		return nil, serviceerror.NewUnavailablef("GetTasks operation failed")
 	}

service/matching/task.go

@@ -264,6 +264,10 @@ func (task *internalTask) getPriority() *commonpb.Priority {
 	return nil
 }
 
+func (task *internalTask) fairLevel() fairLevel {
+	return fairLevelFromAllocatedTask(task.event.AllocatedTaskInfo)
+}
+
 // finish marks a task as finished. Should be called after a poller picks up a task
 // and marks it as started. If the task is unable to marked as started, then this
 // method should be called with a non-nil error argument.