feat(domain multi-tenancy): Store normalized task list name in history tasks

Author: Shaddoll

common/persistence/tasks.go

@@ -40,7 +40,13 @@ type Task interface {
 	GetDomainID() string
 	GetWorkflowID() string
 	GetRunID() string
+	// GetTaskList returns the name of the task list the task is currently
+    // associated with. This may differ from the original task list if the
+    // task is a sticky decision task.
 	GetTaskList() string
+	// GetOriginalTaskList returns the task list on which the task was initially
+	// scheduled. It is used to enforce rate limits and ensure fair scheduling
+	// across task lists.
 	GetOriginalTaskList() string
 	GetVersion() int64
 	SetVersion(version int64)

service/history/execution/mutable_state_task_generator.go

@@ -145,6 +145,7 @@ func (r *mutableStateTaskGeneratorImpl) GenerateWorkflowStartTasks(
 			VisibilityTimestamp: workflowTimeoutTimestamp,
 			Version:             startVersion,
 		},
+		TaskList: getNormalizedTaskListName(executionInfo.TaskList, executionInfo.TaskListKind),
 	})
 
 	return nil
@@ -156,6 +157,7 @@ func (r *mutableStateTaskGeneratorImpl) GenerateWorkflowCloseTasks(
 ) error {
 
 	executionInfo := r.mutableState.GetExecutionInfo()
+	taskList := getNormalizedTaskListName(executionInfo.TaskList, executionInfo.TaskListKind)
 	r.mutableState.AddTransferTasks(&persistence.CloseExecutionTask{
 		WorkflowIdentifier: persistence.WorkflowIdentifier{
 			DomainID:   executionInfo.DomainID,
@@ -166,6 +168,7 @@ func (r *mutableStateTaskGeneratorImpl) GenerateWorkflowCloseTasks(
 			// TaskID and VisibilityTimestamp are set by shard context
 			Version: closeEvent.Version,
 		},
+		TaskList: taskList,
 	})
 
 	retentionInDays := defaultWorkflowRetentionInDays
@@ -202,6 +205,7 @@ func (r *mutableStateTaskGeneratorImpl) GenerateWorkflowCloseTasks(
 			VisibilityTimestamp: closeTimestamp.Add(retentionDuration),
 			Version:             closeEvent.Version,
 		},
+		TaskList: taskList,
 	})
 
 	return nil
@@ -250,6 +254,7 @@ func (r *mutableStateTaskGeneratorImpl) GenerateDelayedDecisionTasks(
 			Version:             startVersion,
 		},
 		TimeoutType: firstDecisionDelayType,
+		TaskList:    getNormalizedTaskListName(executionInfo.TaskList, executionInfo.TaskListKind),
 	})
 
 	return nil
@@ -272,6 +277,7 @@ func (r *mutableStateTaskGeneratorImpl) GenerateRecordWorkflowStartedTasks(
 			// TaskID and VisibilityTimestamp are set by shard context
 			Version: startVersion,
 		},
+		TaskList: getNormalizedTaskListName(executionInfo.TaskList, executionInfo.TaskListKind),
 	})
 
 	return nil
@@ -291,6 +297,7 @@ func (r *mutableStateTaskGeneratorImpl) GenerateDecisionScheduleTasks(
 		}
 	}
 
+	originalTaskList := getNormalizedTaskListName(executionInfo.TaskList, executionInfo.TaskListKind)
 	r.mutableState.AddTransferTasks(&persistence.DecisionTask{
 		WorkflowIdentifier: persistence.WorkflowIdentifier{
 			DomainID:   executionInfo.DomainID,
@@ -301,9 +308,10 @@ func (r *mutableStateTaskGeneratorImpl) GenerateDecisionScheduleTasks(
 			// TaskID and VisibilityTimestamp are set by shard context
 			Version: decision.Version,
 		},
-		TargetDomainID: executionInfo.DomainID,
-		TaskList:       decision.TaskList,
-		ScheduleID:     decision.ScheduleID,
+		TargetDomainID:   executionInfo.DomainID,
+		TaskList:         decision.TaskList,
+		ScheduleID:       decision.ScheduleID,
+		OriginalTaskList: originalTaskList,
 	})
 
 	if scheduleToStartTimeout := r.mutableState.GetDecisionScheduleToStartTimeout(); scheduleToStartTimeout != 0 {
@@ -322,6 +330,7 @@ func (r *mutableStateTaskGeneratorImpl) GenerateDecisionScheduleTasks(
 			TimeoutType:     int(TimerTypeScheduleToStart),
 			EventID:         decision.ScheduleID,
 			ScheduleAttempt: decision.Attempt,
+			TaskList:        originalTaskList,
 		})
 	}
 
@@ -369,6 +378,7 @@ func (r *mutableStateTaskGeneratorImpl) GenerateDecisionStartTasks(
 		TimeoutType:     int(TimerTypeStartToClose),
 		EventID:         decision.ScheduleID,
 		ScheduleAttempt: decision.Attempt,
+		TaskList:        getNormalizedTaskListName(executionInfo.TaskList, executionInfo.TaskListKind),
 	})
 
 	return nil
@@ -445,8 +455,9 @@ func (r *mutableStateTaskGeneratorImpl) GenerateActivityRetryTasks(
 			Version:             ai.Version,
 			VisibilityTimestamp: ai.ScheduledTime,
 		},
-		EventID: ai.ScheduleID,
-		Attempt: int64(ai.Attempt),
+		EventID:  ai.ScheduleID,
+		Attempt:  int64(ai.Attempt),
+		TaskList: getNormalizedTaskListName(ai.TaskList, ai.TaskListKind),
 	})
 	return nil
 }
@@ -490,6 +501,7 @@ func (r *mutableStateTaskGeneratorImpl) GenerateChildWorkflowTasks(
 		TargetDomainID:   targetDomainID,
 		TargetWorkflowID: childWorkflowInfo.StartedWorkflowID,
 		InitiatedID:      childWorkflowInfo.InitiatedID,
+		TaskList:         getNormalizedTaskListName(executionInfo.TaskList, executionInfo.TaskListKind),
 	}
 
 	r.mutableState.AddTransferTasks(startChildExecutionTask)
@@ -536,6 +548,7 @@ func (r *mutableStateTaskGeneratorImpl) GenerateRequestCancelExternalTasks(
 		TargetRunID:             targetRunID,
 		TargetChildWorkflowOnly: targetChildOnly,
 		InitiatedID:             scheduleID,
+		TaskList:                getNormalizedTaskListName(executionInfo.TaskList, executionInfo.TaskListKind),
 	}
 
 	r.mutableState.AddTransferTasks(cancelExecutionTask)
@@ -583,6 +596,7 @@ func (r *mutableStateTaskGeneratorImpl) GenerateSignalExternalTasks(
 		TargetRunID:             targetRunID,
 		TargetChildWorkflowOnly: targetChildOnly,
 		InitiatedID:             scheduleID,
+		TaskList:                getNormalizedTaskListName(executionInfo.TaskList, executionInfo.TaskListKind),
 	}
 
 	r.mutableState.AddTransferTasks(signalExecutionTask)
@@ -605,6 +619,7 @@ func (r *mutableStateTaskGeneratorImpl) GenerateWorkflowSearchAttrTasks() error
 			// TaskID and VisibilityTimestamp are set by shard context
 			Version: currentVersion, // task processing does not check this version
 		},
+		TaskList: getNormalizedTaskListName(executionInfo.TaskList, executionInfo.TaskListKind),
 	})
 
 	return nil
@@ -625,6 +640,7 @@ func (r *mutableStateTaskGeneratorImpl) GenerateWorkflowResetTasks() error {
 			// TaskID and VisibilityTimestamp are set by shard context
 			Version: currentVersion,
 		},
+		TaskList: getNormalizedTaskListName(executionInfo.TaskList, executionInfo.TaskListKind),
 	})
 
 	return nil
@@ -681,6 +697,13 @@ func (r *mutableStateTaskGeneratorImpl) validateChildWorkflowParameters(msbDomai
 	return nil
 }
 
+func getNormalizedTaskListName(name string, kind types.TaskListKind) string {
+	if kind == types.TaskListKindEphemeral {
+		return "__ephemeral__"
+	}
+	return name
+}
+
 func getNextDecisionTimeout(attempt int64, defaultStartToCloseTimeout time.Duration) time.Duration {
 	if attempt <= 1 {
 		return defaultStartToCloseTimeout