Don't attempt to dispatch expired activities (#7204)

We check whether an activity is expired before putting it into the buffer, but once it's there the only way we will find out that it has expired is from the Record(Decision/Activity)TaskStarted call returning an error. This is particularly problematic for TaskLists that have an Activity Rate Limit, since they'll only be able to remove tasks from the buffer according to that Rate Limit. They can enter a steady state where new non-expired tasks are added to the end of the buffer but they'll always expire by the time they actually are attempted.

Eagerly removing expired tasks ensures we can promptly dispatch relevant ones.

Author: natemort

service/matching/tasklist/task_reader.go

@@ -427,6 +427,19 @@ func (tr *taskReader) dispatchSingleTaskFromBufferWithRetries(taskInfo *persiste
 }
 
 func (tr *taskReader) dispatchSingleTaskFromBuffer(taskInfo *persistence.TaskInfo) (breakDispatchLoop bool, breakRetries bool) {
+	e := event.E{
+		TaskListName: tr.taskListID.GetName(),
+		TaskListType: tr.taskListID.GetType(),
+		TaskListKind: &tr.tlMgr.taskListKind,
+		TaskInfo:     *taskInfo,
+	}
+	if tr.isTaskExpired(taskInfo) {
+		e.EventName = "Task Expired"
+		event.Log(e)
+		tr.scope.IncCounter(metrics.ExpiredTasksPerTaskListCounter)
+		tr.taskAckManager.AckItem(taskInfo.TaskID)
+		return false, true
+	}
 	isolationGroup, isolationDuration := tr.getIsolationGroupForTask(tr.cancelCtx, taskInfo)
 	_, isolationGroupIsKnown := tr.taskBuffers[isolationGroup]
 	if !isolationGroupIsKnown {
@@ -440,13 +453,6 @@ func (tr *taskReader) dispatchSingleTaskFromBuffer(taskInfo *persistence.TaskInf
 	timerScope.Stop()
 	cancel()
 
-	e := event.E{
-		TaskListName: tr.taskListID.GetName(),
-		TaskListType: tr.taskListID.GetType(),
-		TaskListKind: &tr.tlMgr.taskListKind,
-		TaskInfo:     *taskInfo,
-	}
-
 	if err == nil {
 		e.EventName = "Dispatched Buffered Task"
 		event.Log(e)
@@ -488,9 +494,9 @@ func (tr *taskReader) dispatchSingleTaskFromBuffer(taskInfo *persistence.TaskInf
 	}
 
 	if errors.Is(err, errTaskNotStarted) {
-		e.EventName = "Dispatch failed on completing task on the passive side because task not started. Will retry dispatch if task is not expired"
+		e.EventName = "Dispatch failed on completing task on the passive side because task not started."
 		event.Log(e)
-		return false, tr.isTaskExpired(taskInfo)
+		return false, false
 	}
 
 	if errors.Is(err, errWaitTimeNotReachedForEntityNotExists) {

service/matching/tasklist/task_reader_test.go

@@ -54,14 +54,14 @@ var defaultIsolationGroups = []string{
 func TestDispatchSingleTaskFromBuffer(t *testing.T) {
 	testCases := []struct {
 		name          string
-		allowances    func(t *testing.T, reader *taskReader)
+		allowances    func(t *testing.T, reader *taskReader, mockTime clock.MockedTimeSource)
 		ttl           int
 		breakDispatch bool
 		breakRetries  bool
 	}{
 		{
 			name: "success - no isolation",
-			allowances: func(t *testing.T, reader *taskReader) {
+			allowances: func(t *testing.T, reader *taskReader, mockTime clock.MockedTimeSource) {
 				reader.getIsolationGroupForTask = func(ctx context.Context, info *persistence.TaskInfo) (string, time.Duration) {
 					return "", -1
 				}
@@ -77,7 +77,7 @@ func TestDispatchSingleTaskFromBuffer(t *testing.T) {
 		},
 		{
 			name: "success - isolation",
-			allowances: func(t *testing.T, reader *taskReader) {
+			allowances: func(t *testing.T, reader *taskReader, mockTime clock.MockedTimeSource) {
 				reader.getIsolationGroupForTask = func(ctx context.Context, info *persistence.TaskInfo) (string, time.Duration) {
 					return defaultIsolationGroup, -1
 				}
@@ -93,7 +93,7 @@ func TestDispatchSingleTaskFromBuffer(t *testing.T) {
 		},
 		{
 			name: "success - unknown isolation group",
-			allowances: func(t *testing.T, reader *taskReader) {
+			allowances: func(t *testing.T, reader *taskReader, mockTime clock.MockedTimeSource) {
 				reader.getIsolationGroupForTask = func(ctx context.Context, info *persistence.TaskInfo) (string, time.Duration) {
 					return "mystery group", -1
 				}
@@ -108,9 +108,24 @@ func TestDispatchSingleTaskFromBuffer(t *testing.T) {
 			breakDispatch: false,
 			breakRetries:  true,
 		},
+		{
+			name: "success - skip expired tasks",
+			allowances: func(t *testing.T, reader *taskReader, mockTime clock.MockedTimeSource) {
+				reader.getIsolationGroupForTask = func(ctx context.Context, info *persistence.TaskInfo) (string, time.Duration) {
+					return defaultIsolationGroup, -1
+				}
+				reader.dispatchTask = func(ctx context.Context, task *InternalTask) error {
+					t.Fatal("task must not be dispatched")
+					return nil
+				}
+			},
+			ttl:           -2,
+			breakDispatch: false,
+			breakRetries:  true,
+		},
 		{
 			name: "Error - context cancelled, should stop",
-			allowances: func(t *testing.T, reader *taskReader) {
+			allowances: func(t *testing.T, reader *taskReader, mockTime clock.MockedTimeSource) {
 				reader.getIsolationGroupForTask = func(ctx context.Context, info *persistence.TaskInfo) (string, time.Duration) {
 					return defaultIsolationGroup, -1
 				}
@@ -124,7 +139,7 @@ func TestDispatchSingleTaskFromBuffer(t *testing.T) {
 		},
 		{
 			name: "Error - Deadline Exceeded, should retry",
-			allowances: func(t *testing.T, reader *taskReader) {
+			allowances: func(t *testing.T, reader *taskReader, mockTime clock.MockedTimeSource) {
 				reader.getIsolationGroupForTask = func(ctx context.Context, info *persistence.TaskInfo) (string, time.Duration) {
 					return defaultIsolationGroup, -1
 				}
@@ -138,7 +153,7 @@ func TestDispatchSingleTaskFromBuffer(t *testing.T) {
 		},
 		{
 			name: "Error - throttled, should retry",
-			allowances: func(t *testing.T, reader *taskReader) {
+			allowances: func(t *testing.T, reader *taskReader, mockTime clock.MockedTimeSource) {
 				reader.getIsolationGroupForTask = func(ctx context.Context, info *persistence.TaskInfo) (string, time.Duration) {
 					return defaultIsolationGroup, -1
 				}
@@ -151,7 +166,7 @@ func TestDispatchSingleTaskFromBuffer(t *testing.T) {
 		},
 		{
 			name: "Error - unknown, should retry",
-			allowances: func(t *testing.T, reader *taskReader) {
+			allowances: func(t *testing.T, reader *taskReader, mockTime clock.MockedTimeSource) {
 				reader.getIsolationGroupForTask = func(ctx context.Context, info *persistence.TaskInfo) (string, time.Duration) {
 					return defaultIsolationGroup, -1
 				}
@@ -164,7 +179,7 @@ func TestDispatchSingleTaskFromBuffer(t *testing.T) {
 		},
 		{
 			name: "Error - task not started and not expired, should retry",
-			allowances: func(t *testing.T, reader *taskReader) {
+			allowances: func(t *testing.T, reader *taskReader, mockTime clock.MockedTimeSource) {
 				reader.getIsolationGroupForTask = func(ctx context.Context, info *persistence.TaskInfo) (string, time.Duration) {
 					return defaultIsolationGroup, -1
 				}
@@ -177,22 +192,23 @@ func TestDispatchSingleTaskFromBuffer(t *testing.T) {
 			breakRetries:  false,
 		},
 		{
-			name: "Error - task not started and expired, should not retry",
-			allowances: func(t *testing.T, reader *taskReader) {
+			name: "Error - task not started and expired, should retry",
+			allowances: func(t *testing.T, reader *taskReader, mockTime clock.MockedTimeSource) {
 				reader.getIsolationGroupForTask = func(ctx context.Context, info *persistence.TaskInfo) (string, time.Duration) {
 					return defaultIsolationGroup, -1
 				}
 				reader.dispatchTask = func(ctx context.Context, task *InternalTask) error {
+					mockTime.Advance(time.Hour)
 					return errTaskNotStarted
 				}
 			},
-			ttl:           -2,
+			ttl:           1,
 			breakDispatch: false,
-			breakRetries:  true,
+			breakRetries:  false,
 		},
 		{
 			name: "Error - time not reached to complete task without workflow execution, should retry",
-			allowances: func(t *testing.T, reader *taskReader) {
+			allowances: func(t *testing.T, reader *taskReader, mockTime clock.MockedTimeSource) {
 				reader.getIsolationGroupForTask = func(ctx context.Context, info *persistence.TaskInfo) (string, time.Duration) {
 					return defaultIsolationGroup, -1
 				}
@@ -211,7 +227,7 @@ func TestDispatchSingleTaskFromBuffer(t *testing.T) {
 			c := defaultConfig()
 			tlm := createTestTaskListManagerWithConfig(t, testlogger.New(t), controller, c, timeSource)
 			reader := tlm.taskReader
-			tc.allowances(t, reader)
+			tc.allowances(t, reader, timeSource)
 			taskInfo := newTask(timeSource)
 			taskInfo.Expiry = timeSource.Now().Add(time.Duration(tc.ttl) * time.Second)