mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-30 14:56:19 +00:00
Stephan BehnkeandGitHub
8aebcfce9f
## What changed? - added "stamp" field to both workflow execution and transfer/timer workflow task. - bumping stamp up when rescheduling a workflow task - making sure stamp is current when dequeuing workflow task - making sure stamp is current when recoding workflow start ## Why? Activity tasks already have the "stamp" mechanism. This is the same idea for workflow tasks. It essentially provides a unified approach to invalidate activity and workflow tasks when rescheduling them. ## How did you test it? - [ ] built - [ ] run locally and tested manually - [x] covered by existing tests - [x] added new unit test(s) - [ ] added new functional test(s) I intend to write more tests for when I'm implementing the update priority feature. It's much easier to setup and test than versioning overrides (which are the only way right now to trigger a stamp increase).
69 lines
1.8 KiB
Go
69 lines
1.8 KiB
Go
package isworkflowtaskvalid
|
|
|
|
import (
|
|
"context"
|
|
|
|
"go.temporal.io/server/api/historyservice/v1"
|
|
"go.temporal.io/server/common"
|
|
"go.temporal.io/server/common/definition"
|
|
"go.temporal.io/server/service/history/api"
|
|
"go.temporal.io/server/service/history/consts"
|
|
historyi "go.temporal.io/server/service/history/interfaces"
|
|
)
|
|
|
|
func Invoke(
|
|
ctx context.Context,
|
|
req *historyservice.IsWorkflowTaskValidRequest,
|
|
shardContext historyi.ShardContext,
|
|
workflowConsistencyChecker api.WorkflowConsistencyChecker,
|
|
) (resp *historyservice.IsWorkflowTaskValidResponse, retError error) {
|
|
isValid := false
|
|
err := api.GetAndUpdateWorkflowWithNew(
|
|
ctx,
|
|
req.Clock,
|
|
definition.NewWorkflowKey(
|
|
req.NamespaceId,
|
|
req.Execution.WorkflowId,
|
|
req.Execution.RunId,
|
|
),
|
|
func(workflowLease api.WorkflowLease) (*api.UpdateWorkflowAction, error) {
|
|
isTaskValid, err := isWorkflowTaskValid(workflowLease, req.ScheduledEventId, req.GetStamp())
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
isValid = isTaskValid
|
|
return &api.UpdateWorkflowAction{
|
|
Noop: true,
|
|
CreateWorkflowTask: false,
|
|
}, nil
|
|
},
|
|
nil,
|
|
shardContext,
|
|
workflowConsistencyChecker,
|
|
)
|
|
return &historyservice.IsWorkflowTaskValidResponse{
|
|
IsValid: isValid,
|
|
}, err
|
|
}
|
|
|
|
func isWorkflowTaskValid(
|
|
workflowLease api.WorkflowLease,
|
|
scheduledEventID int64,
|
|
stamp int32,
|
|
) (bool, error) {
|
|
mutableState := workflowLease.GetMutableState()
|
|
if !mutableState.IsWorkflowExecutionRunning() {
|
|
return false, consts.ErrWorkflowCompleted
|
|
}
|
|
|
|
workflowTask := mutableState.GetWorkflowTaskByID(scheduledEventID)
|
|
if workflowTask == nil {
|
|
return false, nil
|
|
}
|
|
if stamp != workflowTask.Stamp {
|
|
// This happens when the workflow task was rescheduled.
|
|
return false, nil
|
|
}
|
|
return workflowTask.StartedEventID == common.EmptyEventID, nil
|
|
}
|