91 lines
4.7 KiB
Go
91 lines
4.7 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
orbit "git.yuntuo.hk/k2safe/orbit/sdk/go"
|
|
)
|
|
|
|
// verifiedTask 保存已核验分支的任务和尝试，最终与本地调用逐个关联。
|
|
type verifiedTask struct {
|
|
sequence int // sequence 实例请求序号。
|
|
taskID string // taskID 分支唯一任务身份。
|
|
attemptID string // attemptID 唯一成功尝试身份。
|
|
}
|
|
|
|
func verifySteps(ctx context.Context, cfg config, api workflowAPI, instance orbit.WorkflowInstance, runID string, sequence int, steps orbit.WorkflowSteps) ([]verifiedTask, string) {
|
|
if steps.InstanceID != instance.ID || steps.Revision != instance.Revision || len(steps.Steps) != 3 {
|
|
return nil, "STEP_COUNT_OR_PARENT_MISMATCH"
|
|
}
|
|
keys, ids, tasks, activations := map[string]bool{}, map[string]bool{}, map[string]bool{}, map[string]bool{}
|
|
for _, s := range steps.Steps {
|
|
kind := map[string]string{"pair": "PARALLEL", "left": "TASK", "right": "TASK"}[s.Key]
|
|
if kind == "" || s.Kind != kind || s.State != "SUCCEEDED" || !validID(s.ID) || s.Revision == 0 || ids[s.ID] || keys[s.Key] || s.ActivationSequence != 1 || s.ActivationID == nil || !validID(*s.ActivationID) || activations[*s.ActivationID] {
|
|
return nil, "STEP_MISMATCH"
|
|
}
|
|
keys[s.Key] = true
|
|
ids[s.ID] = true
|
|
activations[*s.ActivationID] = true
|
|
if kind == "PARALLEL" {
|
|
if s.TaskID != nil {
|
|
return nil, "JOIN_TASK_MISMATCH"
|
|
}
|
|
continue
|
|
}
|
|
if s.TaskID == nil || !validID(*s.TaskID) || tasks[*s.TaskID] {
|
|
return nil, "DUPLICATE_OR_INVALID_TASK"
|
|
}
|
|
tasks[*s.TaskID] = true
|
|
}
|
|
proofs := make([]verifiedTask, 0, 2)
|
|
for _, step := range steps.Steps {
|
|
if step.Kind != "TASK" {
|
|
continue
|
|
}
|
|
call, stop := context.WithTimeout(ctx, cfg.Timeout)
|
|
task, err := api.GetTask(call, *step.TaskID)
|
|
stop()
|
|
if err != nil {
|
|
return nil, "TASK_" + classify(err)
|
|
}
|
|
result, ok := decodePayload(task.Result)
|
|
if task.ID != *step.TaskID || !matchesTask(task, cfg, runID, sequence) || task.NamespaceID != instance.NamespaceID || task.State != orbit.TaskSucceeded || task.EffectUncertain || task.AttemptCount != 1 || task.CompletedAt == nil || task.CancelRequestedAt != nil || !ok || result != (workflowPayload{RunID: runID, Sequence: sequence}) {
|
|
return nil, "TASK_MISMATCH"
|
|
}
|
|
if task.CreatedAt.Before(instance.CreatedAt) || !task.AvailableAt.Equal(task.CreatedAt) || !task.DeadlineAt.Equal(instance.DeadlineAt) || task.CompletedAt.Before(task.CreatedAt) || instance.CompletedAt == nil || task.CompletedAt.After(*instance.CompletedAt) {
|
|
return nil, "TASK_TIMESTAMP_MISMATCH"
|
|
}
|
|
call, stop = context.WithTimeout(ctx, cfg.Timeout)
|
|
history, err := api.ListTaskAttempts(call, task.ID, orbit.HistoryPageRequest{Limit: 2})
|
|
stop()
|
|
if err != nil {
|
|
return nil, "HISTORY_" + classify(err)
|
|
}
|
|
if len(history.Items) != 1 || history.NextCursor != "" {
|
|
return nil, "ATTEMPT_COUNT_MISMATCH"
|
|
}
|
|
a := history.Items[0]
|
|
if !validID(a.ID) || a.TaskID != task.ID || a.State != orbit.AttemptSucceeded || a.Epoch != 1 || a.AuthorizedAt == nil || a.EndedAt == nil || a.ErrorCode != "" || a.Retryable || a.ConcurrencyToken != 0 {
|
|
return nil, "ATTEMPT_MISMATCH"
|
|
}
|
|
if a.CreatedAt.Before(task.AvailableAt) || a.AuthorizedAt.Before(a.CreatedAt) || a.EndedAt.Before(*a.AuthorizedAt) || !a.EndedAt.Equal(*task.CompletedAt) {
|
|
return nil, "ATTEMPT_TIMESTAMP_MISMATCH"
|
|
}
|
|
call, stop = context.WithTimeout(ctx, cfg.Timeout)
|
|
source, err := api.GetTaskSource(call, task.ID)
|
|
stop()
|
|
if err != nil {
|
|
return nil, "SOURCE_" + classify(err)
|
|
}
|
|
s := source.Workflow
|
|
if source.TaskID != task.ID || source.Kind != "WORKFLOW" || source.RetryOf != nil || source.Schedule != nil || s == nil || s.WorkflowID != instance.ID || s.DefinitionID != instance.DefinitionID || s.DefinitionVersionID != instance.DefinitionVersionID || s.Version != instance.Version || s.StepID != step.ID || s.StepKey != step.Key || s.StepKind != "TASK" || s.ActivationID != *step.ActivationID || s.ActivationSequence != 1 || s.PollRound != 0 || s.PollRoundID != nil || s.RetryOf != nil {
|
|
return nil, "SOURCE_MISMATCH"
|
|
}
|
|
proofs = append(proofs, verifiedTask{sequence: sequence, taskID: task.ID, attemptID: a.ID})
|
|
}
|
|
return proofs, ""
|
|
}
|
|
func matchesTask(t orbit.Task, c config, runID string, sequence int) bool {
|
|
input, ok := decodePayload(t.Payload)
|
|
return validID(t.ID) && t.QueueID == c.QueueID && t.Handler == handlerName(runID) && t.HandlerVersion == "v1" && t.PayloadVersion == "v1" && t.BusinessKey == "" && t.ConcurrencyKey == "" && t.Priority == 0 && t.RetryOf == nil && t.StartDeadlineAt == nil && t.RecoverySafety == orbit.RecoveryReconcileOnly && t.Retry == (orbit.RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: uint64(c.RunTimeout.Milliseconds())}) && ok && input == (workflowPayload{RunID: runID, Sequence: sequence})
|
|
}
|