Protect against legacy queries when state is destroyed (#1568)

Author: Quinn-With-Two-Ns

internal/internal_task_handlers.go

@@ -769,14 +769,14 @@ func (wth *workflowTaskHandlerImpl) GetOrCreateWorkflowContext(
 		if task.Query != nil && !isFullHistory && wth == workflowContext.wth && !workflowContext.IsDestroyed() {
 			// query task and we have a valid cached state
 			metricsHandler.Counter(metrics.StickyCacheHit).Inc(1)
-		} else if history.Events[0].GetEventId() == workflowContext.previousStartedEventID+1 && wth == workflowContext.wth && !workflowContext.IsDestroyed() {
+		} else if len(history.Events) > 0 && history.Events[0].GetEventId() == workflowContext.previousStartedEventID+1 && wth == workflowContext.wth && !workflowContext.IsDestroyed() {
 			// non query task and we have a valid cached state
 			metricsHandler.Counter(metrics.StickyCacheHit).Inc(1)
 		} else {
 			// possible another task already destroyed this context.
 			if !workflowContext.IsDestroyed() {
 				// non query task and cached state is missing events, we need to discard the cached state and build a new one.
-				if history.Events[0].GetEventId() != workflowContext.previousStartedEventID+1 {
+				if len(history.Events) > 0 && history.Events[0].GetEventId() != workflowContext.previousStartedEventID+1 {
 					wth.logger.Debug("Cached state staled, new task has unexpected events",
 						tagWorkflowID, task.WorkflowExecution.GetWorkflowId(),
 						tagRunID, task.WorkflowExecution.GetRunId(),

internal/internal_task_handlers_test.go

@@ -650,6 +650,34 @@ func (t *TaskHandlersTestSuite) TestRespondsToWFTWithWorkerBinaryID() {
 	params.cache.getWorkflowCache().Delete(task.WorkflowExecution.RunId)
 }
 
+func (t *TaskHandlersTestSuite) TestStickyLegacyQueryTaskOnEvictedCache() {
+	taskQueue := "tq1"
+	testEvents := []*historypb.HistoryEvent{
+		createTestEventWorkflowExecutionStarted(1, &historypb.WorkflowExecutionStartedEventAttributes{TaskQueue: &taskqueuepb.TaskQueue{Name: taskQueue}}),
+		createTestEventWorkflowTaskScheduled(2, &historypb.WorkflowTaskScheduledEventAttributes{TaskQueue: &taskqueuepb.TaskQueue{Name: taskQueue}}),
+		createTestEventWorkflowTaskStarted(3),
+	}
+	task := createWorkflowTask(testEvents, 0, "HelloWorld_Workflow")
+	params := t.getTestWorkerExecutionParams()
+	taskHandler := newWorkflowTaskHandler(params, nil, t.registry)
+	wftask := workflowTask{task: task}
+	wfctx := t.mustWorkflowContextImpl(&wftask, taskHandler)
+	wfctx.Unlock(nil)
+	wfctx.clearState()
+	// Now make the task look like a legacy query task on the sticky queue
+	task.History = &historypb.History{}
+	task.Query = &querypb.WorkflowQuery{}
+	wfQueryTask := workflowTask{task: task, historyIterator: &historyIteratorImpl{
+		iteratorFunc: func(nextToken []byte) (*historypb.History, []byte, error) {
+			return &historypb.History{Events: testEvents}, nil, nil
+		},
+	}}
+	wfctx = t.mustWorkflowContextImpl(&wfQueryTask, taskHandler)
+	t.NotNil(wfctx)
+	// clean up workflow left in cache
+	params.cache.getWorkflowCache().Delete(task.WorkflowExecution.RunId)
+}
+
 func (t *TaskHandlersTestSuite) TestWorkflowTask_ActivityTaskScheduled() {
 	// Schedule an activity and see if we complete workflow.
 	taskQueue := "tq1"