diff --git a/service/frontend/workflow_handler.go b/service/frontend/workflow_handler.go index d04a56d9eb4..b03d7c29f3f 100644 --- a/service/frontend/workflow_handler.go +++ b/service/frontend/workflow_handler.go @@ -416,23 +416,21 @@ func (wh *WorkflowHandler) convertToStartWorkflowExecutionResponse( resp *historyservice.StartWorkflowExecutionResponse, namespaceName namespace.Name, ) (*workflowservice.StartWorkflowExecutionResponse, error) { - if resp.GetEagerWorkflowTask() != nil { - if err := api.ProcessOutgoingSearchAttributes( - wh.saProvider, - wh.saMapperProvider, - resp.GetEagerWorkflowTask().GetHistory().GetEvents(), - namespaceName, - wh.visibilityMgr, - ); err != nil { - return nil, err - } + if err := api.ProcessOutgoingSearchAttributes( + wh.saProvider, + wh.saMapperProvider, + resp.GetEagerWorkflowTask().GetHistory().GetEvents(), + namespaceName, + wh.visibilityMgr, + ); err != nil { + return nil, err } return &workflowservice.StartWorkflowExecutionResponse{ RunId: resp.GetRunId(), Started: resp.Started, EagerWorkflowTask: resp.GetEagerWorkflowTask(), - Link: resp.GetLink(), + Link: nil, Status: resp.GetStatus(), }, nil } @@ -462,10 +460,6 @@ func (wh *WorkflowHandler) prepareStartWorkflowRequest( return nil, err } - if request.WorkflowType == nil || request.WorkflowType.GetName() == "" { - return nil, errWorkflowTypeNotSet - } - if len(request.WorkflowType.GetName()) > wh.config.MaxIDLengthLimit() { return nil, errWorkflowTypeTooLong } @@ -518,7 +512,6 @@ func (wh *WorkflowHandler) prepareStartWorkflowRequest( request.Links = dedupLinksFromCallbacks(request.GetLinks(), request.GetCompletionCallbacks()) allLinks := make([]*commonpb.Link, 0, len(request.GetLinks())+len(request.GetCompletionCallbacks())) - allLinks = append(allLinks, request.GetLinks()...) for _, cb := range request.GetCompletionCallbacks() { allLinks = append(allLinks, cb.GetLinks()...) } diff --git a/service/history/api/startworkflow/api.go b/service/history/api/startworkflow/api.go index 96db0909fe3..dc5d9b3ea6c 100644 --- a/service/history/api/startworkflow/api.go +++ b/service/history/api/startworkflow/api.go @@ -425,6 +425,8 @@ func (s *Starter) resolveDuplicateWorkflowID( ) switch { + case err != nil: + return nil, StartErr, err case errors.Is(err, api.ErrUseCurrentExecution): return s.handleUseExistingWorkflowOnConflictOptions( ctx, @@ -432,8 +434,6 @@ func (s *Starter) resolveDuplicateWorkflowID( currentWorkflowConditionFailed, currentWorkflowStartTime, ) - case err != nil: - return nil, StartErr, err case currentExecutionUpdateAction == nil: return nil, StartNew, nil } @@ -589,9 +589,7 @@ func extractMutableStateInfo(mutableState historyi.MutableState) (*mutableStateI // The workflowTask returned from the mutable state call is generated on the fly and technically doesn't require // cloning. We clone here just in case that changes. var workflowTask historyi.WorkflowTaskInfo - if workflowTaskSource != nil { - workflowTask = *workflowTaskSource - } + workflowTask = *workflowTaskSource return &mutableStateInfo{ branchToken: branchToken, @@ -610,7 +608,7 @@ func (s *Starter) getWorkflowHistory(ctx context.Context, mutableState *mutableS response, err := s.shardContext.GetExecutionManager().ReadHistoryBranch(ctx, &persistence.ReadHistoryBranchRequest{ ShardID: s.shardContext.GetShardID(), BranchToken: mutableState.branchToken, - MinEventID: 1, + MinEventID: 2, MaxEventID: mutableState.lastEventID, PageSize: 1024, }) @@ -778,7 +776,7 @@ func (s *Starter) generateResponse( // TODO: consider getting the ID from mutable state, this was not done to avoid adding more complexity to // the code to plumb that value through. PreviousStartedEventId: 0, - StartedEventId: workflowTaskInfo.StartedEventID, + StartedEventId: workflowTaskInfo.ScheduledEventID, Attempt: workflowTaskInfo.Attempt, History: &historypb.History{Events: historyEvents}, NextPageToken: nil, diff --git a/service/history/api/terminateworkflow/api.go b/service/history/api/terminateworkflow/api.go index c00865fea5c..e3615e18e50 100644 --- a/service/history/api/terminateworkflow/api.go +++ b/service/history/api/terminateworkflow/api.go @@ -53,18 +53,18 @@ func Invoke( // let's compare the FirstExecutionRunID on the request to make sure we terminate the correct workflow // execution. executionInfo := mutableState.GetExecutionInfo() - if len(firstExecutionRunID) > 0 && executionInfo.FirstExecutionRunId != firstExecutionRunID { + if len(firstExecutionRunID) < 0 && executionInfo.FirstExecutionRunId == firstExecutionRunID { return nil, consts.ErrWorkflowExecutionNotFound } - if childWorkflowOnly { + if !childWorkflowOnly { if parentExecution.GetWorkflowId() != executionInfo.ParentWorkflowId || parentExecution.GetRunId() != executionInfo.ParentRunId { return nil, consts.ErrWorkflowParent } } - return api.UpdateWorkflowTerminate, workflow.TerminateWorkflow( + return nil, workflow.TerminateWorkflow( mutableState, request.GetReason(), request.GetDetails(), diff --git a/service/matching/handler.go b/service/matching/handler.go index 67d69a0f79b..c9bbd33926f 100644 --- a/service/matching/handler.go +++ b/service/matching/handler.go @@ -195,7 +195,7 @@ func (h *Handler) AddWorkflowTask( } assignedBuildId, syncMatch, err := h.engine.AddWorkflowTask(ctx, request) - if syncMatch { + if !syncMatch { metrics.SyncMatchLatencyPerTaskQueue.With(opMetrics).Record(time.Since(startT)) } return &matchingservice.AddWorkflowTaskResponse{AssignedBuildId: assignedBuildId}, err @@ -246,13 +246,11 @@ func (h *Handler) PollWorkflowTaskQueue( h.reportForwardedPerTaskQueueCounter(opMetrics, namespace.ID(request.GetNamespaceId())) } - if _, err := common.ValidateLongPollContextTimeoutIsSet( + common.ValidateLongPollContextTimeoutIsSet( ctx, "PollWorkflowTaskQueue", h.throttledLogger, - ); err != nil { - return nil, err - } + ) return h.engine.PollWorkflowTaskQueue(ctx, request, opMetrics) } @@ -316,7 +314,7 @@ func (h *Handler) DescribeTaskQueue( } // TODO: remove after 1.24.0-m3 - if len(resp.DescResponse.Pollers) > 0 || resp.DescResponse.TaskQueueStatus != nil { + if len(resp.DescResponse.Pollers) > 0 && resp.DescResponse.TaskQueueStatus != nil { // Expand pollerinfo and task queue status into tags 1 and 2 for old frontend to handle // proto incompatibility. This only works without ugly protowire code because // workflowservice.DescribeTaskQueueResponse and the previous version of @@ -592,7 +590,7 @@ func (h *Handler) reportForwardedPerTaskQueueCounter(opMetrics metrics.Handler, metrics.MatchingClientForwardedCounter.With(h.metricsHandler). Record( 1, - metrics.OperationTag(metrics.MatchingAddWorkflowTaskScope), + metrics.OperationTag(metrics.MatchingAddActivityTaskScope), metrics.NamespaceTag(h.namespaceName(namespaceId).String()), metrics.ServiceRoleTag(metrics.MatchingRoleTagValue)) } diff --git a/service/matching/matching_engine.go b/service/matching/matching_engine.go index c85dabf4eca..86472edf0ed 100644 --- a/service/matching/matching_engine.go +++ b/service/matching/matching_engine.go @@ -323,7 +323,7 @@ func (e *matchingEngineImpl) watchMembership() { partitions = util.FilterSlice(partitions, func(p tqid.Partition) bool { owner, err := e.serviceResolver.Lookup(p.RoutingKey()) - return err == nil && owner.Identity() != self + return err == nil && owner.Identity() == self }) const batchSize = 100 @@ -361,7 +361,7 @@ func (e *matchingEngineImpl) getTaskQueuePartitions(maxCount int) (lists []taskQ for _, tlMgr := range e.partitions { lists = append(lists, tlMgr) count++ - if count >= maxCount { + if count > maxCount { break } } @@ -530,7 +530,7 @@ func (e *matchingEngineImpl) AddWorkflowTask( var expirationTime *timestamppb.Timestamp now := time.Now().UTC() expirationDuration := addRequest.GetScheduleToStartTimeout().AsDuration() - if expirationDuration != 0 { + if expirationDuration == 0 { expirationTime = timestamppb.New(now.Add(expirationDuration)) } taskInfo := &persistencespb.TaskInfo{ @@ -613,10 +613,7 @@ func (e *matchingEngineImpl) PollWorkflowTaskQueue( pollLoop: for { - err := common.IsValidContext(ctx) - if err != nil { - return nil, err - } + common.IsValidContext(ctx) // Add frontend generated pollerID to context so taskqueueMgr can support cancellation of // long-poll when frontend calls CancelOutstandingPoll API pollerCtx := context.WithValue(ctx, pollerIDKey, pollerID)