[Experiment] CHASM scheduler action and policy contracts

chasm/chasmtest/test_engine_test.go

@@ -19,6 +19,29 @@ import (
 	"go.temporal.io/server/service/history/tasks"
 )
 
+// TestTasksArePhysicallyGenerated: a task a component adds must reach the backend as a physical task, in
+// the category its TaskAttributes imply, whether it was added while starting the execution or while
+// updating it.
+func TestTasksArePhysicallyGenerated(t *testing.T) {
+	const ttl = time.Hour
+
+	t.Run("added while starting", func(t *testing.T) {
+		e, ref := startStore(t, ttl)
+		require.Equal(t, 1, countTasks(t, e, ref, tasks.CategoryTimer))
+	})
+
+	t.Run("added while updating", func(t *testing.T) {
+		e, ref := startStore(t, 0)
+		require.Equal(t, 0, countTasks(t, e, ref, tasks.CategoryTimer))
+		_, _, err := chasm.UpdateComponent(engineContext(e), ref,
+			func(s *tests.PayloadStore, mc chasm.MutableContext, _ any) (any, error) {
+				return nil, addPayload(s, mc, "second", ttl)
+			}, nil)
+		require.NoError(t, err)
+		require.Equal(t, 1, countTasks(t, e, ref, tasks.CategoryTimer))
+	})
+}
+
 func TestInvariantCheckRunsAfterSuccessfulTransitions(t *testing.T) {
 	var calls []string
 	check := func(name string) chasmtest.InvariantCheck {
@@ -31,27 +54,35 @@ func TestInvariantCheckRunsAfterSuccessfulTransitions(t *testing.T) {
 		}
 	}
 
-	e, ref := startStoreWithEngineOptions(t, time.Hour, []chasmtest.EngineOption{
-		chasmtest.WithInvariantCheck(check("first")),
-		chasmtest.WithInvariantCheck(check("second")),
-	})
+	e, ref := startStoreWithEngineOptions(
+		t,
+		time.Hour,
+		[]chasmtest.EngineOption{
+			chasmtest.WithInvariantCheck(check("first")),
+			chasmtest.WithInvariantCheck(check("second")),
+		},
+	)
 	require.Equal(t, []string{"first", "second"}, calls)
 
 	_, _, err := chasm.UpdateComponent(engineContext(e), ref,
-		func(*tests.PayloadStore, chasm.MutableContext, any) (any, error) { return nil, nil }, nil,
-		chasm.WithRequestID("invariant-check"))
+		func(*tests.PayloadStore, chasm.MutableContext, any) (any, error) {
+			return nil, nil
+		}, nil, chasm.WithRequestID("invariant-check"))
 	require.NoError(t, err)
 	require.Equal(t, []string{"first", "second", "first", "second"}, calls)
 
 	_, _, err = chasm.UpdateComponent(engineContext(e), ref,
-		func(*tests.PayloadStore, chasm.MutableContext, any) (any, error) { return nil, nil }, nil,
-		chasm.WithRequestID("invariant-check"))
+		func(*tests.PayloadStore, chasm.MutableContext, any) (any, error) {
+			return nil, nil
+		}, nil, chasm.WithRequestID("invariant-check"))
 	require.ErrorIs(t, err, chasm.ErrRequestIDAlreadyUsed)
 	require.Equal(t, []string{"first", "second", "first", "second"}, calls)
 
 	transitionErr := errors.New("transition failed")
 	_, _, err = chasm.UpdateComponent(engineContext(e), ref,
-		func(*tests.PayloadStore, chasm.MutableContext, any) (any, error) { return nil, transitionErr }, nil)
+		func(*tests.PayloadStore, chasm.MutableContext, any) (any, error) {
+			return nil, transitionErr
+		}, nil)
 	require.ErrorIs(t, err, transitionErr)
 	require.Equal(t, []string{"first", "second", "first", "second"}, calls)
 
@@ -61,39 +92,27 @@ func TestInvariantCheckRunsAfterSuccessfulTransitions(t *testing.T) {
 	require.Equal(t, []string{"first", "second", "first", "second", "first", "second"}, calls)
 
 	key := chasm.ExecutionKey{NamespaceID: "test-ns", BusinessID: "update-with-start"}
-	result, err := chasm.UpdateWithStartExecution(engineContext(e), key,
-		func(mc chasm.MutableContext, _ any) (*tests.PayloadStore, error) { return tests.NewPayloadStore(mc) },
-		func(*tests.PayloadStore, chasm.MutableContext, any) (any, error) { return nil, nil }, nil)
+	result, err := chasm.UpdateWithStartExecution(
+		engineContext(e),
+		key,
+		func(mc chasm.MutableContext, _ any) (*tests.PayloadStore, error) {
+			return tests.NewPayloadStore(mc)
+		},
+		func(*tests.PayloadStore, chasm.MutableContext, any) (any, error) {
+			return nil, nil
+		},
+		nil,
+	)
 	require.NoError(t, err)
 	require.True(t, result.Created)
 	require.Equal(t, []string{
-		"first", "second", "first", "second", "first", "second", "first", "second",
+		"first", "second",
+		"first", "second",
+		"first", "second",
+		"first", "second",
 	}, calls)
 }
 
-// TestTasksArePhysicallyGenerated: a task a component adds must reach the backend as a physical task, in
-// the category its TaskAttributes imply, whether it was added while starting the execution or while
-// updating it.
-func TestTasksArePhysicallyGenerated(t *testing.T) {
-	const ttl = time.Hour
-
-	t.Run("added while starting", func(t *testing.T) {
-		e, ref := startStore(t, ttl)
-		require.Equal(t, 1, countTasks(t, e, ref, tasks.CategoryTimer))
-	})
-
-	t.Run("added while updating", func(t *testing.T) {
-		e, ref := startStore(t, 0)
-		require.Equal(t, 0, countTasks(t, e, ref, tasks.CategoryTimer))
-		_, _, err := chasm.UpdateComponent(engineContext(e), ref,
-			func(s *tests.PayloadStore, mc chasm.MutableContext, _ any) (any, error) {
-				return nil, addPayload(s, mc, "second", ttl)
-			}, nil)
-		require.NoError(t, err)
-		require.Equal(t, 1, countTasks(t, e, ref, tasks.CategoryTimer))
-	})
-}
-
 func TestFireSideEffectTasks(t *testing.T) {
 	const ttl = time.Hour
 	e, ref := startStore(t, ttl)
@@ -150,6 +169,7 @@ func TestUpdateComponentDeduplicatesRequestID(t *testing.T) {
 	updatedRef, err = update()
 	var failedPrecondition *serviceerror.FailedPrecondition
 	require.ErrorAs(t, err, &failedPrecondition)
+	require.ErrorIs(t, err, chasm.ErrRequestIDAlreadyUsed)
 	require.Nil(t, updatedRef)
 	require.Equal(t, 1, updateCount)
 }

chasm/lib/scheduler/action.go

@@ -0,0 +1,83 @@
+package scheduler
+
+import (
+	"context"
+	"time"
+
+	commonpb "go.temporal.io/api/common/v1"
+	enumspb "go.temporal.io/api/enums/v1"
+	schedulepb "go.temporal.io/api/schedule/v1"
+	"go.temporal.io/api/serviceerror"
+	"go.temporal.io/api/workflowservice/v1"
+	persistencespb "go.temporal.io/server/api/persistence/v1"
+	schedulespb "go.temporal.io/server/api/schedule/v1"
+	"go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1"
+	"go.temporal.io/server/chasm/lib/scheduler/internal"
+	"go.temporal.io/server/common/resource"
+)
+
+type occurrenceContext struct {
+	NominalTime, ScheduledTime time.Time
+	Manual                     bool
+	RequestID                  string
+}
+
+type actionMetadata struct {
+	Kind                    enumspb.ExecutionType
+	Type, IDBase, TaskQueue string
+	SearchAttributes        *commonpb.SearchAttributes
+}
+
+type actionStartInput struct {
+	Scheduler                *Scheduler
+	Occurrence               *schedulespb.BufferedStart
+	Callback                 *commonpb.Callback
+	Previous                 *schedulerpb.LastCompletionResult
+	EnableVersioningOverride bool
+}
+
+type actionClients struct {
+	Frontend workflowservice.WorkflowServiceClient
+	History  resource.HistoryClient
+}
+
+type actionCompletion struct {
+	Result *commonpb.ActionExecutionResult
+	Failed bool
+}
+
+type actionImplementation interface {
+	Metadata(*schedulepb.ScheduleAction) actionMetadata
+	Validate(*schedulepb.ScheduleAction) error
+	Policies() *internal.PolicyRegistry
+	GenerateTargetID(string, occurrenceContext) string
+	Start(context.Context, actionClients, actionStartInput) (string, error)
+	Cancel(context.Context, actionClients, *Scheduler, *commonpb.Execution) error
+	Terminate(context.Context, actionClients, *Scheduler, *commonpb.Execution) error
+	Completion(*persistencespb.ChasmNexusCompletion, *commonpb.Execution) actionCompletion
+	ParticipatesInCompletionHistory() bool
+}
+
+func implementation(_ *schedulepb.ScheduleAction) actionImplementation { return workflowAction{} }
+
+func (s *Scheduler) actionMetadata() actionMetadata {
+	return implementation(s.Schedule.GetAction()).Metadata(s.Schedule.GetAction())
+}
+
+func (s *Scheduler) newBufferedExecution(start *schedulespb.BufferedStart, base string) {
+	action := implementation(s.Schedule.GetAction())
+	id := action.GenerateTargetID(base, occurrenceContext{NominalTime: start.GetNominalTime().AsTime(), ScheduledTime: start.GetActualTime().AsTime(), Manual: start.GetManual(), RequestID: start.GetRequestId()})
+	start.Execution = &commonpb.Execution{Type: action.Metadata(s.Schedule.GetAction()).Kind, BusinessId: id}
+	if start.Execution.Type == enumspb.EXECUTION_TYPE_WORKFLOW {
+		start.WorkflowId = id
+	}
+}
+
+func (s *Scheduler) targetIDBase() string { return s.actionMetadata().IDBase }
+
+func validateActionKind(previous, next *schedulepb.ScheduleAction) error {
+	if implementation(previous).Metadata(previous).Kind != implementation(next).Metadata(next).Kind {
+		return serviceerror.NewInvalidArgument("schedule action kind cannot be changed")
+	}
+	return nil
+}

chasm/lib/scheduler/action_policy.go

@@ -0,0 +1,54 @@
+package scheduler
+
+import (
+	enumspb "go.temporal.io/api/enums/v1"
+	schedulepb "go.temporal.io/api/schedule/v1"
+	"go.temporal.io/api/serviceerror"
+	"go.temporal.io/server/chasm/lib/scheduler/internal"
+)
+
+func policySelection(builtin enumspb.ScheduleOverlapPolicy, custom *schedulepb.CustomOverlapPolicy) internal.PolicyIdentity {
+	return internal.PolicyIdentity{Builtin: builtin, Custom: custom.GetName()}
+}
+
+func actionPolicies(action *schedulepb.ScheduleAction) *internal.PolicyRegistry {
+	return implementation(action).Policies()
+}
+
+// ValidateScheduleActionPolicies validates action-specific selections independently of legacy validation settings.
+func ValidateScheduleActionPolicies(schedule *schedulepb.Schedule, patch *schedulepb.SchedulePatch) error {
+	return ValidateActionPolicyOverrides(schedule.GetAction(), schedule.GetPolicies(), patch)
+}
+
+func ValidateActionPolicyOverrides(action *schedulepb.ScheduleAction, policies *schedulepb.SchedulePolicies, patch *schedulepb.SchedulePatch) error {
+	registry := actionPolicies(action)
+	configured := policySelection(policies.GetOverlapPolicy(), policies.GetCustomOverlapPolicy())
+	if policies.GetCustomOverlapPolicy() != nil && (configured.Custom == "" || configured.Builtin != 0) {
+		return serviceerror.NewInvalidArgument("custom overlap policy requires a name and cannot be combined with overlap_policy")
+	}
+	if _, err := registry.Resolve(internal.PolicyIdentity{}, configured); err != nil {
+		return err
+	}
+	validate := func(builtin enumspb.ScheduleOverlapPolicy, custom *schedulepb.CustomOverlapPolicy) error {
+		if custom != nil && (custom.GetName() == "" || builtin != 0) {
+			return serviceerror.NewInvalidArgument("custom overlap policy requires a name and cannot be combined with overlap_policy")
+		}
+		_, err := registry.Resolve(policySelection(builtin, custom), configured)
+		return err
+	}
+	if trigger := patch.GetTriggerImmediately(); trigger != nil {
+		if err := validate(trigger.GetOverlapPolicy(), trigger.GetCustomOverlapPolicy()); err != nil {
+			return err
+		}
+	}
+	for _, backfill := range patch.GetBackfillRequest() {
+		if err := validate(backfill.GetOverlapPolicy(), backfill.GetCustomOverlapPolicy()); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (s *Scheduler) resolvedPolicy(builtin enumspb.ScheduleOverlapPolicy, custom *schedulepb.CustomOverlapPolicy) (internal.PolicyIdentity, error) {
+	return actionPolicies(s.Schedule.GetAction()).Resolve(policySelection(builtin, custom), policySelection(s.Schedule.GetPolicies().GetOverlapPolicy(), s.Schedule.GetPolicies().GetCustomOverlapPolicy()))
+}

chasm/lib/scheduler/action_results.go

@@ -0,0 +1,19 @@
+package scheduler
+
+import (
+	commonpb "go.temporal.io/api/common/v1"
+	enumspb "go.temporal.io/api/enums/v1"
+	schedulepb "go.temporal.io/api/schedule/v1"
+	schedulespb "go.temporal.io/server/api/schedule/v1"
+	"go.temporal.io/server/chasm/lib/scheduler/internal"
+	"go.temporal.io/server/common"
+)
+
+func actionResult(start *schedulespb.BufferedStart) *schedulepb.ScheduleActionResult {
+	result := &schedulepb.ScheduleActionResult{ScheduleTime: start.GetActualTime(), ActualTime: start.GetStartTime(), ActionExecutionResult: common.CloneProto(internal.ExecutionResult(start)), CloseTime: internal.CompletionTime(start)}
+	if result.ActionExecutionResult.Execution.GetType() == enumspb.EXECUTION_TYPE_WORKFLOW {
+		result.StartWorkflowResult = &commonpb.WorkflowExecution{WorkflowId: internal.TargetID(start), RunId: internal.RunID(start)}
+		result.StartWorkflowStatus = result.ActionExecutionResult.GetWorkflowStatus()
+	}
+	return result
+}