Add support for Ephemeral TaskLists

Update Sessions to use Ephemeral TaskLists behind a feature flag. This ensures that the per-host TaskList is automatically removed once it is no longer used.

This should only be enabled once the server fully supports Ephemeral TaskLists as it will otherwise return errors for the unknown TaskListKind.

Signed-off-by: Nathanael Mortensen <[email protected]>

Author: natemort

internal/activity_task_handler.go

@@ -80,7 +80,7 @@ func newActivityTaskHandlerWithCustomProvider(
 	}
 	return &activityTaskHandlerImpl{
 		clock:              clock,
-		taskListName:       params.TaskList,
+		taskListName:       params.TaskList.GetName(),
 		identity:           params.Identity,
 		service:            service,
 		logger:             params.Logger,

internal/client.go

@@ -656,6 +656,7 @@ func getFeatureFlags(options *ClientOptions) FeatureFlags {
 		return FeatureFlags{
 			WorkflowExecutionAlreadyCompletedErrorEnabled: options.FeatureFlags.WorkflowExecutionAlreadyCompletedErrorEnabled,
 			PollerAutoScalerEnabled:                       options.FeatureFlags.PollerAutoScalerEnabled,
+			EphemeralTaskListsEnabled:                     options.FeatureFlags.EphemeralTaskListsEnabled,
 		}
 	}
 	return FeatureFlags{}

internal/compatibility/enum_test.go

@@ -266,6 +266,7 @@ func TestTaskListKind(t *testing.T) {
 		apiv1.TaskListKind_TASK_LIST_KIND_INVALID,
 		apiv1.TaskListKind_TASK_LIST_KIND_NORMAL,
 		apiv1.TaskListKind_TASK_LIST_KIND_STICKY,
+		apiv1.TaskListKind_TASK_LIST_KIND_EPHEMERAL,
 	} {
 		assert.Equal(t, item, proto.TaskListKind(thrift.TaskListKind(item)))
 	}

internal/compatibility/proto/enum.go

@@ -35,6 +35,8 @@ func TaskListKind(t *shared.TaskListKind) apiv1.TaskListKind {
 		return apiv1.TaskListKind_TASK_LIST_KIND_NORMAL
 	case shared.TaskListKindSticky:
 		return apiv1.TaskListKind_TASK_LIST_KIND_STICKY
+	case shared.TaskListKindEphemeral:
+		return apiv1.TaskListKind_TASK_LIST_KIND_EPHEMERAL
 	}
 	panic("unexpected enum value")
 }

internal/compatibility/thrift/enum.go

@@ -34,6 +34,8 @@ func TaskListKind(t apiv1.TaskListKind) *shared.TaskListKind {
 		return shared.TaskListKindNormal.Ptr()
 	case apiv1.TaskListKind_TASK_LIST_KIND_STICKY:
 		return shared.TaskListKindSticky.Ptr()
+	case apiv1.TaskListKind_TASK_LIST_KIND_EPHEMERAL:
+		return shared.TaskListKindEphemeral.Ptr()
 	}
 	panic("unexpected enum value")
 }

internal/internal_activity.go

@@ -59,6 +59,7 @@ type (
 	activityOptions struct {
 		ActivityID                    *string // Users can choose IDs but our framework makes it optional to decrease the crust.
 		TaskListName                  string
+		TaskListKind                  shared.TaskListKind
 		ScheduleToCloseTimeoutSeconds int32
 		ScheduleToStartTimeoutSeconds int32
 		StartToCloseTimeoutSeconds    int32

internal/internal_event_handlers.go

@@ -121,6 +121,7 @@ type (
 		contextPropagators           []ContextPropagator
 		tracer                       opentracing.Tracer
 		workflowInterceptorFactories []WorkflowInterceptorFactory
+		featureFlags                 FeatureFlags
 	}
 
 	localActivityTask struct {
@@ -205,6 +206,7 @@ func newWorkflowExecutionEventHandler(
 	contextPropagators []ContextPropagator,
 	tracer opentracing.Tracer,
 	workflowInterceptorFactories []WorkflowInterceptorFactory,
+	featureFlags FeatureFlags,
 ) workflowExecutionEventHandler {
 	context := &workflowEnvironmentImpl{
 		workflowInfo:                 workflowInfo,
@@ -222,6 +224,7 @@ func newWorkflowExecutionEventHandler(
 		contextPropagators:           contextPropagators,
 		tracer:                       tracer,
 		workflowInterceptorFactories: workflowInterceptorFactories,
+		featureFlags:                 featureFlags,
 	}
 	context.logger = logger.With(
 		zapcore.Field{Key: tagWorkflowType, Type: zapcore.StringType, String: workflowInfo.WorkflowType.Name},
@@ -472,7 +475,7 @@ func (wc *workflowEnvironmentImpl) ExecuteActivity(parameters executeActivityPar
 	}
 	activityID := scheduleTaskAttr.GetActivityId()
 	scheduleTaskAttr.ActivityType = activityTypePtr(parameters.ActivityType)
-	scheduleTaskAttr.TaskList = common.TaskListPtr(m.TaskList{Name: common.StringPtr(parameters.TaskListName)})
+	scheduleTaskAttr.TaskList = common.TaskListPtr(m.TaskList{Name: common.StringPtr(parameters.TaskListName), Kind: parameters.TaskListKind.Ptr()})
 	scheduleTaskAttr.Input = parameters.Input
 	scheduleTaskAttr.ScheduleToCloseTimeoutSeconds = common.Int32Ptr(parameters.ScheduleToCloseTimeoutSeconds)
 	scheduleTaskAttr.StartToCloseTimeoutSeconds = common.Int32Ptr(parameters.StartToCloseTimeoutSeconds)
@@ -804,6 +807,10 @@ func (wc *workflowEnvironmentImpl) GetWorkflowInterceptors() []WorkflowIntercept
 	return wc.workflowInterceptorFactories
 }
 
+func (wc *workflowEnvironmentImpl) GetFeatureFlags() FeatureFlags {
+	return wc.featureFlags
+}
+
 func (weh *workflowExecutionEventHandlerImpl) ProcessEvent(
 	event *m.HistoryEvent,
 	isReplay bool,

internal/internal_event_handlers_test.go

@@ -1065,6 +1065,7 @@ func testWorkflowExecutionEventHandler(t *testing.T, registry *registry) *workfl
 		nil,
 		opentracing.NoopTracer{},
 		nil,
+		FeatureFlags{},
 	).(*workflowExecutionEventHandlerImpl)
 }
 

internal/internal_task_handlers.go

@@ -145,6 +145,7 @@ type (
 		tracer                         opentracing.Tracer
 		workflowInterceptorFactories   []WorkflowInterceptorFactory
 		disableStrictNonDeterminism    bool
+		featureFlags                   FeatureFlags
 	}
 
 	activityProvider func(name string) activity
@@ -420,6 +421,7 @@ func newWorkflowTaskHandler(
 		tracer:                         params.Tracer,
 		workflowInterceptorFactories:   params.WorkflowInterceptorChainFactories,
 		disableStrictNonDeterminism:    params.WorkerBugPorts.DisableStrictNonDeterminismCheck,
+		featureFlags:                   params.FeatureFlags,
 	}
 
 	traceLog(func() {
@@ -623,6 +625,7 @@ func (w *workflowExecutionContextImpl) createEventHandler() {
 		w.wth.contextPropagators,
 		w.wth.tracer,
 		w.wth.workflowInterceptorFactories,
+		w.wth.featureFlags,
 	)
 	w.eventHandler.Store(eventHandler)
 }