Remove ExecutableWrapper (#5033)

<!-- Describe what has changed in this PR -->
**What changed?**
I discovered an issue with the `ExecutableWrapper`
[here](https://github.com/temporalio/temporal/blob/df4705d6488485ae12e27f4cb1c719b62c980304/service/history/queues/executable.go#L451).
Because we don't wrap the `Reschedule` method, it adds the unwrapped
version to the rescheduler, which loses all of the wrapped behavior. I
implemented a fix using the "delegate" pattern, where the base instance
holds a reference to the wrapped instance, but it is fraught with peril:
you need to remember to use the delegate everywhere it's applicable, and
you need to remember to set it--also only one delegate can be set, so
you can't have different wrappers for the same object. As a result, I
decided to get rid of the wrapper-based approach, and instead added the
code directly to the base executable implementation.

<!-- Tell your future self why have you made these changes -->
**Why?**
The huge refactoring is necessary because the `ExecutableWrapper`
approach is too complicated and risky.

<!-- How have you verified this change? Tested locally? Added a unit
test? Checked in staging env? -->
**How did you test it?**
There's still 100% test coverage for the DLQ code. Most imporantly,
there is now an end-to-end integration test which starts a workflow that
produces terminally failing tasks, verifies that it's added to the DLQ,
and then uses the tdbg read command to delete the DLQ tasks. This wasn't
possible before due to the bug in the `ExecutableWrapper`.

<!-- Assuming the worst case, what can be broken when deploying this
change to production? -->
**Potential risks**


<!-- Is this PR a hotfix candidate or require that a notification be
sent to the broader community? (Yes/No) -->
**Is hotfix candidate?**

common/persistence/data_interfaces.go

@@ -1269,7 +1269,7 @@ type (
 
 	RawHistoryTask struct {
 		MessageMetadata MessageMetadata
-		Task            *persistencespb.HistoryTask
+		Payload         *persistencespb.HistoryTask
 	}
 
 	ReadRawTasksResponse struct {

common/persistence/history_task_queue_manager.go

@@ -151,7 +151,7 @@ func (m *HistoryTaskQueueManagerImpl) ReadRawTasks(
 			return nil, fmt.Errorf("%v: %w", ErrMsgDeserializeRawHistoryTask, err)
 		}
 		responseTasks[i].MessageMetadata = message.MetaData
-		responseTasks[i].Task = &task
+		responseTasks[i].Payload = &task
 	}
 
 	return &ReadRawTasksResponse{
@@ -170,7 +170,7 @@ func (m *HistoryTaskQueueManagerImpl) ReadTasks(ctx context.Context, request *Re
 	resTasks := make([]HistoryTask, len(response.Tasks))
 
 	for i, rawTask := range response.Tasks {
-		blob := rawTask.Task.Blob
+		blob := rawTask.Payload.Blob
 		if blob == nil {
 			return nil, serialization.NewDeserializationError(enums.ENCODING_TYPE_PROTO3, ErrHistoryTaskBlobIsNil)
 		}

common/persistence/tests/history_task_queue_manager_test_suite.go

@@ -42,7 +42,6 @@ import (
 	"go.temporal.io/server/common/persistence/persistencetest"
 	"go.temporal.io/server/service/history/api/deletedlqtasks/deletedlqtaskstest"
 	"go.temporal.io/server/service/history/api/getdlqtasks/getdlqtaskstest"
-	"go.temporal.io/server/service/history/queues/queuestest"
 	"go.temporal.io/server/service/history/tasks"
 )
 
@@ -136,10 +135,6 @@ func RunHistoryTaskQueueManagerTestSuite(t *testing.T, queue persistence.QueueV2
 		t.Parallel()
 		historytest.TestClient(t, historyTaskQueueManager)
 	})
-	t.Run("ExecutableTest", func(t *testing.T) {
-		t.Parallel()
-		queuestest.TestExecutable(t, historyTaskQueueManager)
-	})
 }
 
 func testHistoryTaskQueueManagerCreateQueueErr(t *testing.T, queue persistence.QueueV2) {

proto/internal/temporal/server/api/common/v1/dlq.proto

@@ -8,7 +8,7 @@ import "temporal/api/common/v1/message.proto";
 message HistoryTask {
   // shard_id is included to avoid having to deserialize the task blob.
   int32 shard_id = 1;
-  temporal.api.common.v1.DataBlob task = 2;
+  temporal.api.common.v1.DataBlob blob = 2;
 }
 
 message HistoryDLQTaskMetadata {
@@ -20,7 +20,8 @@ message HistoryDLQTaskMetadata {
 // queue).
 message HistoryDLQTask {
   HistoryDLQTaskMetadata metadata = 1;
-  HistoryTask task = 2;
+  // This is named payload to prevent stuttering (e.g. task.Task).
+  HistoryTask payload = 2;
 }
 
 // HistoryDLQKey is a compound key that identifies a history DLQ.

service/frontend/admin_handler.go

@@ -1808,13 +1808,13 @@ func (adh *AdminHandler) PurgeDLQTasks(
 	}
 	workflowID := fmt.Sprintf("delete-dlq-tasks-%s", key.GetQueueName())
 	client := adh.sdkClientFactory.GetSystemClient()
-	future, err := client.ExecuteWorkflow(ctx, sdkclient.StartWorkflowOptions{
+	run, err := client.ExecuteWorkflow(ctx, sdkclient.StartWorkflowOptions{
 		ID:        workflowID,
 		TaskQueue: primitives.DefaultWorkerTaskQueue,
 	}, dlq.WorkflowName, dlq.WorkflowParams{
 		WorkflowType: dlq.WorkflowTypeDelete,
 		DeleteParams: dlq.DeleteParams{
-			TaskCategory:  int(category.ID()),
+			TaskCategory:  category.ID(),
 			SourceCluster: sourceCluster,
 			TargetCluster: targetCluster,
 			MaxMessageID:  request.InclusiveMaxTaskMetadata.MessageId,
@@ -1823,9 +1823,10 @@ func (adh *AdminHandler) PurgeDLQTasks(
 	if err != nil {
 		return nil, err
 	}
+	runID := run.GetRunID()
 	jobToken := adminservice.DLQJobToken{
 		WorkflowId: workflowID,
-		RunId:      future.GetRunID(),
+		RunId:      runID,
 	}
 	jobTokenBytes, _ := jobToken.Marshal()
 	return &adminservice.PurgeDLQTasksResponse{

service/frontend/admin_handler_test.go

@@ -1173,9 +1173,9 @@ func (s *adminHandlerSuite) TestGetDLQTasks() {
 							Metadata: &commonspb.HistoryDLQTaskMetadata{
 								MessageId: 21,
 							},
-							Task: &commonspb.HistoryTask{
+							Payload: &commonspb.HistoryTask{
 								ShardId: 34,
-								Task:    blob,
+								Blob:    blob,
 							},
 						},
 					},
@@ -1202,9 +1202,9 @@ func (s *adminHandlerSuite) TestGetDLQTasks() {
 						Metadata: &commonspb.HistoryDLQTaskMetadata{
 							MessageId: 21,
 						},
-						Task: &commonspb.HistoryTask{
+						Payload: &commonspb.HistoryTask{
 							ShardId: 34,
-							Task:    blob,
+							Blob:    blob,
 						},
 					},
 				},

service/history/api/getdlqtasks/api.go

@@ -77,9 +77,9 @@ func Invoke(
 			Metadata: &commonspb.HistoryDLQTaskMetadata{
 				MessageId: task.MessageMetadata.ID,
 			},
-			Task: &commonspb.HistoryTask{
-				ShardId: task.Task.ShardId,
-				Task:    task.Task.Blob,
+			Payload: &commonspb.HistoryTask{
+				ShardId: task.Payload.ShardId,
+				Blob:    task.Payload.Blob,
 			},
 		}
 	}

service/history/api/getdlqtasks/getdlqtaskstest/apitest.go

@@ -81,9 +81,9 @@ func TestInvoke(t *testing.T, manager persistence.HistoryTaskQueueManager) {
 	require.NoError(t, err)
 	require.Equal(t, 1, len(res.DlqTasks))
 	assert.Equal(t, int64(persistence.FirstQueueMessageID), res.DlqTasks[0].Metadata.MessageId)
-	assert.Equal(t, 1, int(res.DlqTasks[0].Task.ShardId))
+	assert.Equal(t, 1, int(res.DlqTasks[0].Payload.ShardId))
 	serializer := serialization.NewTaskSerializer()
-	outTask, err := serializer.DeserializeTask(tasks.CategoryTransfer, *res.DlqTasks[0].Task.Task)
+	outTask, err := serializer.DeserializeTask(tasks.CategoryTransfer, *res.DlqTasks[0].Payload.Blob)
 	require.NoError(t, err)
 	assert.Equal(t, inTask, outTask)
 }