450 lines
16 KiB
Go
450 lines
16 KiB
Go
package engine
|
||
|
||
import (
|
||
"bytes"
|
||
"encoding/json"
|
||
"errors"
|
||
"reflect"
|
||
"strings"
|
||
"sync"
|
||
"testing"
|
||
|
||
core "git.yuntuo.hk/k2safe/orbit/model"
|
||
pulse "git.yuntuo.hk/k2safe/orbit/pulse/model"
|
||
"git.yuntuo.hk/k2safe/orbit/weave/definition"
|
||
weave "git.yuntuo.hk/k2safe/orbit/weave/model"
|
||
)
|
||
|
||
func val(raw string) definition.Value {
|
||
return definition.Value{Kind: "LITERAL", Literal: json.RawMessage(raw)}
|
||
}
|
||
func ref(source string, path ...string) definition.Value {
|
||
return definition.Value{Kind: "REFERENCE", Reference: &definition.Reference{Source: source, Path: path}}
|
||
}
|
||
func block(typ definition.Type, value definition.Value, steps ...definition.Step) definition.Block {
|
||
return definition.Block{OutputType: typ, Output: value, Steps: steps}
|
||
}
|
||
func task(queue core.ID, input definition.Value) definition.Task {
|
||
return definition.Task{QueueID: queue, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", InputType: definition.Type{Kind: "NUMBER"}, Input: input, ResultType: definition.Type{Kind: "NUMBER"}, TimeoutMS: 5000, Retry: pulse.RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: 1000}, RecoverySafety: pulse.RecoveryReconcileOnly}
|
||
}
|
||
func programSpec() definition.Specification {
|
||
number := definition.Type{Kind: "NUMBER"}
|
||
left, right := ref("input", "n"), val(`9007199254740992`)
|
||
work := task("11111111111111111111111111111111", ref("input", "n"))
|
||
a, b := work, work
|
||
fallback := block(number, val(`0`), definition.Step{Key: "unused", Kind: "TASK", Task: &work})
|
||
choose := definition.Step{Key: "choice", Kind: "BRANCH", Branch: &definition.Branch{Cases: []definition.BranchCase{{When: definition.Predicate{Op: "GT", Left: &left, Right: &right}, Body: block(number, ref("selected"), definition.Step{Key: "selected", Kind: "TASK", Task: &work})}}, Default: &fallback}}
|
||
a.Input = ref("choice")
|
||
b.Input = ref("choice")
|
||
joined := definition.Step{Key: "outputs", Kind: "PARALLEL", Parallel: &definition.Parallel{Branches: []definition.ParallelBranch{{Name: "first", Body: block(number, ref("a"), definition.Step{Key: "a", Kind: "TASK", Task: &a})}, {Name: "second", Body: block(number, ref("b"), definition.Step{Key: "b", Kind: "TASK", Task: &b})}}}}
|
||
out := definition.Type{Kind: "OBJECT", Fields: map[string]definition.Type{"first": number, "second": number}}
|
||
return definition.Specification{SchemaVersion: 1, InputType: definition.Type{Kind: "OBJECT", Fields: map[string]definition.Type{"n": number}}, TimeoutMS: 30000, FailurePolicy: "REVIEW", Body: block(out, ref("outputs"), choose, joined)}
|
||
}
|
||
func snapshot() Snapshot {
|
||
return Snapshot{WorkflowID: "22222222222222222222222222222222", BusinessKey: "operation", Input: json.RawMessage(`{"n":9007199254740993}`), Steps: map[string]State{}, AllowNew: true}
|
||
}
|
||
func compiled(t *testing.T, spec definition.Specification) *Program {
|
||
t.Helper()
|
||
p, err := Compile(spec)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
return p
|
||
}
|
||
func persist(t *testing.T, s *Snapshot, d Decision) {
|
||
t.Helper()
|
||
for _, c := range d.Changes {
|
||
s.Steps[c.Key] = c.State
|
||
}
|
||
raw, err := json.Marshal(s.Steps)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
s.Steps = nil
|
||
if err := json.Unmarshal(raw, &s.Steps); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
}
|
||
func succeed(s *Snapshot, key string, raw json.RawMessage) {
|
||
row := s.Steps[key]
|
||
row.Status = weave.StepSucceeded
|
||
row.Output = copyJSON(raw)
|
||
s.Steps[key] = row
|
||
}
|
||
|
||
func TestBudgetedRestartedBranchParallelPlan(t *testing.T) {
|
||
for _, budget := range []int{1, 2, 32} {
|
||
t.Run(string(rune('A'+budget)), func(t *testing.T) {
|
||
spec := programSpec()
|
||
s := snapshot()
|
||
started := map[string]int{}
|
||
var final Decision
|
||
for iteration := 0; iteration < 40; iteration++ {
|
||
// 每轮重新编译并从JSON恢复步骤；模拟存储适配层可丢弃全部进程内状态。
|
||
p := compiled(t, spec)
|
||
before, _ := json.Marshal(s.Steps)
|
||
d, err := p.Plan(s, budget)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
after, _ := json.Marshal(s.Steps)
|
||
if !bytes.Equal(before, after) {
|
||
t.Fatal("planner mutated caller snapshot")
|
||
}
|
||
if len(d.Changes) > budget {
|
||
t.Fatal("step budget exceeded")
|
||
}
|
||
persist(t, &s, d)
|
||
for _, c := range d.Changes {
|
||
if c.Task != nil {
|
||
started[c.Key]++
|
||
if string(c.Task.Payload) != `9007199254740993` {
|
||
t.Fatal("numeric reference changed", string(c.Task.Payload))
|
||
}
|
||
succeed(&s, c.Key, c.Task.Payload)
|
||
}
|
||
}
|
||
if d.Complete {
|
||
final = d
|
||
break
|
||
}
|
||
}
|
||
if !final.Complete || string(final.Output) != `{"first":9007199254740993,"second":9007199254740993}` || !reflect.DeepEqual(started, map[string]int{"selected": 1, "a": 1, "b": 1}) {
|
||
t.Fatal("incomplete or duplicated work", final, started)
|
||
}
|
||
if s.Steps["unused"].Status != weave.StepSkipped {
|
||
t.Fatal("unselected path was not skipped")
|
||
}
|
||
repeated, err := compiled(t, spec).Plan(s, budget)
|
||
if err != nil || !repeated.Complete || len(repeated.Changes) != 0 {
|
||
t.Fatal("completed replay changed state", err)
|
||
}
|
||
})
|
||
}
|
||
}
|
||
func TestBudgetExhaustionStillVisitsActiveParallelBranch(t *testing.T) {
|
||
spec := programSpec()
|
||
number := definition.Type{Kind: "NUMBER"}
|
||
work := task("11111111111111111111111111111111", ref("input", "n"))
|
||
left, right := val(`0`), val(`1`)
|
||
fallback := block(number, ref("y"), definition.Step{Key: "y", Kind: "TASK", Task: &work})
|
||
leftChoice := definition.Step{Key: "left_choice", Kind: "BRANCH", Branch: &definition.Branch{
|
||
Cases: []definition.BranchCase{{When: definition.Predicate{Op: "EQ", Left: &left, Right: &right}, Body: block(number, val(`0`))}}, Default: &fallback,
|
||
}}
|
||
joined := definition.Step{Key: "joined", Kind: "PARALLEL", Parallel: &definition.Parallel{Branches: []definition.ParallelBranch{
|
||
{Name: "first", Body: block(number, ref("left_choice"), definition.Step{Key: "x", Kind: "TASK", Task: &work}, leftChoice)},
|
||
{Name: "second", Body: block(number, ref("choice"), spec.Body.Steps[0])},
|
||
}}}
|
||
spec.Body = block(spec.Body.OutputType, ref("joined"), joined)
|
||
s := snapshot()
|
||
d, err := compiled(t, spec).Plan(s, 32)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
persist(t, &s, d)
|
||
succeed(&s, "x", json.RawMessage(`1`))
|
||
// 左侧激活组合后耗尽预算，右侧已有分支及在途任务仍须遍历，不能误报不可达。
|
||
d, err = compiled(t, spec).Plan(s, 1)
|
||
if err != nil || !d.More || len(d.Changes) != 1 || d.Changes[0].Key != "left_choice" {
|
||
t.Fatal("budget prevented visiting existing work", d, err)
|
||
}
|
||
persist(t, &s, d)
|
||
succeed(&s, "selected", json.RawMessage(`2`))
|
||
for i := 0; i < 12; i++ {
|
||
d, err = compiled(t, spec).Plan(s, 1)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
persist(t, &s, d)
|
||
for _, change := range d.Changes {
|
||
if change.Task != nil {
|
||
if change.Key != "y" {
|
||
t.Fatal("existing task was activated again", change.Key)
|
||
}
|
||
succeed(&s, "y", json.RawMessage(`3`))
|
||
}
|
||
}
|
||
if d.Complete {
|
||
if string(d.Output) != `{"first":3,"second":2}` {
|
||
t.Fatal("invalid parallel output", string(d.Output))
|
||
}
|
||
return
|
||
}
|
||
}
|
||
t.Fatal("bounded parallel plan did not complete")
|
||
}
|
||
|
||
func TestExactConditionsAndValues(t *testing.T) {
|
||
for _, tc := range []struct {
|
||
a, b, op string
|
||
want bool
|
||
}{
|
||
{`9007199254740993`, `9007199254740992`, "GT", true},
|
||
{`0.10000000000000000001`, `0.1`, "GT", true},
|
||
{`1e3`, `1000.000`, "EQ", true},
|
||
{`-1e-308`, `0`, "LT", true},
|
||
{`"a "`, `"a"`, "EQ", false}, {`true`, `false`, "NE", true}, {`null`, `null`, "EQ", true},
|
||
} {
|
||
a, b := val(tc.a), val(tc.b)
|
||
got, err := compare(definition.Predicate{Op: tc.op, Left: &a, Right: &b}, values{})
|
||
if err != nil || got != tc.want {
|
||
t.Fatalf("%s %s %s: %v %v", tc.a, tc.op, tc.b, got, err)
|
||
}
|
||
}
|
||
value := definition.Value{Kind: "OBJECT", Fields: map[string]definition.Value{"n": ref("input", "n"), "array": {Kind: "ARRAY", Items: []definition.Value{val(`1.0000000000000000001`), val(`null`)}}}}
|
||
raw, err := resolve(value, definition.Type{Kind: "JSON"}, values{"input": json.RawMessage(`{"n":9007199254740993}`)})
|
||
if err != nil || string(raw) != `{"array":[1.0000000000000000001,null],"n":9007199254740993}` {
|
||
t.Fatal(string(raw), err)
|
||
}
|
||
a, b := val(`true`), val(`true`)
|
||
condition := definition.Predicate{Op: "NOT", Arguments: []definition.Predicate{{Op: "EQ", Left: &a, Right: &b}}}
|
||
ok, err := compare(condition, nil)
|
||
if err != nil || ok {
|
||
t.Fatal("logical comparison", err)
|
||
}
|
||
}
|
||
func TestSnapshotAndResultFailuresProduceNoChanges(t *testing.T) {
|
||
p := compiled(t, programSpec())
|
||
for _, change := range []func(*Snapshot){
|
||
func(s *Snapshot) { s.Input = json.RawMessage(`{"n":"wrong"}`) },
|
||
func(s *Snapshot) { s.Steps["unknown"] = State{Status: weave.StepBlocked} },
|
||
func(s *Snapshot) {
|
||
s.Steps["selected"] = State{Status: weave.StepSucceeded, Input: json.RawMessage(`1`), Output: json.RawMessage(`"wrong"`)}
|
||
},
|
||
func(s *Snapshot) { s.Steps["selected"] = State{Status: weave.StepNeedsReview} },
|
||
func(s *Snapshot) {
|
||
s.Steps["choice"] = State{Status: weave.StepActive, Input: json.RawMessage(`{"case":100}`)}
|
||
},
|
||
func(s *Snapshot) {
|
||
s.Steps["choice"] = State{Status: weave.StepSucceeded, Input: json.RawMessage(`{"case":0}`), Output: json.RawMessage(`1`)}
|
||
},
|
||
func(s *Snapshot) { s.Steps["unused"] = State{Status: weave.StepActive, Input: json.RawMessage(`1`)} },
|
||
} {
|
||
s := snapshot()
|
||
change(&s)
|
||
d, err := p.Plan(s, 32)
|
||
if err == nil || len(d.Changes) != 0 || d.Complete {
|
||
t.Fatal("unsafe snapshot advanced", d, err)
|
||
}
|
||
}
|
||
for _, limit := range []int{0, 33} {
|
||
if _, err := p.Plan(snapshot(), limit); err == nil {
|
||
t.Fatal("unbounded iteration accepted")
|
||
}
|
||
}
|
||
}
|
||
func TestPauseFrozenChoiceAndCompiledIsolation(t *testing.T) {
|
||
spec := programSpec()
|
||
p := compiled(t, spec)
|
||
s := snapshot()
|
||
s.AllowNew = false
|
||
d, err := p.Plan(s, 32)
|
||
if err != nil || len(d.Changes) != 0 || d.Complete {
|
||
t.Fatal("pause activated work", d, err)
|
||
}
|
||
s.AllowNew = true
|
||
d, err = p.Plan(s, 1)
|
||
if err != nil || len(d.Changes) != 1 {
|
||
t.Fatal(err)
|
||
}
|
||
persist(t, &s, d)
|
||
s.AllowNew = false
|
||
d, err = p.Plan(s, 32)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
for _, c := range d.Changes {
|
||
if c.Operation == "ACTIVATE" {
|
||
t.Fatal("pause activated branch child")
|
||
}
|
||
}
|
||
persist(t, &s, d)
|
||
// 编译之后修改调用方定义和公开副本均不能改变正在解释的固定版本。
|
||
spec.Body.Steps[0].Branch.Cases[0].Body.Steps[0].Task.Handler = "changed"
|
||
copy, ok := p.Step("selected")
|
||
if !ok {
|
||
t.Fatal("missing metadata")
|
||
}
|
||
copy.Task.Handler = "also.changed"
|
||
s.AllowNew = true
|
||
d, err = p.Plan(s, 32)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
found := false
|
||
for _, c := range d.Changes {
|
||
if c.Task != nil {
|
||
found = true
|
||
if c.Task.Handler != "object.prepare" {
|
||
t.Fatal("compiled program changed")
|
||
}
|
||
}
|
||
}
|
||
if !found {
|
||
t.Fatal("frozen branch did not resume")
|
||
}
|
||
// 已保存的默认分支选择即便重新求值会匹配case，也必须继续原选择。
|
||
s = snapshot()
|
||
s.Steps["choice"] = State{Status: weave.StepActive, Input: json.RawMessage(`{"case":-1}`)}
|
||
d, err = p.Plan(s, 32)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
for _, c := range d.Changes {
|
||
if c.Task != nil && c.Key != "unused" {
|
||
t.Fatal("branch was reevaluated")
|
||
}
|
||
}
|
||
}
|
||
func TestWaitAndPollPreparationAndRuntimeLimits(t *testing.T) {
|
||
spec := programSpec()
|
||
str := definition.Type{Kind: "STRING"}
|
||
status := definition.Type{Kind: "STRING", Enum: []string{"PENDING", "DONE"}}
|
||
spec.InputType = str
|
||
signal := definition.Signal{Name: "job.final", PayloadVersion: "v1", Correlation: ref("input"), PayloadType: status, TimeoutMS: 10000}
|
||
query := task("11111111111111111111111111111111", val(`1`))
|
||
query.ResultType = status
|
||
poll := definition.Poll{Task: query, MaxRounds: 2, TimeoutMS: 10000, IntervalMS: 1000, Pending: []string{"PENDING"}, Succeeded: []string{"DONE"}, FinalSignal: &signal}
|
||
spec.Body = block(status, ref("poll"), definition.Step{Key: "signal", Kind: "SIGNAL", Signal: &signal}, definition.Step{Key: "timer", Kind: "TIMER", Timer: &definition.Timer{DelayMS: 1000}}, definition.Step{Key: "poll", Kind: "POLL", Poll: &poll})
|
||
p := compiled(t, spec)
|
||
s := snapshot()
|
||
s.Input = json.RawMessage(`"operation"`)
|
||
d, err := p.Plan(s, 32)
|
||
if err != nil || len(d.Changes) != 1 || d.Changes[0].Wait.Correlation != "operation" || d.Changes[0].Task != nil {
|
||
t.Fatal("signal", d, err)
|
||
}
|
||
persist(t, &s, d)
|
||
again, err := p.Plan(s, 32)
|
||
if err != nil || len(again.Changes) != 0 {
|
||
t.Fatal("wait restarted", err)
|
||
}
|
||
succeed(&s, "signal", json.RawMessage(`"DONE"`))
|
||
d, err = p.Plan(s, 32)
|
||
if err != nil || d.Changes[0].Wait.TimeoutMS != 1000 {
|
||
t.Fatal("timer", err)
|
||
}
|
||
persist(t, &s, d)
|
||
succeed(&s, "timer", json.RawMessage(`null`))
|
||
d, err = p.Plan(s, 32)
|
||
if err != nil || d.Changes[0].Task == nil || d.Changes[0].Wait.Kind != "POLL" {
|
||
t.Fatal("poll", err)
|
||
}
|
||
s = snapshot()
|
||
s.Input = json.RawMessage(`""`)
|
||
if d, err := p.Plan(s, 32); err == nil || len(d.Changes) != 0 {
|
||
t.Fatal("empty runtime correlation accepted")
|
||
}
|
||
s.Input, _ = json.Marshal(strings.Repeat("x", 513))
|
||
if _, err := p.Plan(s, 32); err == nil {
|
||
t.Fatal("unbounded correlation accepted")
|
||
}
|
||
// 同一引用同时用作两个Pulse键，不能因指针相同漏掉一个键。
|
||
shared := ref("key")
|
||
q := task("11111111111111111111111111111111", val(`1`))
|
||
q.BusinessKey = &shared
|
||
q.ConcurrencyKey = &shared
|
||
prepared, err := prepareTask(q, values{"key": json.RawMessage(`"operation"`)})
|
||
if err != nil || prepared.BusinessKey != "operation" || prepared.ConcurrencyKey != "operation" {
|
||
t.Fatal(prepared, err)
|
||
}
|
||
}
|
||
func TestParallelCompletionIntegrityAndConcurrentPlanning(t *testing.T) {
|
||
spec := programSpec()
|
||
p := compiled(t, spec)
|
||
s := snapshot()
|
||
for i := 0; i < 15; i++ {
|
||
d, err := p.Plan(s, 32)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
persist(t, &s, d)
|
||
for _, c := range d.Changes {
|
||
if c.Task != nil {
|
||
succeed(&s, c.Key, c.Task.Payload)
|
||
}
|
||
}
|
||
if d.Complete {
|
||
break
|
||
}
|
||
}
|
||
if !reflect.DeepEqual(s.Steps["outputs"].Status, weave.StepSucceeded) {
|
||
t.Fatal("not completed")
|
||
}
|
||
// 完成的组合不能掩盖未结清子步骤或伪造输出。
|
||
broken := snapshot()
|
||
for k, v := range s.Steps {
|
||
broken.Steps[k] = v
|
||
}
|
||
row := broken.Steps["a"]
|
||
row.Status = weave.StepActive
|
||
row.Output = nil
|
||
broken.Steps["a"] = row
|
||
_, err := p.Plan(broken, 32)
|
||
var issue *Problem
|
||
if !errors.As(err, &issue) || issue.Code != "INCOMPLETE_COMPOSITE_CHILD" {
|
||
t.Fatal("incomplete child hidden", err)
|
||
}
|
||
broken.Steps["a"] = s.Steps["a"]
|
||
row = broken.Steps["outputs"]
|
||
row.Output = json.RawMessage(`{"first":0,"second":0}`)
|
||
broken.Steps["outputs"] = row
|
||
if _, err := p.Plan(broken, 32); err == nil {
|
||
t.Fatal("fabricated composite result accepted")
|
||
}
|
||
var wg sync.WaitGroup
|
||
for i := 0; i < 16; i++ {
|
||
wg.Add(1)
|
||
go func() {
|
||
defer wg.Done()
|
||
d, err := p.Plan(s, 32)
|
||
if err != nil || !d.Complete || len(d.Changes) != 0 {
|
||
t.Error("concurrent immutable plan", err)
|
||
}
|
||
}()
|
||
}
|
||
wg.Wait()
|
||
}
|
||
|
||
func TestUnreachableExecutionsAndAggregateOutputLimit(t *testing.T) {
|
||
p := compiled(t, programSpec())
|
||
s := snapshot()
|
||
s.Steps["outputs"] = State{Status: weave.StepActive, Input: json.RawMessage(`{}`)}
|
||
if d, err := p.Plan(s, 32); err == nil || len(d.Changes) != 0 {
|
||
t.Fatal("future execution was ignored")
|
||
}
|
||
large, _ := json.Marshal(strings.Repeat("x", 140000))
|
||
expr := definition.Value{Kind: "OBJECT", Fields: map[string]definition.Value{"a": ref("value"), "b": ref("value")}}
|
||
if _, err := resolve(expr, definition.Type{Kind: "JSON"}, values{"value": large}); err == nil {
|
||
t.Fatal("aggregate output limit ignored")
|
||
}
|
||
}
|
||
|
||
func TestPollCompletionMustHaveFinalSuccess(t *testing.T) {
|
||
spec := programSpec()
|
||
query := task("11111111111111111111111111111111", val(`1`))
|
||
query.ResultType = definition.Type{Kind: "STRING", Enum: []string{"WORKING", "DONE", "FAILED", "REVIEW"}}
|
||
poll := definition.Poll{Task: query, MaxRounds: 2, TimeoutMS: 10000, IntervalMS: 1000, Pending: []string{"WORKING"}, Succeeded: []string{"DONE"}, Failed: []string{"FAILED"}, Review: []string{"REVIEW"}}
|
||
spec.Body = block(query.ResultType, ref("poll"), definition.Step{Key: "poll", Kind: "POLL", Poll: &poll})
|
||
p := compiled(t, spec)
|
||
for status, want := range map[string]PollOutcome{"WORKING": PollPending, "DONE": PollSucceeded, "FAILED": PollFailed, "REVIEW": PollNeedsReview} {
|
||
raw, _ := json.Marshal(status)
|
||
out, err := p.ClassifyPoll("poll", raw)
|
||
if err != nil || out != want {
|
||
t.Fatal(status, out, err)
|
||
}
|
||
s := snapshot()
|
||
s.Steps["poll"] = State{Status: weave.StepSucceeded, Input: json.RawMessage(`1`), Output: raw}
|
||
decision, err := p.Plan(s, 32)
|
||
if status == "DONE" {
|
||
if err != nil || !decision.Complete {
|
||
t.Fatal("final success rejected", err)
|
||
}
|
||
} else if err == nil || len(decision.Changes) != 0 {
|
||
t.Fatal("nonfinal poll result advanced")
|
||
}
|
||
}
|
||
if _, err := p.ClassifyPoll("poll", json.RawMessage(`"unexpected"`)); err == nil {
|
||
t.Fatal("unknown status accepted")
|
||
}
|
||
}
|