161 lines
7.3 KiB
Go
161 lines
7.3 KiB
Go
package mysql
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"encoding/json"
|
||
"errors"
|
||
"time"
|
||
|
||
"git.yuntuo.hk/k2safe/orbit/internal/fault"
|
||
core "git.yuntuo.hk/k2safe/orbit/model"
|
||
"git.yuntuo.hk/k2safe/orbit/platform/auth"
|
||
weave "git.yuntuo.hk/k2safe/orbit/weave/model"
|
||
"git.yuntuo.hk/k2safe/orbit/weave/recovery"
|
||
)
|
||
|
||
// RearmWait 仅对独立SIGNAL/TIMER换代，固定首次等待的预算；到期事实不能通过新建等待清除。
|
||
func (s *WorkflowRecoveryStore) RearmWait(ctx context.Context, t auth.Token, id core.ID, key string, in recovery.RearmWaitInput) (recovery.Result, error) {
|
||
return s.mutate(ctx, t, id, key, "workflow.wait.rearm", in, in.Decision, func(tx *sql.Tx, g *workflowGraph, _ core.ID, now time.Time, out *recovery.Result) (map[string]any, error) {
|
||
row, err := workflowRecoveryTarget(g, in.StepDecision)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
node, ok := g.program.Step(row.view.Key)
|
||
if !ok || node.Signal == nil && node.Timer == nil {
|
||
return nil, fault.Invalid("wait rearm requires a SIGNAL or TIMER step")
|
||
}
|
||
if row.view.State != weave.StepWaiting && row.view.State != weave.StepNeedsReview && row.view.State != weave.StepFailed && row.view.State != weave.StepCancelled {
|
||
return nil, fault.Conflict
|
||
}
|
||
if g.view.StopKind != nil || g.view.CancelRequestedAt != nil || !now.Before(g.view.DeadlineAt) {
|
||
return nil, fault.Conflict
|
||
}
|
||
if err := workflowStepSettled(ctx, tx, g, row); err != nil {
|
||
return nil, err
|
||
}
|
||
origin, err := workflowActivationOrigin(ctx, tx, g, row)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if origin.sequence >= recovery.MaxWaitActivations {
|
||
return nil, fault.Conflict
|
||
}
|
||
if err := verifyWorkflowWaitHistory(ctx, tx, g, row, origin); err != nil {
|
||
return nil, err
|
||
}
|
||
var w workflowSignalWait
|
||
if node.Signal != nil {
|
||
w, err = readWorkflowSignalWait(ctx, tx, g, row)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if err := noPendingWorkflowSignals(ctx, tx, g, row, w.sequence); err != nil {
|
||
return nil, err
|
||
}
|
||
} else {
|
||
var count int
|
||
if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM weave_waits WHERE namespace_id=? AND workflow_id=? AND activation_id=?`, t.NamespaceID, id, row.view.ActivationID).Scan(&count); err != nil {
|
||
return nil, err
|
||
}
|
||
if count != 1 {
|
||
return nil, errors.New("timer rearm wait set invalid")
|
||
}
|
||
var kind string
|
||
var name, version *string
|
||
var wake *core.ID
|
||
if err := tx.QueryRowContext(ctx, `SELECT id,kind,state,deadline_at,signal_name,payload_version,correlation_key,winner_signal_id,wake_task_id FROM weave_waits WHERE namespace_id=? AND workflow_id=? AND activation_id=? FOR UPDATE`, t.NamespaceID, id, row.view.ActivationID).Scan(&w.id, &kind, &w.state, &w.deadline, &name, &version, &w.correlation, &w.winner, &wake); err != nil {
|
||
return nil, err
|
||
}
|
||
expectedInput, _ := json.Marshal(map[string]uint64{"delay_ms": node.Timer.DelayMS})
|
||
if kind != "TIMER" || name != nil || version != nil || w.correlation != "" || wake != nil || !sameWorkflowJSON(row.input, expectedInput) || !w.deadline.Equal(minimumTime(origin.started.Add(time.Duration(node.Timer.DelayMS)*time.Millisecond), g.view.DeadlineAt)) {
|
||
return nil, errors.New("timer rearm contract mismatch")
|
||
}
|
||
}
|
||
if w.winner != nil || w.state != "WAITING" && w.state != "CANCELLED" || !now.Before(w.deadline) {
|
||
return nil, fault.Conflict
|
||
}
|
||
activation, err := core.NewID()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
wait, err := core.NewID()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
beforeState := row.view.State
|
||
if row.view.State == weave.StepWaiting {
|
||
result, err := tx.ExecContext(ctx, `UPDATE weave_activations SET state='CANCELLED',completed_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND workflow_id=? AND step_id=? AND state='WAITING'`, now, now, t.NamespaceID, in.ActivationID, id, row.view.ID)
|
||
if err := exactlyOne(result, err); err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
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, t.NamespaceID, w.id); 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 (?,?,?,?,?,'WAITING',?,?,?)`, t.NamespaceID, activation, id, row.view.ID, origin.sequence+1, string(origin.input), now, now); err != nil {
|
||
return nil, err
|
||
}
|
||
var name, version *string
|
||
if w.contract != nil {
|
||
name = &w.contract.Name
|
||
version = &w.contract.PayloadVersion
|
||
}
|
||
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 (?,?,?,?,?,?,?,?,?,?,?)`, t.NamespaceID, wait, id, activation, row.view.Kind, []byte(w.correlation), w.deadline, name, version, now, now); err != nil {
|
||
return nil, err
|
||
}
|
||
result, err := tx.ExecContext(ctx, `UPDATE weave_steps SET state='WAITING',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, in.ActivationID)
|
||
if err := exactlyOne(result, err); err != nil {
|
||
return nil, err
|
||
}
|
||
row.view.State, row.view.ActivationID = weave.StepWaiting, &activation
|
||
row.view.ActivationSequence = origin.sequence + 1
|
||
row.view.Revision++
|
||
row.output = nil
|
||
due := w.deadline
|
||
if node.Signal != nil {
|
||
due = now
|
||
}
|
||
if err := saveWorkflowClock(ctx, tx, g, row, string(row.view.Kind), row.view.Revision, due); 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, "state": row.view.State, "activation_id": activation}); err != nil {
|
||
return nil, err
|
||
}
|
||
out.Step = &row.view
|
||
return map[string]any{"step_id": row.view.ID, "previous_activation_id": in.ActivationID, "activation_id": activation, "sequence": origin.sequence + 1, "previous_state": beforeState, "previous_wait_id": w.id, "previous_wait_state": w.state, "wait_id": wait, "deadline_at": w.deadline, "wait_verified": true}, nil
|
||
})
|
||
}
|
||
|
||
func verifyWorkflowWaitHistory(ctx context.Context, tx *sql.Tx, g *workflowGraph, row *workflowStep, origin workflowActivationBase) error {
|
||
rows, err := tx.QueryContext(ctx, `SELECT id,sequence,state,input,task_id FROM weave_activations WHERE namespace_id=? AND workflow_id=? AND step_id=? ORDER BY sequence LIMIT 33`, g.view.NamespaceID, g.view.ID, row.view.ID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer rows.Close()
|
||
var count uint32
|
||
var last core.ID
|
||
for rows.Next() {
|
||
var id core.ID
|
||
var sequence uint32
|
||
var state string
|
||
var input []byte
|
||
var task *core.ID
|
||
if err := rows.Scan(&id, &sequence, &state, &input, &task); err != nil {
|
||
return err
|
||
}
|
||
count++
|
||
if sequence != count || task != nil || !sameWorkflowJSON(input, origin.input) || id != *row.view.ActivationID && state != "NEEDS_REVIEW" && state != "FAILED" && state != "CANCELLED" {
|
||
return errors.New("wait activation history invalid")
|
||
}
|
||
last = id
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
return err
|
||
}
|
||
if count != origin.sequence || last != *row.view.ActivationID {
|
||
return errors.New("wait activation history incomplete")
|
||
}
|
||
return nil
|
||
}
|