161 lines
7.6 KiB
Go
161 lines
7.6 KiB
Go
package mysql
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"errors"
|
||
"time"
|
||
|
||
"git.yuntuo.hk/k2safe/orbit/internal/fault"
|
||
core "git.yuntuo.hk/k2safe/orbit/model"
|
||
"git.yuntuo.hk/k2safe/orbit/platform/auth"
|
||
"git.yuntuo.hk/k2safe/orbit/pulse/service"
|
||
"git.yuntuo.hk/k2safe/orbit/weave/definition"
|
||
"git.yuntuo.hk/k2safe/orbit/weave/engine"
|
||
weave "git.yuntuo.hk/k2safe/orbit/weave/model"
|
||
"git.yuntuo.hk/k2safe/orbit/weave/recovery"
|
||
)
|
||
|
||
// RetryStep 为TASK或POLL建立新的执行激活；保留原输入、绝对截止和跨激活尝试预算。
|
||
func (s *WorkflowRecoveryStore) RetryStep(ctx context.Context, t auth.Token, id core.ID, key string, in recovery.RetryStepInput) (recovery.Result, error) {
|
||
return s.mutate(ctx, t, id, key, "workflow.step.retry", in, in.Decision, func(tx *sql.Tx, g *workflowGraph, _ core.ID, now time.Time, out *recovery.Result) (map[string]any, error) {
|
||
row, err := workflowRecoveryStep(g, in.StepDecision)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if g.view.StopKind != nil || g.view.CancelRequestedAt != nil || !now.Before(g.view.DeadlineAt) {
|
||
return nil, fault.Conflict
|
||
}
|
||
node, ok := g.program.Step(row.view.Key)
|
||
if !ok || node.Task == nil && node.Poll == nil {
|
||
return nil, fault.Invalid("retry currently supports TASK and POLL steps")
|
||
}
|
||
if err := workflowStepSettled(ctx, tx, g, row); err != nil {
|
||
return nil, err
|
||
}
|
||
if node.Poll != nil {
|
||
return retryWorkflowPoll(ctx, tx, g, row, in, now, out)
|
||
}
|
||
var sequence uint32
|
||
if err := tx.QueryRowContext(ctx, `SELECT sequence FROM weave_activations WHERE namespace_id=? AND id=?`, t.NamespaceID, row.view.ActivationID).Scan(&sequence); err != nil {
|
||
return nil, err
|
||
}
|
||
// 人工取消尚未开始的任务也不能无限创建激活；每个静态TASK的激活数不超过原max_attempts。
|
||
if sequence == 0 || sequence >= node.Task.Retry.MaxAttempts {
|
||
return nil, fault.Conflict
|
||
}
|
||
history, err := tx.QueryContext(ctx, `SELECT id,sequence,task_id FROM weave_activations WHERE namespace_id=? AND workflow_id=? AND step_id=? ORDER BY sequence LIMIT 101`, t.NamespaceID, id, row.view.ID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
type prior struct {
|
||
activation, task core.ID
|
||
sequence uint32
|
||
}
|
||
var previous []prior
|
||
for history.Next() {
|
||
var v prior
|
||
if err := history.Scan(&v.activation, &v.sequence, &v.task); err != nil {
|
||
history.Close()
|
||
return nil, err
|
||
}
|
||
previous = append(previous, v)
|
||
}
|
||
err = history.Err()
|
||
history.Close()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if len(previous) != int(sequence) || previous[len(previous)-1].activation != in.ActivationID {
|
||
return nil, errors.New("workflow retry activation history incomplete")
|
||
}
|
||
var first, last service.TaskView
|
||
var used uint32
|
||
for i, prior := range previous {
|
||
if prior.sequence != uint32(i+1) {
|
||
return nil, errors.New("workflow retry sequence invalid")
|
||
}
|
||
v, err := readTask(ctx, tx, t.NamespaceID, prior.task, false)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if (v.State != "FAILED" && v.State != "CANCELLED") || v.EffectUncertain {
|
||
return nil, fault.Conflict
|
||
}
|
||
if i == 0 {
|
||
first = v
|
||
if definition.ValidateData(node.Task.InputType, first.Payload) != nil || !first.DeadlineAt.Equal(minimumTime(first.CreatedAt.Add(time.Duration(node.Task.TimeoutMS)*time.Millisecond), g.view.DeadlineAt)) {
|
||
return nil, errors.New("workflow retry initial contract or deadline mismatch")
|
||
}
|
||
}
|
||
expectedPolicy := node.Task.Retry
|
||
if used >= expectedPolicy.MaxAttempts {
|
||
return nil, fault.Conflict
|
||
}
|
||
expectedPolicy.MaxAttempts -= used
|
||
if v.Retry != expectedPolicy || v.AttemptCount > v.Retry.MaxAttempts || v.QueueID != node.Task.QueueID || v.Handler != node.Task.Handler || v.HandlerVersion != node.Task.HandlerVersion || v.PayloadVersion != node.Task.PayloadVersion || v.RecoverySafety != node.Task.RecoverySafety || v.Priority != node.Task.Priority || !sameWorkflowJSON(v.Payload, row.input) || !v.DeadlineAt.Equal(first.DeadlineAt) || v.BusinessKey != first.BusinessKey || v.ConcurrencyKey != first.ConcurrencyKey {
|
||
return nil, errors.New("workflow retry frozen task history mismatch")
|
||
}
|
||
if i == 0 && v.RetryOf != nil || i > 0 && (v.RetryOf == nil || *v.RetryOf != last.ID) {
|
||
return nil, errors.New("workflow retry task ancestry mismatch")
|
||
}
|
||
var attempts uint32
|
||
if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM pulse_attempts WHERE namespace_id=? AND task_id=?`, t.NamespaceID, v.ID).Scan(&attempts); err != nil {
|
||
return nil, err
|
||
}
|
||
if attempts != v.AttemptCount {
|
||
return nil, errors.New("workflow retry attempt history incomplete")
|
||
}
|
||
used += v.AttemptCount
|
||
last = v
|
||
}
|
||
if used >= node.Task.Retry.MaxAttempts || !now.Before(first.DeadlineAt) || first.StartDeadlineAt != nil && !now.Before(*first.StartDeadlineAt) {
|
||
return nil, fault.Conflict
|
||
}
|
||
policy := node.Task.Retry
|
||
policy.MaxAttempts -= used
|
||
input, err := service.NormalizeSubmit(service.SubmitInput{QueueID: first.QueueID, Handler: first.Handler, HandlerVersion: first.HandlerVersion, PayloadVersion: first.PayloadVersion, Payload: first.Payload, BusinessKey: first.BusinessKey, ConcurrencyKey: first.ConcurrencyKey, Priority: first.Priority, RecoverySafety: first.RecoverySafety, Retry: policy, TaskTimeoutMS: node.Task.TimeoutMS, StartDeadlineAt: first.StartDeadlineAt})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
activation, err := core.NewID()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
taskID, err := core.NewID()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
task, err := insertAcceptedTaskUntil(ctx, tx, t.NamespaceID, taskID, input, &last.ID, &first.DeadlineAt)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if _, err := tx.ExecContext(ctx, `INSERT INTO weave_activations(namespace_id,id,workflow_id,step_id,sequence,state,task_id,input,created_at,updated_at) VALUES (?,?,?,?,?,'ACTIVE',?,?,?,?)`, t.NamespaceID, activation, id, row.view.ID, sequence+1, taskID, string(row.input), now, now); err != nil {
|
||
return nil, err
|
||
}
|
||
res, err := tx.ExecContext(ctx, `UPDATE pulse_tasks SET owner_kind=?,owner_id=?,gate_id=? WHERE namespace_id=? AND id=? AND owner_kind IS NULL`, engine.TaskOwnerKind, activation, g.gate, t.NamespaceID, taskID)
|
||
if err := exactlyOne(res, err); err != nil {
|
||
return nil, err
|
||
}
|
||
res, err = tx.ExecContext(ctx, `UPDATE weave_steps SET state='ACTIVE',current_activation_id=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND revision=? AND current_activation_id=?`, activation, now, t.NamespaceID, row.view.ID, row.view.Revision, row.view.ActivationID)
|
||
if err := exactlyOne(res, err); err != nil {
|
||
return nil, err
|
||
}
|
||
row.view.State, row.view.ActivationID, row.view.TaskID = weave.StepActive, &activation, &taskID
|
||
row.view.ActivationSequence = sequence + 1
|
||
row.view.Revision++
|
||
row.output = nil
|
||
if err := deleteWorkflowClock(ctx, tx, g, row); err != nil {
|
||
return nil, err
|
||
}
|
||
if err := emitEvent(ctx, tx, t.NamespaceID, taskID, task.Revision, "task", "task.accepted", map[string]any{"task_id": taskID, "state": task.State, "workflow_id": id, "activation_id": activation, "retry_of": last.ID}); err != nil {
|
||
return nil, err
|
||
}
|
||
if err := emitEvent(ctx, tx, t.NamespaceID, row.view.ID, row.view.Revision, "workflow_step", "workflow.step.changed", map[string]any{"workflow_id": id, "activation_id": activation, "state": row.view.State}); err != nil {
|
||
return nil, err
|
||
}
|
||
out.Step, out.Task = &row.view, &task
|
||
return map[string]any{"step_id": row.view.ID, "previous_activation_id": in.ActivationID, "activation_id": activation, "sequence": sequence + 1, "retry_of": last.ID, "task_id": taskID, "safe_to_retry": true, "used_attempts": used, "remaining_attempts": policy.MaxAttempts, "deadline_at": first.DeadlineAt}, nil
|
||
})
|
||
}
|