diff --git a/chasm/lib/scheduler/action.go b/chasm/lib/scheduler/action.go index 36606727f3e..29cd6d5bb15 100644 --- a/chasm/lib/scheduler/action.go +++ b/chasm/lib/scheduler/action.go @@ -58,7 +58,12 @@ type actionImplementation interface { ParticipatesInCompletionHistory() bool } -func implementation(_ *schedulepb.ScheduleAction) actionImplementation { return workflowAction{} } +func implementation(action *schedulepb.ScheduleAction) actionImplementation { + if action.GetStartActivity() != nil { + return activityAction{} + } + return workflowAction{} +} func (s *Scheduler) actionMetadata() actionMetadata { return implementation(s.Schedule.GetAction()).Metadata(s.Schedule.GetAction()) diff --git a/chasm/lib/scheduler/action_activity.go b/chasm/lib/scheduler/action_activity.go new file mode 100644 index 00000000000..ffdedf4cea2 --- /dev/null +++ b/chasm/lib/scheduler/action_activity.go @@ -0,0 +1,84 @@ +package scheduler + +import ( + "context" + + commonpb "go.temporal.io/api/common/v1" + enumspb "go.temporal.io/api/enums/v1" + failurepb "go.temporal.io/api/failure/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" + "go.temporal.io/server/chasm/lib/scheduler/internal" +) + +type activityAction struct{} + +func (activityAction) Metadata(action *schedulepb.ScheduleAction) actionMetadata { + spec := action.GetStartActivity() + return actionMetadata{Kind: enumspb.EXECUTION_TYPE_ACTIVITY, Type: spec.GetActivityType().GetName(), IDBase: spec.GetActivityId(), TaskQueue: spec.GetTaskQueue().GetName(), SearchAttributes: spec.GetSearchAttributes()} +} +func (activityAction) Validate(action *schedulepb.ScheduleAction) error { + spec := action.GetStartActivity() + if spec == nil || spec.GetActivityId() == "" || spec.GetActivityType().GetName() == "" || spec.GetTaskQueue().GetName() == "" { + return serviceerror.NewInvalidArgument("scheduled activity requires ID, type, and task queue") + } + if spec.GetScheduleToCloseTimeout().AsDuration() <= 0 && spec.GetStartToCloseTimeout().AsDuration() <= 0 { + return serviceerror.NewInvalidArgument("scheduled activity requires a positive schedule-to-close or start-to-close timeout") + } + return nil +} +func (activityAction) Policies() *internal.PolicyRegistry { return internal.ActivityPolicies() } +func (activityAction) GenerateTargetID(base string, occurrence occurrenceContext) string { + return internal.GenerateTimestampTargetID(base, occurrence.NominalTime) +} +func (activityAction) ParticipatesInCompletionHistory() bool { return false } +func (activityAction) Start(ctx context.Context, clients actionClients, input actionStartInput) (string, error) { + scheduler, start := input.Scheduler, input.Occurrence + spec := scheduler.Schedule.GetAction().GetStartActivity() + reuse := enumspb.ACTIVITY_ID_REUSE_POLICY_REJECT_DUPLICATE + if start.Manual { + reuse = enumspb.ACTIVITY_ID_REUSE_POLICY_ALLOW_DUPLICATE + } + response, err := clients.Frontend.StartActivityExecution(ctx, &workflowservice.StartActivityExecutionRequest{ + Namespace: scheduler.Namespace, Identity: scheduler.identity(), RequestId: start.RequestId, + ActivityId: internal.TargetID(start), ActivityType: spec.ActivityType, TaskQueue: spec.TaskQueue, + ScheduleToCloseTimeout: spec.ScheduleToCloseTimeout, ScheduleToStartTimeout: spec.ScheduleToStartTimeout, + StartToCloseTimeout: spec.StartToCloseTimeout, HeartbeatTimeout: spec.HeartbeatTimeout, RetryPolicy: spec.RetryPolicy, + Input: spec.Input, IdReusePolicy: reuse, IdConflictPolicy: enumspb.ACTIVITY_ID_CONFLICT_POLICY_FAIL, + SearchAttributes: spec.SearchAttributes, Header: spec.Header, + UserMetadata: spec.UserMetadata, Priority: spec.Priority, CompletionCallbacks: []*commonpb.Callback{input.Callback}, StartDelay: spec.StartDelay, + }) + if err != nil { + return "", err + } + return response.RunId, nil +} +func (activityAction) Cancel(context.Context, actionClients, *Scheduler, *commonpb.Execution) error { + return serviceerror.NewInvalidArgument("scheduled activities do not support cancellation overlap policies") +} +func (activityAction) Terminate(ctx context.Context, clients actionClients, scheduler *Scheduler, target *commonpb.Execution) error { + _, err := clients.Frontend.TerminateActivityExecution(ctx, &workflowservice.TerminateActivityExecutionRequest{Namespace: scheduler.Namespace, ActivityId: target.BusinessId, RunId: target.RunId, Reason: "terminated by schedule overlap policy", Identity: scheduler.identity()}) + return err +} +func (activityAction) Completion(info *persistencespb.ChasmNexusCompletion, execution *commonpb.Execution) actionCompletion { + status := enumspb.ACTIVITY_EXECUTION_STATUS_FAILED + if _, ok := info.Outcome.(*persistencespb.ChasmNexusCompletion_Success); ok { + status = enumspb.ACTIVITY_EXECUTION_STATUS_COMPLETED + } + if failure := info.GetFailure(); failure != nil { + switch failure.FailureInfo.(type) { + case *failurepb.Failure_CanceledFailureInfo: + status = enumspb.ACTIVITY_EXECUTION_STATUS_CANCELED + case *failurepb.Failure_TimeoutFailureInfo: + status = enumspb.ACTIVITY_EXECUTION_STATUS_TIMED_OUT + case *failurepb.Failure_TerminatedFailureInfo: + status = enumspb.ACTIVITY_EXECUTION_STATUS_TERMINATED + default: + status = enumspb.ACTIVITY_EXECUTION_STATUS_FAILED + } + } + failed := status == enumspb.ACTIVITY_EXECUTION_STATUS_FAILED || status == enumspb.ACTIVITY_EXECUTION_STATUS_TIMED_OUT + return actionCompletion{Result: &commonpb.ActionExecutionResult{Execution: execution, Status: &commonpb.ActionExecutionResult_ActivityStatus{ActivityStatus: status}}, Failed: failed} +} diff --git a/chasm/lib/scheduler/action_completion_internal_test.go b/chasm/lib/scheduler/action_completion_internal_test.go new file mode 100644 index 00000000000..862e9dac545 --- /dev/null +++ b/chasm/lib/scheduler/action_completion_internal_test.go @@ -0,0 +1,72 @@ +package scheduler + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" + commonpb "go.temporal.io/api/common/v1" + enumspb "go.temporal.io/api/enums/v1" + failurepb "go.temporal.io/api/failure/v1" + persistencespb "go.temporal.io/server/api/persistence/v1" +) + +func TestActionCompletion(t *testing.T) { + execution := &commonpb.Execution{Type: enumspb.EXECUTION_TYPE_ACTIVITY, BusinessId: "activity-id", RunId: "run-id"} + activityCases := []struct { + name string + info *persistencespb.ChasmNexusCompletion + status enumspb.ActivityExecutionStatus + failed bool + }{ + {name: "success", info: &persistencespb.ChasmNexusCompletion{Outcome: &persistencespb.ChasmNexusCompletion_Success{}}, status: enumspb.ACTIVITY_EXECUTION_STATUS_COMPLETED}, + {name: "application failure", info: completionWithFailure(&failurepb.Failure{}), status: enumspb.ACTIVITY_EXECUTION_STATUS_FAILED, failed: true}, + {name: "timeout", info: completionWithFailure(&failurepb.Failure{FailureInfo: &failurepb.Failure_TimeoutFailureInfo{TimeoutFailureInfo: &failurepb.TimeoutFailureInfo{}}}), status: enumspb.ACTIVITY_EXECUTION_STATUS_TIMED_OUT, failed: true}, + {name: "canceled", info: completionWithFailure(&failurepb.Failure{FailureInfo: &failurepb.Failure_CanceledFailureInfo{CanceledFailureInfo: &failurepb.CanceledFailureInfo{}}}), status: enumspb.ACTIVITY_EXECUTION_STATUS_CANCELED}, + {name: "terminated", info: completionWithFailure(&failurepb.Failure{FailureInfo: &failurepb.Failure_TerminatedFailureInfo{TerminatedFailureInfo: &failurepb.TerminatedFailureInfo{}}}), status: enumspb.ACTIVITY_EXECUTION_STATUS_TERMINATED}, + } + for _, tc := range activityCases { + t.Run(tc.name, func(t *testing.T) { + completion := (activityAction{}).Completion(tc.info, execution) + require.Equal(t, tc.failed, completion.Failed) + require.Equal(t, execution, completion.Result.GetExecution()) + require.Equal(t, tc.status, completion.Result.GetActivityStatus()) + require.Equal(t, enumspb.WORKFLOW_EXECUTION_STATUS_UNSPECIFIED, completion.Result.GetWorkflowStatus()) + }) + } + + workflowExecution := &commonpb.Execution{Type: enumspb.EXECUTION_TYPE_WORKFLOW, BusinessId: "workflow-id", RunId: "run-id"} + workflowCases := []struct { + name string + info *persistencespb.ChasmNexusCompletion + status enumspb.WorkflowExecutionStatus + failed bool + }{ + {name: "success", info: &persistencespb.ChasmNexusCompletion{Outcome: &persistencespb.ChasmNexusCompletion_Success{}}, status: enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED}, + {name: "application failure", info: completionWithFailure(&failurepb.Failure{}), status: enumspb.WORKFLOW_EXECUTION_STATUS_FAILED, failed: true}, + {name: "timeout", info: completionWithFailure(&failurepb.Failure{FailureInfo: &failurepb.Failure_TimeoutFailureInfo{TimeoutFailureInfo: &failurepb.TimeoutFailureInfo{}}}), status: enumspb.WORKFLOW_EXECUTION_STATUS_TIMED_OUT, failed: true}, + {name: "canceled", info: completionWithFailure(&failurepb.Failure{FailureInfo: &failurepb.Failure_CanceledFailureInfo{CanceledFailureInfo: &failurepb.CanceledFailureInfo{}}}), status: enumspb.WORKFLOW_EXECUTION_STATUS_CANCELED}, + {name: "terminated", info: completionWithFailure(&failurepb.Failure{FailureInfo: &failurepb.Failure_TerminatedFailureInfo{TerminatedFailureInfo: &failurepb.TerminatedFailureInfo{}}}), status: enumspb.WORKFLOW_EXECUTION_STATUS_TERMINATED}, + } + for _, tc := range workflowCases { + t.Run("workflow/"+tc.name, func(t *testing.T) { + completion := (workflowAction{}).Completion(tc.info, workflowExecution) + require.Equal(t, tc.failed, completion.Failed) + require.Equal(t, workflowExecution, completion.Result.GetExecution()) + require.Equal(t, tc.status, completion.Result.GetWorkflowStatus()) + require.Equal(t, enumspb.ACTIVITY_EXECUTION_STATUS_UNSPECIFIED, completion.Result.GetActivityStatus()) + }) + } +} + +func completionWithFailure(failure *failurepb.Failure) *persistencespb.ChasmNexusCompletion { + return &persistencespb.ChasmNexusCompletion{Outcome: &persistencespb.ChasmNexusCompletion_Failure{Failure: failure}} +} + +func TestActionTargetID(t *testing.T) { + nominal := time.Date(2024, time.January, 2, 3, 4, 5, 900000000, time.FixedZone("test", -6*60*60)) + want := "base-2024-01-02T09:04:05Z" + occurrence := occurrenceContext{NominalTime: nominal} + require.Equal(t, want, (workflowAction{}).GenerateTargetID("base", occurrence)) + require.Equal(t, want, (activityAction{}).GenerateTargetID("base", occurrence)) +} diff --git a/chasm/lib/scheduler/action_results.go b/chasm/lib/scheduler/action_results.go index e0705500d26..f10044481ca 100644 --- a/chasm/lib/scheduler/action_results.go +++ b/chasm/lib/scheduler/action_results.go @@ -1,12 +1,18 @@ package scheduler import ( + "fmt" + "strings" + commonpb "go.temporal.io/api/common/v1" enumspb "go.temporal.io/api/enums/v1" schedulepb "go.temporal.io/api/schedule/v1" + persistencespb "go.temporal.io/server/api/persistence/v1" schedulespb "go.temporal.io/server/api/schedule/v1" + "go.temporal.io/server/chasm" "go.temporal.io/server/chasm/lib/scheduler/internal" "go.temporal.io/server/common" + "go.temporal.io/server/common/metrics" ) func actionResult(start *schedulespb.BufferedStart) *schedulepb.ScheduleActionResult { @@ -17,3 +23,32 @@ func actionResult(start *schedulespb.BufferedStart) *schedulepb.ScheduleActionRe } return result } + +func (s *Scheduler) recordActivityCompletion(ctx chasm.MutableContext, invoker *Invoker, start *schedulespb.BufferedStart, info *persistencespb.ChasmNexusCompletion) error { + execution := common.CloneProto(internal.Execution(start)) + for _, link := range info.GetLinks() { + if activity := link.GetActivity(); activity != nil && activity.GetActivityId() == execution.BusinessId { + if execution.RunId != "" && execution.RunId != activity.GetRunId() { + return nil + } + execution.RunId = activity.GetRunId() + } + } + completion := implementation(s.Schedule.GetAction()).Completion(info, execution) + start.Completion = completion.Result + start.CompletionTime = info.CloseTime + if start.GetStartTime() == nil && info.StartTime != nil { + start.StartTime = info.StartTime + } + if internal.TracksExecution(start) && completion.Failed && s.Schedule.GetPolicies().GetPauseOnFailure() && !s.Schedule.GetState().GetPaused() { + s.Schedule.State.Paused = true + s.Schedule.State.Notes = fmt.Sprintf("paused, activity %s: %s", strings.ToLower(completion.Result.GetActivityStatus().String()), execution.BusinessId) + s.updateConflictToken() + } + if closeTime := info.GetCloseTime().AsTime(); !closeTime.IsZero() { + newTaggedMetricsHandler(ctx.MetricsHandler(), s).Timer(metrics.ScheduleCallbackLatency.Name()).Record(max(0, ctx.Now(s).Sub(closeTime))) + } + invoker.releaseWaiting(ctx, info.CloseTime) + s.Generator.Get(ctx).Generate(ctx) + return nil +} diff --git a/chasm/lib/scheduler/internal/policy.go b/chasm/lib/scheduler/internal/policy.go index 52d96efc5af..2f85480848c 100644 --- a/chasm/lib/scheduler/internal/policy.go +++ b/chasm/lib/scheduler/internal/policy.go @@ -133,6 +133,25 @@ func BuiltinPolicy(policy enumspb.ScheduleOverlapPolicy) PolicyDefinition { } } +func BufferLatestPolicy() PolicyDefinition { + return PolicyDefinition{Identity: PolicyIdentity{Custom: BufferLatestPolicyName}, Plan: func(s PolicySnapshot) PolicyDecision { + if len(s.Running) == 0 && s.Selected == nil { + return PolicyDecision{Start: true} + } + decision := PolicyDecision{Wait: true} + for _, waiting := range s.Waiting { + if waiting.CustomOverlapPolicy != BufferLatestPolicyName { + continue + } + if waiting.ActualTime.After(s.Occurrence.ActualTime) || waiting.ActualTime.Equal(s.Occurrence.ActualTime) && waiting.Occurrence > s.Occurrence.Occurrence { + return PolicyDecision{} + } + decision.Replace = append(decision.Replace, waiting) + } + return decision + }} +} + func WorkflowPolicies() *PolicyRegistry { definitions := make([]PolicyDefinition, 0, 6) for _, policy := range []enumspb.ScheduleOverlapPolicy{enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE, enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ALL, enumspb.SCHEDULE_OVERLAP_POLICY_CANCEL_OTHER, enumspb.SCHEDULE_OVERLAP_POLICY_TERMINATE_OTHER, enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL} { @@ -145,6 +164,19 @@ func WorkflowPolicies() *PolicyRegistry { return r } +func ActivityPolicies() *PolicyRegistry { + definitions := make([]PolicyDefinition, 0, 6) + for _, policy := range []enumspb.ScheduleOverlapPolicy{enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE, enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ALL, enumspb.SCHEDULE_OVERLAP_POLICY_TERMINATE_OTHER, enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL} { + definitions = append(definitions, BuiltinPolicy(policy)) + } + definitions = append(definitions, BufferLatestPolicy()) + r, err := NewPolicyRegistry(definitions, PolicyIdentity{}, ExecutionOperations{Terminate: true}) + if err != nil { + log.NewCLILogger().Fatal("invalid scheduler overlap policy registry", tag.Error(err)) + } + return r +} + func (r *PolicyRegistry) clone() *PolicyRegistry { return &PolicyRegistry{policies: maps.Clone(r.policies), defaultPolicy: r.defaultPolicy} } diff --git a/chasm/lib/scheduler/internal/policy_test.go b/chasm/lib/scheduler/internal/policy_test.go index f8846b9d213..1916553aa2e 100644 --- a/chasm/lib/scheduler/internal/policy_test.go +++ b/chasm/lib/scheduler/internal/policy_test.go @@ -25,6 +25,51 @@ func TestPolicyRegistrySubsetWithoutDefault(t *testing.T) { require.Error(t, err) } +func TestActivityPolicySelections(t *testing.T) { + registry := ActivityPolicies() + for _, policy := range []enumspb.ScheduleOverlapPolicy{enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE, enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ALL, enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL, enumspb.SCHEDULE_OVERLAP_POLICY_TERMINATE_OTHER} { + _, err := registry.Resolve(PolicyIdentity{}, PolicyIdentity{Builtin: policy}) + require.NoError(t, err) + } + for _, selection := range []PolicyIdentity{{}, {Builtin: enumspb.SCHEDULE_OVERLAP_POLICY_CANCEL_OTHER}, {Custom: "unknown"}, {Builtin: enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, Custom: BufferLatestPolicyName}} { + _, err := registry.Resolve(PolicyIdentity{}, selection) + require.Error(t, err) + } + _, err := registry.Resolve(PolicyIdentity{}, PolicyIdentity{Custom: BufferLatestPolicyName}) + require.NoError(t, err) + _, err = WorkflowPolicies().Resolve(PolicyIdentity{}, PolicyIdentity{Custom: BufferLatestPolicyName}) + require.Error(t, err) +} + +func TestBufferLatestReplacesOnlyOlderWaitingOccurrences(t *testing.T) { + now := time.Date(2026, 9, 4, 12, 0, 0, 0, time.UTC) + latest := func(id string, at time.Time) BufferedStartSnapshot { + return BufferedStartSnapshot{RequestID: id, ActualTime: at, CustomOverlapPolicy: BufferLatestPolicyName} + } + for _, active := range []bool{false, true} { + t.Run(map[bool]string{false: "selected", true: "running"}[active], func(t *testing.T) { + starts := []BufferedStartSnapshot{latest("first", now), latest("second", now.Add(time.Second)), {RequestID: "other-policy", ActualTime: now, OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ALL}, latest("older", now.Add(-time.Second)), latest("tie-newer-insertion", now.Add(time.Second))} + snapshot := BufferProcessingSnapshot{Starts: starts, Policies: ActivityPolicies(), CatchupWindow: time.Hour, LimitedActions: true, RemainingActions: 3} + if active { + snapshot.RunningExecutions = []ExecutionSnapshot{{TargetID: "active", RunID: "run"}} + } + plan := PlanBufferProcessing(snapshot, now) + require.EqualValues(t, map[bool]int{false: 2, true: 3}[active], plan.OverlapSkipped) + require.Equal(t, plan.OverlapSkipped, plan.OverlapSkippedByCustomPolicy[BufferLatestPolicyName]) + decisions := make(map[string]BufferDecision) + for _, decision := range plan.Decisions { + decisions[decision.RequestID] = decision + } + require.Equal(t, BufferDecisionDefer, decisions["tie-newer-insertion"].Action) + require.Equal(t, BufferDecisionDefer, decisions["other-policy"].Action) + require.Equal(t, BufferDecisionDiscard, decisions["older"].Action) + require.False(t, decisions["second"].ConsumesScheduledAction) + require.EqualValues(t, 3, snapshot.RemainingActions) + require.Equal(t, starts, snapshot.Starts) + }) + } +} + func TestPolicyPlannerReceivesDetachedSnapshots(t *testing.T) { id := PolicyIdentity{Custom: "test.only"} registry, err := NewPolicyRegistry([]PolicyDefinition{{Identity: id, Plan: func(snapshot PolicySnapshot) PolicyDecision { diff --git a/chasm/lib/scheduler/scheduler.go b/chasm/lib/scheduler/scheduler.go index 9b61ee22a68..b30e9a89238 100644 --- a/chasm/lib/scheduler/scheduler.go +++ b/chasm/lib/scheduler/scheduler.go @@ -739,6 +739,9 @@ func (s *Scheduler) HandleNexusCompletion( ) return nil } + if s.Schedule.GetAction().GetStartActivity() != nil { + return s.recordActivityCompletion(ctx, invoker, start, info) + } workflowID := start.GetWorkflowId() tracksCompletionResult := implementation(s.Schedule.GetAction()).ParticipatesInCompletionHistory() && internal.TracksExecution(start) diff --git a/service/frontend/workflow_handler.go b/service/frontend/workflow_handler.go index 3d0159f5307..fa8bcbfa105 100644 --- a/service/frontend/workflow_handler.go +++ b/service/frontend/workflow_handler.go @@ -3904,6 +3904,9 @@ func (wh *WorkflowHandler) CreateSchedule( if request.Schedule == nil { request.Schedule = &schedulepb.Schedule{} } + if err := chasmscheduler.ValidateScheduleActionPolicies(request.Schedule, request.InitialPatch); err != nil { + return nil, err + } if err := wh.validateScheduleOverlapPolicies(request.Schedule, request.InitialPatch, namespaceName.String()); err != nil { return nil, err } @@ -3923,6 +3926,9 @@ func (wh *WorkflowHandler) CreateSchedule( metricsHandler := wh.metricsScope(ctx).WithTags(metrics.HeaderCallsiteTag("CreateSchedule")) metrics.HeaderSize.With(metricsHandler).Record(int64(startWorkflow.GetHeader().Size())) } + if request.GetSchedule().GetAction().GetStartActivity() != nil && !useChasmScheduler { + return nil, serviceerror.NewFailedPrecondition("activity schedules require the CHASM scheduler") + } if useChasmScheduler { return wh.createScheduleCHASM(ctx, request) @@ -4725,6 +4731,9 @@ func (wh *WorkflowHandler) UpdateSchedule( if request.Schedule == nil { request.Schedule = &schedulepb.Schedule{} } + if err := chasmscheduler.ValidateScheduleActionPolicies(request.Schedule, nil); err != nil { + return nil, err + } if err := wh.validateScheduleOverlapPolicies(request.Schedule, nil, namespaceName.String()); err != nil { return nil, err } @@ -4762,6 +4771,10 @@ func (wh *WorkflowHandler) UpdateSchedule( } } + if request.GetSchedule().GetAction().GetStartActivity() != nil { + return nil, serviceerror.NewFailedPrecondition("activity schedules require the CHASM scheduler") + } + // Reject memo updates for V1 schedules. if request.GetMemo() != nil { return nil, serviceerror.NewFailedPrecondition("memo updates are not supported on workflow-backed schedules") @@ -6989,11 +7002,16 @@ func (wh *WorkflowHandler) validateSchedulePayloadSize( memo *commonpb.Memo, operation string, ) error { - action := schedule.GetAction().GetStartWorkflow() - if action == nil { - return serviceerror.NewInvalidArgument("Only StartWorkflow action is supported for schedules") + var input *commonpb.Payloads + switch action := schedule.GetAction().GetAction().(type) { + case *schedulepb.ScheduleAction_StartWorkflow: + input = action.StartWorkflow.GetInput() + case *schedulepb.ScheduleAction_StartActivity: + input = action.StartActivity.GetInput() + default: + return serviceerror.NewInvalidArgument("schedule action is required") } - payloadSize := memo.Size() + action.GetInput().Size() + payloadSize := memo.Size() + input.Size() sizeLimitError := wh.config.BlobSizeLimitError(namespaceName) sizeLimitWarn := wh.config.BlobSizeLimitWarn(namespaceName) return common.CheckEventBlobSizeLimit( diff --git a/service/frontend/workflow_handler_test.go b/service/frontend/workflow_handler_test.go index c6aee79dc7a..37e0184d3f9 100644 --- a/service/frontend/workflow_handler_test.go +++ b/service/frontend/workflow_handler_test.go @@ -6158,6 +6158,59 @@ func (s *WorkflowHandlerSuite) TestScheduleValidation() { } } +func (s *WorkflowHandlerSuite) TestActivitySchedulePolicyAndRoutingValidation() { + config := s.newConfig() + config.EnableSchedules = dc.GetBoolPropertyFnFilteredByNamespace(true) + wh := s.getWorkflowHandler(config) + + activitySchedule := func(policy enumspb.ScheduleOverlapPolicy, custom string) *schedulepb.Schedule { + policies := &schedulepb.SchedulePolicies{OverlapPolicy: policy} + if custom != "" { + policies.CustomOverlapPolicy = &schedulepb.CustomOverlapPolicy{Name: custom} + } + return &schedulepb.Schedule{ + Action: &schedulepb.ScheduleAction{Action: &schedulepb.ScheduleAction_StartActivity{ + StartActivity: &schedulepb.StartActivityExecutionInfo{ + ActivityId: "activity-id", + ActivityType: &commonpb.ActivityType{Name: "activity-type"}, + TaskQueue: &taskqueuepb.TaskQueue{Name: "task-queue"}, + }, + }}, + Policies: policies, + } + } + create := func(schedule *schedulepb.Schedule) error { + _, err := wh.CreateSchedule(context.Background(), &workflowservice.CreateScheduleRequest{ + Namespace: s.testNamespace.String(), + ScheduleId: "test-schedule", + RequestId: uuid.NewString(), + Schedule: schedule, + }) + return err + } + + s.Run("requires explicit policy", func() { + err := create(activitySchedule(enumspb.SCHEDULE_OVERLAP_POLICY_UNSPECIFIED, "")) + var invalidArgument *serviceerror.InvalidArgument + s.ErrorAs(err, &invalidArgument) + s.ErrorContains(err, "unsupported or missing overlap policy") + }) + + s.Run("rejects cancel other", func() { + err := create(activitySchedule(enumspb.SCHEDULE_OVERLAP_POLICY_CANCEL_OTHER, "")) + var invalidArgument *serviceerror.InvalidArgument + s.ErrorAs(err, &invalidArgument) + s.ErrorContains(err, "unsupported or missing overlap policy") + }) + + s.Run("rejects legacy scheduler routing", func() { + err := create(activitySchedule(enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, "")) + var failedPrecondition *serviceerror.FailedPrecondition + s.ErrorAs(err, &failedPrecondition) + s.ErrorContains(err, "activity schedules require the CHASM scheduler") + }) +} + func (s *WorkflowHandlerSuite) TestUpdateSchedule_ValidationAndErrors() { config := s.newConfig() config.EnableSchedules = dc.GetBoolPropertyFnFilteredByNamespace(true)