291 lines
12 KiB
Go
291 lines
12 KiB
Go
package mysql
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"errors"
|
||
"time"
|
||
|
||
core "git.yuntuo.hk/k2safe/orbit/model"
|
||
"git.yuntuo.hk/k2safe/orbit/weave/definition"
|
||
"git.yuntuo.hk/k2safe/orbit/weave/engine"
|
||
"git.yuntuo.hk/k2safe/orbit/weave/instance"
|
||
weave "git.yuntuo.hk/k2safe/orbit/weave/model"
|
||
)
|
||
|
||
func createWorkflowSteps(ctx context.Context, tx *sql.Tx, g *workflowGraph, now time.Time) error {
|
||
create := func(key, kind string) error {
|
||
id, err := core.NewID()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if _, err := tx.ExecContext(ctx, "INSERT INTO weave_steps(namespace_id,id,workflow_id,`key`,kind,created_at,updated_at) VALUES (?,?,?,?,?,?,?)", g.view.NamespaceID, id, g.view.ID, key, kind, now, now); err != nil {
|
||
return err
|
||
}
|
||
g.steps[key] = &workflowStep{view: instance.StepView{ID: id, Key: key, Kind: weave.NodeKind(kind), State: weave.StepBlocked, Revision: 1}}
|
||
return nil
|
||
}
|
||
var visit func(definition.Block) error
|
||
visit = func(b definition.Block) error {
|
||
for _, s := range b.Steps {
|
||
if err := create(s.Key, s.Kind); err != nil {
|
||
return err
|
||
}
|
||
if s.Branch != nil {
|
||
for _, c := range s.Branch.Cases {
|
||
if err := visit(c.Body); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
if err := visit(*s.Branch.Default); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
if s.Parallel != nil {
|
||
for _, branch := range s.Parallel.Branches {
|
||
if err := visit(branch.Body); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
if err := visit(g.version.Specification.Body); err != nil {
|
||
return err
|
||
}
|
||
if err := create(workflowRootKey, "JOIN"); err != nil {
|
||
return err
|
||
}
|
||
root := g.steps[workflowRootKey]
|
||
activation, err := core.NewID()
|
||
if err != nil {
|
||
return 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 (?,?,?,?,1,'ACTIVE',?,?,?)`, g.view.NamespaceID, activation, g.view.ID, root.view.ID, string(g.view.Input), now, now); err != nil {
|
||
return err
|
||
}
|
||
r, err := tx.ExecContext(ctx, `UPDATE weave_steps SET state='ACTIVE',current_activation_id=? WHERE namespace_id=? AND id=?`, activation, g.view.NamespaceID, root.view.ID)
|
||
if err := exactlyOne(r, err); err != nil {
|
||
return err
|
||
}
|
||
root.view.State, root.view.ActivationID, root.input = weave.StepActive, &activation, g.view.Input
|
||
return nil
|
||
}
|
||
|
||
// applyWorkflowDecision 调用方必须已按应用→排序队列→实例持锁，结果和待续状态一并提交。
|
||
// 初始受理的实例行尚未对其他事务可见，同样由应用锁保护，不在函数内重排父级锁。
|
||
func applyWorkflowDecision(ctx context.Context, tx *sql.Tx, g *workflowGraph, d engine.Decision, now time.Time) error {
|
||
if len(d.Changes) > 32 {
|
||
return errors.New("workflow change budget exceeded")
|
||
}
|
||
for _, c := range d.Changes {
|
||
row, ok := g.steps[c.Key]
|
||
if !ok || string(row.view.Kind) != c.Kind {
|
||
return errors.New("workflow decision step mismatch")
|
||
}
|
||
switch c.Operation {
|
||
case "ACTIVATE":
|
||
if row.view.State != weave.StepBlocked || row.view.ActivationID != nil {
|
||
return errors.New("workflow step already activated")
|
||
}
|
||
if err := activateWorkflowStep(ctx, tx, g, row, c, now); err != nil {
|
||
return err
|
||
}
|
||
case "SKIP":
|
||
if row.view.State != weave.StepBlocked || row.view.ActivationID != nil || c.State.Status != weave.StepSkipped {
|
||
return errors.New("workflow skip state invalid")
|
||
}
|
||
case "COMPLETE":
|
||
if row.view.ActivationID == nil || (row.view.State != weave.StepActive && row.view.State != weave.StepWaiting) || c.State.Status != weave.StepSucceeded {
|
||
return errors.New("workflow completion state invalid")
|
||
}
|
||
r, err := tx.ExecContext(ctx, `UPDATE weave_activations SET state='SUCCEEDED',output=?,completed_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND workflow_id=? AND step_id=? AND state IN ('ACTIVE','WAITING')`, string(c.State.Output), now, now, g.view.NamespaceID, row.view.ActivationID, g.view.ID, row.view.ID)
|
||
if err := exactlyOne(r, err); err != nil {
|
||
return err
|
||
}
|
||
default:
|
||
return errors.New("unsupported workflow decision")
|
||
}
|
||
r, err := tx.ExecContext(ctx, `UPDATE weave_steps SET state=?,current_activation_id=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND workflow_id=? AND revision=?`, c.State.Status, row.view.ActivationID, now, g.view.NamespaceID, row.view.ID, g.view.ID, row.view.Revision)
|
||
if err := exactlyOne(r, err); err != nil {
|
||
return err
|
||
}
|
||
row.view.State, row.input, row.output = c.State.Status, c.State.Input, c.State.Output
|
||
row.view.Revision++
|
||
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, "key": c.Key, "state": c.State.Status, "activation_id": row.view.ActivationID}); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
root := g.steps[workflowRootKey]
|
||
if root == nil || root.view.ActivationID == nil {
|
||
return errors.New("workflow root missing")
|
||
}
|
||
state := weave.InstanceRunning
|
||
var completed *time.Time
|
||
var output any
|
||
rootState := weave.StepActive
|
||
if d.Complete {
|
||
if d.More {
|
||
return errors.New("workflow complete decision has pending work")
|
||
}
|
||
for key, row := range g.steps {
|
||
if key != workflowRootKey && row.view.State != weave.StepSucceeded && row.view.State != weave.StepSkipped {
|
||
return errors.New("workflow completion has unfinished steps")
|
||
}
|
||
}
|
||
state, rootState, completed, output = weave.InstanceSucceeded, weave.StepSucceeded, &now, string(d.Output)
|
||
} else if !d.More {
|
||
waiting, active := false, false
|
||
for key, row := range g.steps {
|
||
if key == workflowRootKey {
|
||
continue
|
||
}
|
||
if row.view.State == weave.StepWaiting {
|
||
waiting = true
|
||
}
|
||
if row.view.State == weave.StepActive && (row.view.Kind == weave.NodeTask || row.view.Kind == weave.NodePoll) {
|
||
active = true
|
||
}
|
||
}
|
||
if waiting && !active {
|
||
state = weave.InstanceWaiting
|
||
}
|
||
}
|
||
r, err := tx.ExecContext(ctx, `UPDATE weave_activations SET state=?,output=?,completed_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND workflow_id=? AND step_id=? AND state='ACTIVE'`, rootState, output, completed, now, g.view.NamespaceID, root.view.ActivationID, g.view.ID, root.view.ID)
|
||
if err := exactlyOne(r, err); err != nil {
|
||
return err
|
||
}
|
||
r, err = tx.ExecContext(ctx, `UPDATE weave_steps SET state=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND revision=?`, rootState, now, g.view.NamespaceID, root.view.ID, root.view.Revision)
|
||
if err := exactlyOne(r, err); err != nil {
|
||
return err
|
||
}
|
||
root.view.Revision++
|
||
root.view.State = rootState
|
||
// 包括首批决策在内，每次持久化推进递增实例版本；不能以未改变字段的affected_rows判断存在性。
|
||
r, err = tx.ExecContext(ctx, `UPDATE weave_instances SET state=?,output=?,completed_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND revision=?`, state, output, completed, now, g.view.NamespaceID, g.view.ID, g.view.Revision)
|
||
if err := exactlyOne(r, err); err != nil {
|
||
return err
|
||
}
|
||
if d.Complete {
|
||
if _, err := tx.ExecContext(ctx, `UPDATE pulse_execution_gates SET paused=TRUE,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=?`, now, g.view.NamespaceID, g.gate); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
if d.More {
|
||
id, err := core.NewID()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
_, err = tx.ExecContext(ctx, `INSERT INTO weave_pending_advances(namespace_id,id,workflow_id,step_id,expected_step_revision,checkpoint,available_at,created_at,updated_at) VALUES (?,?,?,?,?,'{"schema_version":1,"kind":"ROOT"}',?,?,?) ON DUPLICATE KEY UPDATE expected_step_revision=VALUES(expected_step_revision),checkpoint=VALUES(checkpoint),available_at=VALUES(available_at),revision=revision+1,updated_at=VALUES(updated_at)`, g.view.NamespaceID, id, g.view.ID, root.view.ID, root.view.Revision, now, now, now)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
} else {
|
||
if _, err := tx.ExecContext(ctx, `DELETE FROM weave_pending_advances WHERE namespace_id=? AND workflow_id=? AND step_id=?`, g.view.NamespaceID, g.view.ID, root.view.ID); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
g.view.State = state
|
||
g.view.Revision++
|
||
g.view.CompletedAt = completed
|
||
g.view.Output = d.Output
|
||
return nil
|
||
}
|
||
|
||
func activateWorkflowStep(ctx context.Context, tx *sql.Tx, g *workflowGraph, row *workflowStep, c engine.Change, now time.Time) error {
|
||
id, err := core.NewID()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
var taskID *core.ID
|
||
if c.Task != nil {
|
||
v, err := core.NewID()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
taskID = &v
|
||
}
|
||
var direct *core.ID
|
||
if c.Kind == "TASK" {
|
||
direct = taskID
|
||
}
|
||
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 (?,?,?,?,1,?,?,?,?,?)`, g.view.NamespaceID, id, g.view.ID, row.view.ID, c.State.Status, direct, string(c.State.Input), now, now); err != nil {
|
||
return err
|
||
}
|
||
row.view.ActivationID = &id
|
||
row.view.ActivationSequence = 1
|
||
row.view.TaskID = direct
|
||
deadline := g.view.DeadlineAt
|
||
if c.Kind == "POLL" {
|
||
node, ok := g.program.Step(c.Key)
|
||
if !ok || node.Poll == nil || taskID == nil {
|
||
return errors.New("workflow poll decision incomplete")
|
||
}
|
||
p := node.Poll
|
||
deadline = minimumTime(deadline, now.Add(time.Duration(p.TimeoutMS)*time.Millisecond))
|
||
pollID, err := core.NewID()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if _, err := tx.ExecContext(ctx, `INSERT INTO weave_poll_states(namespace_id,id,workflow_id,activation_id,max_rounds,used_rounds,max_attempts_per_round,interval_ms,deadline_at,current_task_id,created_at,updated_at) VALUES (?,?,?,?,?,1,?,?,?,?,?,?)`, g.view.NamespaceID, pollID, g.view.ID, id, p.MaxRounds, p.Task.Retry.MaxAttempts, p.IntervalMS, deadline, taskID, now, now); err != nil {
|
||
return err
|
||
}
|
||
roundID, err := core.NewID()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if _, err := tx.ExecContext(ctx, "INSERT INTO weave_poll_rounds(namespace_id,id,workflow_id,activation_id,`round`,task_id,created_at,updated_at) VALUES (?,?,?,?,1,?,?,?)", g.view.NamespaceID, roundID, g.view.ID, id, taskID, now, now); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
if c.Wait != nil {
|
||
waitID, err := core.NewID()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
until := minimumTime(deadline, now.Add(time.Duration(c.Wait.TimeoutMS)*time.Millisecond))
|
||
var name, version any
|
||
if c.Wait.Kind != "TIMER" {
|
||
name, version = c.Wait.Name, c.Wait.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 (?,?,?,?,?,?,?,?,?,?,?)`, g.view.NamespaceID, waitID, g.view.ID, id, c.Wait.Kind, []byte(c.Wait.Correlation), until, name, version, now, now); err != nil {
|
||
return err
|
||
}
|
||
if c.Wait.Kind == "TIMER" {
|
||
if err := saveWorkflowClock(ctx, tx, g, row, "TIMER", row.view.Revision+1, until); err != nil {
|
||
return err
|
||
}
|
||
} else if c.Wait.Kind == "SIGNAL" {
|
||
// 先扫描早到信号；未匹配时运行器按原等待截止继续持久化。
|
||
if err := saveWorkflowClock(ctx, tx, g, row, "SIGNAL", row.view.Revision+1, now); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
}
|
||
if c.Task != nil {
|
||
if taskID == nil {
|
||
return errors.New("workflow task identity missing")
|
||
}
|
||
task, err := insertAcceptedTaskUntil(ctx, tx, g.view.NamespaceID, *taskID, *c.Task, nil, &deadline)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
r, 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, id, g.gate, g.view.NamespaceID, *taskID)
|
||
if err := exactlyOne(r, err); err != nil {
|
||
return err
|
||
}
|
||
if err := emitEvent(ctx, tx, g.view.NamespaceID, *taskID, task.Revision, "task", "task.accepted", map[string]any{"task_id": task.ID, "state": task.State, "workflow_id": g.view.ID, "activation_id": id}); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
if c.Kind == "POLL" {
|
||
due := deadline
|
||
if c.Wait != nil {
|
||
due = now
|
||
}
|
||
return saveWorkflowClock(ctx, tx, g, row, "POLL", row.view.Revision+1, due)
|
||
}
|
||
return nil
|
||
}
|