auto enabling priority and fairness

common/dynamicconfig/constants.go

@@ -1382,6 +1382,11 @@ second per poller by one physical queue manager`,
 The metric has 2 dimensions: namespace_id and plugin_name. Disabled by default as this is 
 an optional feature and also requires a metrics collection system that can handle higher cardinalities.`,
 	)
+	MatchingAutoEnable = NewTaskQueueBoolSetting(
+		"matching.autoEnable",
+		false,
+		`MatchingAutoEnable automatically enables fairness when a fairness key is seen`,
+	)
 
 	// keys for history
 

proto/internal/temporal/server/api/persistence/v1/task_queues.proto

@@ -134,6 +134,13 @@ message TaskQueueTypeUserData {
     DeploymentData deployment_data = 1;
 
     temporal.api.taskqueue.v1.TaskQueueConfig config = 2;
+
+    enum FairnessState {
+        FAIRNESS_STATE_UNSPECIFIED = 0;
+        FAIRNESS_STATE_V1 = 1;
+        FAIRNESS_STATE_V2 = 2;
+    };
+    FairnessState fairness_state = 3;
 }
 
 // Container for all persistent user provided data for a task queue family.

service/matching/config.go

@@ -82,6 +82,7 @@ type (
 		MembershipUnloadDelay                    dynamicconfig.DurationPropertyFn
 		TaskQueueInfoByBuildIdTTL                dynamicconfig.DurationPropertyFnWithTaskQueueFilter
 		PriorityLevels                           dynamicconfig.IntPropertyFnWithTaskQueueFilter
+		AutoEnableSub                            dynamicconfig.TypedSubscribableWithTaskQueueFilter[bool]
 
 		RateLimiterRefreshInterval    time.Duration
 		FairnessKeyRateLimitCacheSize dynamicconfig.IntPropertyFnWithTaskQueueFilter
@@ -149,6 +150,8 @@ type (
 		EnableFairness             bool
 		EnableFairnessSub          func(func(bool)) (bool, func())
 		EnableMigration            func() bool
+		AutoEnable                 bool
+		AutoEnableSub              func(func(bool)) (bool, func())
 		GetTasksBatchSize          func() int
 		GetTasksReloadAt           func() int
 		UpdateAckInterval          func() time.Duration
@@ -308,6 +311,7 @@ func NewConfig(
 		FairnessKeyRateLimitCacheSize:            dynamicconfig.MatchingFairnessKeyRateLimitCacheSize.Get(dc),
 		MaxFairnessKeyWeightOverrides:            dynamicconfig.MatchingMaxFairnessKeyWeightOverrides.Get(dc),
 		MaxIDLengthLimit:                         dynamicconfig.MaxIDLengthLimit.Get(dc),
+		AutoEnableSub:                            dynamicconfig.MatchingAutoEnable.Subscribe(dc),
 
 		AdminNamespaceToPartitionDispatchRate:          dynamicconfig.AdminMatchingNamespaceToPartitionDispatchRate.Get(dc),
 		AdminNamespaceToPartitionRateSub:               dynamicconfig.AdminMatchingNamespaceToPartitionDispatchRate.Subscribe(dc),
@@ -355,6 +359,9 @@ func newTaskQueueConfig(tq *tqid.TaskQueue, config *Config, ns namespace.Name) *
 		EnableMigration: func() bool {
 			return config.EnableMigration(ns.String(), taskQueueName, taskType)
 		},
+		AutoEnableSub: func(cb func(bool)) (bool, func()) {
+			return config.AutoEnableSub(ns.String(), taskQueueName, taskType, cb)
+		},
 		GetTasksBatchSize: func() int {
 			return config.GetTasksBatchSize(ns.String(), taskQueueName, taskType)
 		},

service/matching/matching_engine.go

@@ -430,7 +430,7 @@ func (e *matchingEngineImpl) getTaskQueuePartitionManager(
 	tqConfig.loadCause = loadCause
 	logger, throttledLogger, metricsHandler := e.loggerAndMetricsForPartition(namespaceEntry, partition, tqConfig)
 	onFatalErr := func(cause unloadCause) { newPM.unloadFromEngine(cause) }
-	onUserDataChanged := func() { newPM.userDataChanged() }
+	onUserDataChanged := func(to *persistencespb.VersionedTaskQueueUserData) { newPM.userDataChanged(to) }
 	userDataManager := newUserDataManager(
 		e.taskManager,
 		e.matchingRawClient,

service/matching/matching_engine_test.go

@@ -2119,8 +2119,7 @@ func (s *matchingEngineSuite) TestTaskQueueManagerGetTaskBatch() {
 		s.NoError(err)
 	}
 
-	tlMgr, ok := s.matchingEngine.partitions[dbq.Partition().Key()].(*taskQueuePartitionManagerImpl).defaultQueue.(*physicalTaskQueueManagerImpl)
-	s.True(ok, "taskQueueManger doesn't implement taskQueuePartitionManager interface")
+	tlMgr := s.getPhysicalTaskQueueManagerImpl(dbq)
 	s.EqualValues(taskCount, s.taskManager.getTaskCount(dbq))
 
 	// wait until all tasks are read by the task pump and enqueued into the in-memory buffer
@@ -2254,8 +2253,7 @@ func (s *matchingEngineSuite) TestTaskExpiryAndCompletion() {
 			s.NoError(err)
 		}
 
-		tlMgr, ok := s.matchingEngine.partitions[dbq.Partition().Key()].(*taskQueuePartitionManagerImpl).defaultQueue.(*physicalTaskQueueManagerImpl)
-		s.True(ok, "failed to load task queue")
+		tlMgr := s.getPhysicalTaskQueueManagerImpl(dbq)
 		s.EqualValues(taskCount, s.taskManager.getTaskCount(dbq))
 		blm := tlMgr.backlogMgr.(*backlogManagerImpl)
 
@@ -2988,7 +2986,9 @@ func (s *matchingEngineSuite) TestUpdatePhysicalTaskQueueGauge_UnVersioned() {
 	// the size of the map to 1 and it's counter to 1.
 	s.PhysicalQueueMetricValidator(capture, 1, 1)
 
-	tlmImpl, ok := tqm.(*taskQueuePartitionManagerImpl).defaultQueue.(*physicalTaskQueueManagerImpl)
+	defaultQ, err := tqm.(*taskQueuePartitionManagerImpl).managerReady.Get(context.Background())
+	s.Require().NoError(err)
+	tlmImpl := defaultQ.(*physicalTaskQueueManagerImpl)
 	s.True(ok)
 
 	s.matchingEngine.updatePhysicalTaskQueueGauge(s.ns, prtn, tlmImpl.queue.version, 1)
@@ -3223,8 +3223,9 @@ func (s *matchingEngineSuite) pollWorkflowTasksConcurrently(
 
 // getPhysicalTaskQueueManagerImpl extracts the physicalTaskQueueManagerImpl for the given PhysicalTaskQueueKey
 func (s *matchingEngineSuite) getPhysicalTaskQueueManagerImpl(ptq *PhysicalTaskQueueKey) *physicalTaskQueueManagerImpl {
-	pgMgr, ok := s.matchingEngine.partitions[ptq.Partition().Key()].(*taskQueuePartitionManagerImpl).defaultQueue.(*physicalTaskQueueManagerImpl)
-	s.True(ok, "taskQueueManger doesn't implement taskQueuePartitionManager interface")
+	defaultQ, err := s.matchingEngine.partitions[ptq.Partition().Key()].(*taskQueuePartitionManagerImpl).managerReady.Get(context.Background())
+	s.Require().NoError(err)
+	pgMgr := defaultQ.(*physicalTaskQueueManagerImpl)
 	return pgMgr
 }
 

service/matching/physical_task_queue_manager_test.go

@@ -91,9 +91,16 @@ func (s *PhysicalTaskQueueManagerTestSuite) SetupTest() {
 	s.NoError(err)
 	engine.partitions[prtn.Key()] = prtnMgr
 
+	if s.fairness {
+		prtnMgr.config.NewMatcher = true
+		prtnMgr.config.EnableFairness = true
+	} else if s.newMatcher {
+		prtnMgr.config.NewMatcher = true
+	}
+
 	s.tqMgr, err = newPhysicalTaskQueueManager(prtnMgr, s.physicalTaskQueueKey)
 	s.NoError(err)
-	prtnMgr.defaultQueue = s.tqMgr
+	prtnMgr.managerReady.Set(s.tqMgr, nil)
 }
 
 /*
@@ -334,13 +341,17 @@ func (s *PhysicalTaskQueueManagerTestSuite) TestAddTaskStandby() {
 	s.tqMgr.namespaceRegistry = mockNamespaceCache
 
 	s.tqMgr.Start()
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond)
+	err := s.tqMgr.WaitUntilInitialized(ctx)
+	s.Require().NoError(err)
 	defer s.tqMgr.Stop(unloadCauseShuttingDown)
+	cancel()
 
 	// stop taskWriter so that we can check if there's any call to it
 	// otherwise the task persist process is async and hard to test
 	s.tqMgr.tqCtxCancel()
 
-	err := s.tqMgr.SpoolTask(&persistencespb.TaskInfo{
+	err = s.tqMgr.SpoolTask(&persistencespb.TaskInfo{
 		CreateTime: timestamp.TimePtr(time.Now().UTC()),
 	})
 	s.Equal(errShutdown, err) // task writer was stopped above