146 lines
7.4 KiB
Go
146 lines
7.4 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/pulse/service"
|
||
"git.yuntuo.hk/k2safe/orbit/weave/engine"
|
||
weave "git.yuntuo.hk/k2safe/orbit/weave/model"
|
||
"git.yuntuo.hk/k2safe/orbit/weave/recovery"
|
||
)
|
||
|
||
// retryWorkflowPoll 重新执行当前业务轮；新激活保留旧轮次、同轮尝试、精确输入及两个绝对截止。
|
||
func retryWorkflowPoll(ctx context.Context, tx *sql.Tx, g *workflowGraph, row *workflowStep, in recovery.RetryStepInput, now time.Time, out *recovery.Result) (map[string]any, error) {
|
||
p, last, err := readWorkflowPollBudget(ctx, tx, g, row)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
origin, err := workflowActivationOrigin(ctx, tx, g, row)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if !now.Before(p.DeadlineAt) || origin.sequence >= p.MaxRounds*p.MaxAttemptsPerRound {
|
||
return nil, fault.Conflict
|
||
}
|
||
h, err := readWorkflowPollRoundHistory(ctx, tx, g, row, p, p.UsedRounds)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if h.last.ID != last || !h.last.State.Terminal() || h.last.EffectUncertain || h.used >= p.MaxAttemptsPerRound || h.count >= p.MaxAttemptsPerRound || !now.Before(h.first.DeadlineAt) {
|
||
return nil, fault.Conflict
|
||
}
|
||
// 已确认的PENDING应按间隔和业务轮次预算推进，不能借人工恢复再查同一轮。
|
||
if h.last.State == "SUCCEEDED" {
|
||
kind, err := g.program.ClassifyPoll(row.view.Key, h.last.Result)
|
||
if err == nil && kind == engine.PollPending {
|
||
return nil, fault.Conflict
|
||
}
|
||
}
|
||
node, _ := g.program.Step(row.view.Key)
|
||
var wait *workflowSignalWait
|
||
if node.Poll.FinalSignal != nil {
|
||
w, err := readWorkflowSignalWait(ctx, tx, g, row)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if !now.Before(w.deadline) {
|
||
return nil, fault.Conflict
|
||
}
|
||
var pending int
|
||
if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM weave_signals WHERE namespace_id=? AND workflow_id=? AND target_step_key=? AND target_activation_sequence=? AND state='PENDING'`, g.view.NamespaceID, g.view.ID, row.view.Key, w.sequence).Scan(&pending); err != nil {
|
||
return nil, err
|
||
}
|
||
if pending != 0 {
|
||
return nil, fault.Conflict
|
||
}
|
||
if w.winner != nil {
|
||
if _, err := selectedWorkflowSignal(ctx, tx, g, row, w, now); err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
wait = &w
|
||
}
|
||
policy := h.first.Retry
|
||
policy.MaxAttempts = p.MaxAttemptsPerRound - h.used
|
||
input, err := service.NormalizeSubmit(service.SubmitInput{QueueID: h.first.QueueID, Handler: h.first.Handler, HandlerVersion: h.first.HandlerVersion, PayloadVersion: h.first.PayloadVersion, Payload: h.first.Payload, BusinessKey: h.first.BusinessKey, ConcurrencyKey: h.first.ConcurrencyKey, Priority: h.first.Priority, RecoverySafety: h.first.RecoverySafety, Retry: policy, TaskTimeoutMS: node.Poll.Task.TimeoutMS})
|
||
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, g.view.NamespaceID, taskID, input, &last, &h.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,input,created_at,updated_at) VALUES (?,?,?,?,?,'ACTIVE',?,?,?)`, g.view.NamespaceID, activation, g.view.ID, row.view.ID, origin.sequence+1, string(row.input), now, now); err != nil {
|
||
return nil, err
|
||
}
|
||
result, 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, g.view.NamespaceID, taskID)
|
||
if err := exactlyOne(result, err); err != nil {
|
||
return nil, err
|
||
}
|
||
// 旧预算仅结清检查点；旧任务、旧激活输出、轮次及已匹配信号不改写。
|
||
if _, err := tx.ExecContext(ctx, `UPDATE weave_poll_states SET current_task_id=NULL,next_poll_at=NULL,resolved_at=COALESCE(resolved_at,?),revision=revision+1,updated_at=? WHERE namespace_id=? AND id=?`, now, now, g.view.NamespaceID, p.ID); err != nil {
|
||
return nil, err
|
||
}
|
||
pollID, err := core.NewID()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if _, err := tx.ExecContext(ctx, `INSERT INTO weave_poll_states(namespace_id,id,workflow_id,activation_id,max_rounds,used_rounds,first_round,first_round_attempts_used,max_attempts_per_round,interval_ms,deadline_at,current_task_id,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, g.view.NamespaceID, pollID, g.view.ID, activation, p.MaxRounds, p.UsedRounds, p.UsedRounds, h.used, p.MaxAttemptsPerRound, p.IntervalMS, p.DeadlineAt, taskID, now, now); err != nil {
|
||
return nil, err
|
||
}
|
||
roundID, err := core.NewID()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if _, err := tx.ExecContext(ctx, "INSERT INTO weave_poll_rounds(namespace_id,id,workflow_id,activation_id,`round`,task_id) VALUES (?,?,?,?,?,?)", g.view.NamespaceID, roundID, g.view.ID, activation, p.UsedRounds, taskID); err != nil {
|
||
return nil, err
|
||
}
|
||
if wait != nil {
|
||
if _, err := tx.ExecContext(ctx, `UPDATE weave_waits SET state='CANCELLED',resolved_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND state='WAITING'`, now, now, g.view.NamespaceID, wait.id); err != nil {
|
||
return nil, err
|
||
}
|
||
id, err := core.NewID()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if _, err := tx.ExecContext(ctx, `INSERT INTO weave_waits(namespace_id,id,workflow_id,activation_id,kind,correlation_key,deadline_at,signal_name,payload_version,created_at,updated_at) VALUES (?,?,?,?,'POLL',?,?,?,?,?,?)`, g.view.NamespaceID, id, g.view.ID, activation, []byte(wait.correlation), wait.deadline, wait.contract.Name, wait.contract.PayloadVersion, now, now); err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
result, 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, g.view.NamespaceID, row.view.ID, row.view.Revision, row.view.ActivationID)
|
||
if err := exactlyOne(result, err); err != nil {
|
||
return nil, err
|
||
}
|
||
if row.view.TaskID != nil {
|
||
return nil, errors.New("poll cannot have direct task reference")
|
||
}
|
||
row.view.State, row.view.ActivationID = weave.StepActive, &activation
|
||
row.view.ActivationSequence = origin.sequence + 1
|
||
row.view.Revision++
|
||
row.output = nil
|
||
if err := saveWorkflowClock(ctx, tx, g, row, "POLL", row.view.Revision, p.DeadlineAt); err != nil {
|
||
return nil, err
|
||
}
|
||
if err := emitEvent(ctx, tx, g.view.NamespaceID, taskID, task.Revision, "task", "task.accepted", map[string]any{"task_id": taskID, "state": task.State, "workflow_id": g.view.ID, "activation_id": activation, "retry_of": last, "round": p.UsedRounds}); err != nil {
|
||
return nil, err
|
||
}
|
||
if err := emitEvent(ctx, tx, g.view.NamespaceID, row.view.ID, row.view.Revision, "workflow_step", "workflow.step.changed", map[string]any{"workflow_id": g.view.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": origin.sequence + 1, "retry_of": last, "task_id": taskID, "safe_to_retry": true, "round": p.UsedRounds, "max_rounds": p.MaxRounds, "used_attempts": h.used, "remaining_attempts": policy.MaxAttempts, "deadline_at": p.DeadlineAt, "task_deadline_at": h.first.DeadlineAt}, nil
|
||
}
|