package definition import ( "bytes" "crypto/sha256" "encoding/json" "fmt" "sort" "unicode/utf8" "git.yuntuo.hk/k2safe/orbit/internal/jsonvalue" "git.yuntuo.hk/k2safe/orbit/internal/jsonwire" core "git.yuntuo.hk/k2safe/orbit/model" "git.yuntuo.hk/k2safe/orbit/pulse/service" ) const MaxSpecificationBytes = 256 << 10 const maxTimeoutMS = 30 * 86400000 type checker struct { result Checked keys map[string]bool queues map[core.ID]bool timeout uint64 } type scope map[string]Type // Check 验证完整结构、词法数据依赖和资源边界;不修改调用方定义,也不访问数据库。 // 队列存在性、当前权限和不可变版本身份由发布事务另行核验。 func Check(in Specification) (Checked, error) { if in.SchemaVersion != 1 || in.TimeoutMS == 0 || in.TimeoutMS > maxTimeoutMS || (in.FailurePolicy != "REVIEW" && in.FailurePolicy != "FAIL_AFTER_SETTLED") { return Checked{}, invalid("specification", "schema_version=1, finite timeout and explicit failure policy required") } if err := checkType(in.InputType, 0); err != nil { return Checked{}, err } c := checker{keys: map[string]bool{}, queues: map[core.ID]bool{}, timeout: in.TimeoutMS} initial := scope{"input": in.InputType, "workflow": {Kind: "OBJECT", Fields: map[string]Type{"id": {Kind: "STRING"}, "business_key": {Kind: "STRING"}}}} width, err := c.block(in.Body, initial, 0) if err != nil { return Checked{}, err } if width > 32 || c.result.MaxTaskAttempts > 10000 { return Checked{}, invalid("specification", "parallel or total attempt budget exceeded") } raw, err := jsonwire.Marshal(in) if err != nil || len(raw) > MaxSpecificationBytes { return Checked{}, invalid("specification", "encoded definition exceeds 256 KiB or is invalid") } canonical, err := jsonvalue.Canonical(raw) if err != nil { return Checked{}, invalid("specification", "definition JSON exceeds supported depth or value limits") } c.result.Digest = core.Digest(sha256.Sum256(append([]byte("orbit.weave.spec.v1\x00"), canonical...))) c.result.MaxConcurrency = width for q := range c.queues { c.result.Queues = append(c.result.Queues, q) } sort.Slice(c.result.Queues, func(i, j int) bool { return c.result.Queues[i] < c.result.Queues[j] }) return c.result, nil } func invalid(where, message string) error { return fmt.Errorf("%s: %s", where, message) } func name(s string, limit int) bool { if len(s) == 0 || len(s) > limit { return false } for i, r := range s { if !(r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' || i > 0 && (r == '_' || r == '.' || r == '-')) { return false } } return true } func cloneScope(in scope) scope { out := make(scope, len(in)+8) for key, value := range in { out[key] = value } return out } func (c *checker) block(b Block, parent scope, depth int) (uint32, error) { if depth > 4 || len(b.Steps) > 128 { return 0, invalid("body", "block depth or step limit exceeded") } if err := checkType(b.OutputType, 0); err != nil { return 0, err } env := cloneScope(parent) var width uint32 for _, step := range b.Steps { if !name(step.Key, 64) || step.Key == "input" || step.Key == "workflow" || c.keys[step.Key] { return 0, invalid("step", "key must be unique and nonreserved") } c.keys[step.Key] = true c.result.Steps++ if c.result.Steps > 128 { return 0, invalid("body", "at most 128 static steps allowed") } result, concurrency, err := c.step(step, env, depth) if err != nil { return 0, invalid(step.Key, err.Error()) } env[step.Key] = result if concurrency > width { width = concurrency } } if err := checkValue(b.Output, b.OutputType, env, 0); err != nil { return 0, invalid("output", err.Error()) } return width, nil } func (c *checker) step(s Step, env scope, depth int) (Type, uint32, error) { n := 0 for _, present := range []bool{s.Task != nil, s.Branch != nil, s.Parallel != nil, s.Signal != nil, s.Timer != nil, s.Poll != nil} { if present { n++ } } if n != 1 { return Type{}, 0, invalid("step", "exactly one kind configuration required") } switch s.Kind { case "TASK": if s.Task != nil { if err := c.task(*s.Task, env); err != nil { return Type{}, 0, err } c.result.MaxTaskAttempts += s.Task.Retry.MaxAttempts return s.Task.ResultType, 1, nil } case "SIGNAL": if s.Signal != nil { return s.Signal.PayloadType, 0, c.signal(*s.Signal, env) } case "TIMER": if s.Timer != nil && s.Timer.DelayMS > 0 && s.Timer.DelayMS <= c.timeout { return Type{Kind: "NULL"}, 0, nil } case "BRANCH": if s.Branch != nil { b := s.Branch if b.Default == nil || len(b.Cases) < 1 || len(b.Cases) > 16 { return Type{}, 0, invalid("branch", "1 to 16 cases and explicit default required") } width, err := c.block(*b.Default, env, depth+1) if err != nil { return Type{}, 0, err } for _, path := range b.Cases { if err := checkPredicate(path.When, env, 0); err != nil { return Type{}, 0, err } w, err := c.block(path.Body, env, depth+1) if err != nil { return Type{}, 0, err } if !sameType(path.Body.OutputType, b.Default.OutputType) { return Type{}, 0, invalid("branch", "all path output types must match") } if w > width { width = w } } return b.Default.OutputType, width, nil } case "PARALLEL": if s.Parallel != nil { if len(s.Parallel.Branches) < 2 || len(s.Parallel.Branches) > 8 { return Type{}, 0, invalid("parallel", "2 to 8 named branches required") } result := Type{Kind: "OBJECT", Fields: map[string]Type{}} var width uint32 for _, branch := range s.Parallel.Branches { if !name(branch.Name, 64) { return Type{}, 0, invalid("parallel", "invalid branch name") } if _, exists := result.Fields[branch.Name]; exists { return Type{}, 0, invalid("parallel", "duplicate branch name") } w, err := c.block(branch.Body, env, depth+1) if err != nil { return Type{}, 0, err } width += w result.Fields[branch.Name] = branch.Body.OutputType } if width > 32 { return Type{}, 0, invalid("parallel", "at most 32 concurrent business tasks allowed") } return result, width, nil } case "POLL": if s.Poll != nil { p := s.Poll if p.MaxRounds == 0 || p.MaxRounds > 1000 || p.TimeoutMS == 0 || p.TimeoutMS > c.timeout || p.IntervalMS < 1000 || p.IntervalMS >= p.TimeoutMS || p.Task.TimeoutMS > p.TimeoutMS { return Type{}, 0, invalid("poll", "finite rounds, interval and compatible time budgets required") } if err := c.task(p.Task, env); err != nil { return Type{}, 0, err } status, err := pathType(p.Task.ResultType, p.StatusPath) if err != nil || status.Kind != "STRING" || len(status.Enum) == 0 || len(p.Pending) == 0 || len(p.Succeeded) == 0 { return Type{}, 0, invalid("poll", "status must be a declared string enum with pending and success cases") } seen := map[string]bool{} for _, group := range [][]string{p.Pending, p.Succeeded, p.Failed, p.Review} { for _, v := range group { if !contains(status.Enum, v) || seen[v] { return Type{}, 0, invalid("poll", "status groups must form an exact disjoint partition") } seen[v] = true } } if len(seen) != len(status.Enum) { return Type{}, 0, invalid("poll", "unmapped status enum value") } if p.FinalSignal != nil { if err := c.signal(*p.FinalSignal, env); err != nil { return Type{}, 0, err } if p.FinalSignal.TimeoutMS != p.TimeoutMS || !sameType(p.FinalSignal.PayloadType, p.Task.ResultType) { return Type{}, 0, invalid("poll", "final signal must share result type and poll deadline") } } c.result.MaxTaskAttempts += p.MaxRounds * p.Task.Retry.MaxAttempts return p.Task.ResultType, 1, nil } } return Type{}, 0, invalid("step", "unknown kind, mismatched configuration or invalid timer") } func (c *checker) task(t Task, env scope) error { if t.TimeoutMS == 0 || t.TimeoutMS > c.timeout || t.InputType.Kind == "NULL" || t.ResultType.Kind == "NULL" { return invalid("task", "finite timeout and non-null payload/result types required") } // 复用Pulse的处理器/策略边界,但定义必须显式提供生效值,不能在后续发布时解释不同默认值。 v, err := service.NormalizeSubmit(service.SubmitInput{QueueID: t.QueueID, Handler: t.Handler, HandlerVersion: t.HandlerVersion, PayloadVersion: t.PayloadVersion, Payload: json.RawMessage(`{}`), TaskTimeoutMS: t.TimeoutMS, RecoverySafety: t.RecoverySafety, Retry: t.Retry, Priority: t.Priority}) if err != nil || v.Retry != t.Retry || v.RecoverySafety != t.RecoverySafety { return invalid("task", "invalid or incomplete frozen Pulse policy") } if err := checkType(t.InputType, 0); err != nil { return err } if err := checkType(t.ResultType, 0); err != nil { return err } if err := checkValue(t.Input, t.InputType, env, 0); err != nil { return err } if t.Input.Kind == "LITERAL" && bytes.Equal(bytes.TrimSpace(t.Input.Literal), []byte("null")) { return invalid("task", "null task input is not accepted by Pulse") } for _, key := range []*Value{t.BusinessKey, t.ConcurrencyKey} { if key != nil { if err := checkValue(*key, Type{Kind: "STRING"}, env, 0); err != nil { return err } } } c.queues[t.QueueID] = true return nil } func (c *checker) signal(s Signal, env scope) error { if !name(s.Name, 128) || !name(s.PayloadVersion, 32) || s.TimeoutMS == 0 || s.TimeoutMS > c.timeout { return invalid("signal", "name, payload version and finite timeout required") } if err := checkType(s.PayloadType, 0); err != nil { return err } if err := checkValue(s.Correlation, Type{Kind: "STRING"}, env, 0); err != nil { return err } if s.Correlation.Kind == "LITERAL" { var value string if json.Unmarshal(s.Correlation.Literal, &value) != nil || len(value) == 0 || len(value) > 512 { return invalid("signal", "literal correlation must be a nonempty string up to 512 bytes") } } return nil } func checkType(t Type, depth int) error { if depth > 8 || len(t.Fields) > 64 || len(t.Enum) > 32 { return invalid("type", "schema depth or width limit exceeded") } if t.Kind != "OBJECT" && len(t.Fields) != 0 || t.Kind != "ARRAY" && t.Items != nil || t.Kind != "STRING" && len(t.Enum) != 0 { return invalid("type", "fields, items or enum do not match kind") } switch t.Kind { case "JSON", "NUMBER", "BOOLEAN", "NULL": case "STRING": seen := map[string]bool{} for _, value := range t.Enum { if !utf8.ValidString(value) || len(value) > 128 || seen[value] { return invalid("type", "invalid or duplicate string enum value") } seen[value] = true } case "OBJECT": for key, field := range t.Fields { if !name(key, 64) { return invalid("type", "invalid object field name") } if err := checkType(field, depth+1); err != nil { return err } } case "ARRAY": if t.Items == nil { return invalid("type", "array items type required") } return checkType(*t.Items, depth+1) default: return invalid("type", "unknown kind") } return nil } func sameType(a, b Type) bool { return assignable(a, b) && assignable(b, a) } func assignable(actual, expected Type) bool { if expected.Kind == "JSON" { return true } if actual.Kind != expected.Kind { return false } switch expected.Kind { case "OBJECT": if len(actual.Fields) != len(expected.Fields) { return false } for key, t := range expected.Fields { v, ok := actual.Fields[key] if !ok || !assignable(v, t) { return false } } case "ARRAY": return actual.Items != nil && expected.Items != nil && assignable(*actual.Items, *expected.Items) case "STRING": if len(expected.Enum) == 0 { return true } if len(actual.Enum) == 0 { return false } for _, v := range actual.Enum { if !contains(expected.Enum, v) { return false } } } return true } func contains(values []string, value string) bool { for _, v := range values { if v == value { return true } } return false } func pathType(t Type, path []string) (Type, error) { if len(path) > 8 { return Type{}, invalid("reference", "path exceeds 8 fields") } for _, key := range path { if t.Kind != "OBJECT" { return Type{}, invalid("reference", "path crosses a non-object or untyped JSON value") } var ok bool t, ok = t.Fields[key] if !ok { return Type{}, invalid("reference", "field not declared") } } return t, nil } func referenceType(r Reference, env scope) (Type, error) { t, ok := env[r.Source] if !ok { return Type{}, invalid("reference", "source is missing, forward, conditional or outside this block") } return pathType(t, r.Path) } func checkValue(v Value, expected Type, env scope, depth int) error { if depth > 8 { return invalid("value", "expression depth exceeds 8") } if v.Kind != "LITERAL" && len(v.Literal) != 0 || v.Kind != "REFERENCE" && v.Reference != nil || v.Kind != "OBJECT" && len(v.Fields) != 0 || v.Kind != "ARRAY" && len(v.Items) != 0 { return invalid("value", "configuration does not match kind") } switch v.Kind { case "LITERAL": return ValidateData(expected, v.Literal) case "REFERENCE": if v.Reference == nil { return invalid("reference", "source required") } actual, err := referenceType(*v.Reference, env) if err != nil { return err } if !assignable(actual, expected) { return invalid("reference", "source and target types are incompatible") } case "OBJECT": if expected.Kind != "OBJECT" && expected.Kind != "JSON" || len(v.Fields) > 64 { return invalid("value", "object expression has incompatible type or too many fields") } if expected.Kind == "OBJECT" && len(v.Fields) != len(expected.Fields) { return invalid("value", "object expression must define every field exactly") } for key, child := range v.Fields { if !name(key, 64) { return invalid("value", "invalid object field") } t := Type{Kind: "JSON"} if expected.Kind == "OBJECT" { var ok bool t, ok = expected.Fields[key] if !ok { return invalid("value", "extra object field") } } if err := checkValue(child, t, env, depth+1); err != nil { return err } } case "ARRAY": if expected.Kind != "ARRAY" && expected.Kind != "JSON" || len(v.Items) > 256 { return invalid("value", "array expression has incompatible type or too many elements") } t := Type{Kind: "JSON"} if expected.Kind == "ARRAY" { t = *expected.Items } for _, child := range v.Items { if err := checkValue(child, t, env, depth+1); err != nil { return err } } default: return invalid("value", "unknown kind") } return nil } func scalarType(v Value, env scope) (string, error) { if err := checkValue(v, Type{Kind: "JSON"}, env, 0); err != nil { return "", err } var kind string switch v.Kind { case "REFERENCE": t, err := referenceType(*v.Reference, env) if err != nil { return "", err } kind = t.Kind case "LITERAL": var raw any d := json.NewDecoder(bytes.NewReader(v.Literal)) d.UseNumber() if err := d.Decode(&raw); err != nil { return "", err } switch raw.(type) { case nil: kind = "NULL" case string: kind = "STRING" case bool: kind = "BOOLEAN" case json.Number: kind = "NUMBER" } } if kind != "STRING" && kind != "NUMBER" && kind != "BOOLEAN" && kind != "NULL" { return "", invalid("predicate", "only typed scalar comparisons allowed") } return kind, nil } func checkPredicate(p Predicate, env scope, depth int) error { if depth > 8 { return invalid("predicate", "expression depth exceeds 8") } switch p.Op { case "AND", "OR", "NOT": if p.Left != nil || p.Right != nil || len(p.Arguments) > 16 || (p.Op == "NOT" && len(p.Arguments) != 1) || (p.Op != "NOT" && len(p.Arguments) < 2) { return invalid("predicate", "invalid logical arguments") } for _, child := range p.Arguments { if err := checkPredicate(child, env, depth+1); err != nil { return err } } case "EQ", "NE", "LT", "LE", "GT", "GE": if p.Left == nil || p.Right == nil || len(p.Arguments) != 0 { return invalid("predicate", "exactly two comparison operands required") } left, err := scalarType(*p.Left, env) if err != nil { return err } right, err := scalarType(*p.Right, env) if err != nil { return err } if left != right || (p.Op != "EQ" && p.Op != "NE" && left != "STRING" && left != "NUMBER") { return invalid("predicate", "incompatible comparison operands") } default: return invalid("predicate", "unknown operator") } return nil }