feat: Write DLQ tasks when Standby processing timeout is reached (#8022)

<!-- If you are new to contributing or want a refresher, please read
./pull_request_guidance.md -->
**What changed?**

Adds a new post action function that writes to the History Task DLQ
rather than discarding the task.

**Why?**

Discarding a task is a complete loss of data. In rare scenarios, it
could be possible that tasks are discarded from both clusters (in a
multi-cluster setup), resulting in the inability to continue a workflow.

**How did you test it?**

To run the unit tests:

```
go test ./service/history/...
```

To run a full integration test requires additional persistence changes
which are coming in a follow up PR.

**Potential risks**

This is hidden behind a feature flag. Premature enablement of this flag
should not result in running the code path, as the DLQWriter has yet to
be wired up to enable writing to the DLQ.

**Release notes**

N/A

**Documentation Changes**

N/A

Author: c-warren

service/history/config/config.go

@@ -345,6 +345,10 @@ type Config struct {
 	GlobalRatelimiterDecayAfter     dynamicproperties.DurationPropertyFn
 	GlobalRatelimiterGCAfter        dynamicproperties.DurationPropertyFn
 
+	// History Task DLQ Configuration
+	HistoryTaskDLQMode            dynamicproperties.StringPropertyFnWithDomainFilter
+	HistoryTaskProcessingInterval dynamicproperties.DurationPropertyFnWithShardIDFilter
+
 	// HostName for machine running the service
 	HostName string
 }
@@ -612,6 +616,9 @@ func New(dc *dynamicconfig.Collection, numberOfShards int, maxMessageSize int, i
 		GlobalRatelimiterDecayAfter:     dc.GetDurationProperty(dynamicproperties.HistoryGlobalRatelimiterDecayAfter),
 		GlobalRatelimiterGCAfter:        dc.GetDurationProperty(dynamicproperties.HistoryGlobalRatelimiterGCAfter),
 
+		HistoryTaskDLQMode:            dc.GetStringPropertyFilteredByDomain(dynamicproperties.HistoryTaskDeadLetterQueueMode),
+		HistoryTaskProcessingInterval: dc.GetDurationPropertyFilteredByShardID(dynamicproperties.HistoryTaskDLQProcessorInterval),
+
 		HostName: hostname,
 	}
 

service/history/config/config_test.go

@@ -282,6 +282,8 @@ func TestNewConfig(t *testing.T) {
 		"EnableCorruptionAutoRepair":                           {dynamicproperties.EnableCorruptionAutoRepair, true},
 		"CorruptionRepairTimeout":                              {dynamicproperties.CorruptionRepairTimeout, time.Duration(1)},
 		"RequireChecksumMatchAfterRebuildRepair":               {dynamicproperties.RequireChecksumMatchAfterRebuildRepair, true},
+		"HistoryTaskDLQMode":                                   {dynamicproperties.HistoryTaskDeadLetterQueueMode, "enabled"},
+		"HistoryTaskProcessingInterval":                        {dynamicproperties.HistoryTaskDLQProcessorInterval, time.Second},
 	}
 	client := dynamicconfig.NewInMemoryClient()
 	for fieldName, expected := range fields {

service/history/constants/constants.go

@@ -36,3 +36,12 @@ var (
 	ErrHistoryHostThrottle     = &types.ServiceBusyError{Message: "History host rps exceeded"}
 	ErrShuttingDown            = &types.InternalServiceError{Message: "Shutting down"}
 )
+
+const (
+	// HistoryTaskDLQModeEnabled enables writing tasks to the DLQ.
+	HistoryTaskDLQModeEnabled = "enabled"
+	// HistoryTaskDLQModeDisabled disables writing tasks to the DLQ.
+	HistoryTaskDLQModeDisabled = "disabled"
+	// HistoryTaskDLQModeShadow enables writing tasks to the DLQ but does not process the task.
+	HistoryTaskDLQModeShadow = "shadow"
+)

service/history/queue/timer_queue_processor.go

@@ -128,6 +128,7 @@ func NewTimerQueueProcessor(
 			shard.GetMetricsClient(),
 			clusterName,
 			config,
+			nil, // TODO(c-warren): wire DLQ writer once persistence layer is written
 		)
 		standbyTaskExecutors = append(standbyTaskExecutors, standbyTaskExecutor)
 		standbyQueueProcessors[clusterName], standbyQueueTimerGates[clusterName] = newTimerQueueStandbyProcessor(

service/history/queue/transfer_queue_processor.go

@@ -136,6 +136,7 @@ func NewTransferQueueProcessor(
 			standByLogger,
 			clusterName,
 			config,
+			nil, // TODO(c-warren): wire DLQ writer once persistence layer is written
 		)
 		standbyQueueProcessors[clusterName] = newTransferQueueStandbyProcessor(
 			clusterName,

service/history/queuev2/timer_queue_factory.go

@@ -122,6 +122,7 @@ func (f *timerQueueFactory) createQueuev2(
 		shard.GetMetricsClient(),
 		shard.GetClusterMetadata().GetCurrentClusterName(),
 		shard.GetConfig(),
+		nil, // TODO(c-warren): wire DLQ writer once persistence layer is written
 	)
 	executorWrapper := task.NewExecutorWrapper(
 		shard.GetClusterMetadata().GetCurrentClusterName(),

service/history/queuev2/transfer_queue_factory.go

@@ -128,6 +128,7 @@ func (f *transferQueueFactory) createQueuev2(
 		logger,
 		shard.GetClusterMetadata().GetCurrentClusterName(),
 		shard.GetConfig(),
+		nil, // TODO(c-warren): wire DLQ writer once persistence layer is written
 	)
 
 	executorWrapper := task.NewExecutorWrapper(

service/history/task/standby_task_util.go

@@ -28,18 +28,27 @@ import (
 
 	"github.com/uber/cadence/common"
 	"github.com/uber/cadence/common/activecluster"
+	"github.com/uber/cadence/common/dynamicconfig/dynamicproperties"
 	"github.com/uber/cadence/common/log"
 	"github.com/uber/cadence/common/log/tag"
 	"github.com/uber/cadence/common/persistence"
 	"github.com/uber/cadence/common/types"
+	"github.com/uber/cadence/service/history/constants"
 	"github.com/uber/cadence/service/history/execution"
+	"github.com/uber/cadence/service/history/shard"
+	"github.com/uber/cadence/service/history/taskdlq"
 )
 
 type (
 	standbyActionFn     func(context.Context, execution.Context, execution.MutableState) (interface{}, error)
 	standbyPostActionFn func(context.Context, persistence.Task, interface{}, log.Logger) error
 
 	standbyCurrentTimeFn func(persistence.Task) (time.Time, error)
+
+	// TaskDLQWriter is the subset of taskdlq.HistoryTaskDLQStore used by standby task executors.
+	TaskDLQWriter interface {
+		AddTask(ctx context.Context, request taskdlq.AddTaskRequest) error
+	}
 )
 
 var (
@@ -91,6 +100,82 @@ func standbyTaskPostActionTaskDiscarded(
 	return ErrTaskDiscarded
 }
 
+// standbyTaskPostActionWriteToDLQ returns a standbyPostActionFn that writes the task to the DLQ.
+// If writer is nil, it falls back to standbyTaskPostActionTaskDiscarded (preserving the old discard
+// behavior until the DLQ persistence backend is wired up).
+func standbyTaskPostActionWriteToDLQ(
+	writer TaskDLQWriter,
+	shard shard.Context,
+	enabled dynamicproperties.StringPropertyFnWithDomainFilter,
+) standbyPostActionFn {
+	if writer == nil {
+		return standbyTaskPostActionTaskDiscarded
+	}
+
+	shardID := shard.GetShardID()
+
+	return func(ctx context.Context, task persistence.Task, postActionInfo interface{}, logger log.Logger) error {
+		domainID := task.GetDomainID()
+		isDeadLetterQueueEnabled := enabled(domainID)
+
+		if postActionInfo == nil {
+			return nil
+		}
+
+		clusterAttribute, err := getClusterAttributesForTask(ctx, shard, task)
+		if err != nil {
+			if errors.Is(err, errActiveClusterSelectionPolicyNotFound) {
+				logger.Warn("Active cluster selection policy not found. Defaulting to default scope and name.")
+			} else {
+				return err
+			}
+		}
+
+		if clusterAttribute == nil {
+			clusterAttribute = &types.ClusterAttribute{
+				Scope: taskdlq.DefaultClusterAttributeScope,
+				Name:  taskdlq.DefaultClusterAttributeName,
+			}
+		}
+
+		logger.Warn("Attempting to write standby task to DLQ due to task being pending for too long.",
+			tag.WorkflowID(task.GetWorkflowID()),
+			tag.WorkflowRunID(task.GetRunID()),
+			tag.WorkflowDomainID(task.GetDomainID()),
+			tag.TaskID(task.GetTaskID()),
+			tag.TaskType(task.GetTaskType()),
+			tag.FailoverVersion(task.GetVersion()),
+			tag.Timestamp(task.GetVisibilityTimestamp()),
+			tag.IsShadowModeEnabled(isDeadLetterQueueEnabled == constants.HistoryTaskDLQModeShadow))
+
+		// TODO(c-warren): Move this logic into the writer instead, and return a ErrHistoryDLQNotEnabled error to be handled here
+		switch isDeadLetterQueueEnabled {
+		case constants.HistoryTaskDLQModeEnabled:
+			return writer.AddTask(ctx, taskdlq.AddTaskRequest{
+				ShardID:               shardID,
+				DomainID:              task.GetDomainID(),
+				ClusterAttributeScope: clusterAttribute.Scope,
+				ClusterAttributeName:  clusterAttribute.Name,
+				Task:                  task,
+			})
+		case constants.HistoryTaskDLQModeShadow:
+			err := writer.AddTask(ctx, taskdlq.AddTaskRequest{
+				ShardID:               shardID,
+				DomainID:              task.GetDomainID(),
+				ClusterAttributeScope: clusterAttribute.Scope,
+				ClusterAttributeName:  clusterAttribute.Name,
+				Task:                  task,
+			})
+			if err != nil {
+				logger.Warn("Failed to write standby task to DLQ in shadow mode. Will discard the task.")
+			}
+			return standbyTaskPostActionTaskDiscarded(ctx, task, postActionInfo, logger)
+		default:
+			return standbyTaskPostActionTaskDiscarded(ctx, task, postActionInfo, logger)
+		}
+	}
+}
+
 type (
 	historyResendInfo struct {
 		// used by NDC