k2safe/orbit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: expose workflow runtime through HTTP and Go SDK

Browse Source
This commit is contained in:
k2safe
2026-09-18 06:51:08 +08:00
parent af1ae4cb21
commit b9e4191c6b
50 changed files with 2221 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
internal/storage/mysql/bootstrap.go
View File

@@ -14,17 +14,22 @@ import (
// BootstrapInput 仅供持有数据库管理权限的初始化命令使用，不暴露为公共 HTTP。
type BootstrapInput struct {
WorkflowAccess bool // WorkflowAccess 显式授予流程定义读取和发布权限，不含实例执行权限。
RecoveryAccess bool // RecoveryAccess 显式授予任务核查、人工重试及操作审计读取权限。
EventAccess bool // EventAccess 显式增加结果订阅查询、管理和消费权限。
Application string // Application 新应用名称，与 ExistingApplicationID 二选一。
Namespace string // Namespace 命名空间名称。
Environment string // Environment 环境名称，必须显式指定。
Queue string // Queue 初始队列名称。
WorkerAccess bool // WorkerAccess 显式增加开发用 Worker 接入及消费权限。
ScheduleAccess bool // ScheduleAccess 显式增加开发用计划查询及管理权限。
ResourceAccess bool // ResourceAccess 显式授予整个应用的额度查询及管理权限。
ExistingApplicationID core.ID // ExistingApplicationID 在指定现有应用下创建命名空间，与 Application 二选一。
WorkflowAuditAccess bool // WorkflowAuditAccess 显式授予流程操作审计读取；不授予其他管理操作。
WorkflowRunAccess bool // WorkflowRunAccess 显式授予流程启动和读取，不包含发布或管理。
WorkflowControlAccess bool // WorkflowControlAccess 显式授予流程暂停、继续和取消及读取。
WorkflowSignalAccess bool // WorkflowSignalAccess 显式授予流程信号受理及读取。
WorkflowRecoveryAccess bool // WorkflowRecoveryAccess 显式授予有证据的流程人工恢复及读取。
WorkflowAccess bool // WorkflowAccess 显式授予流程定义读取和发布权限，不含实例执行权限。
RecoveryAccess bool // RecoveryAccess 显式授予任务核查、人工重试及操作审计读取权限。
EventAccess bool // EventAccess 显式增加结果订阅查询、管理和消费权限。
Application string // Application 新应用名称，与 ExistingApplicationID 二选一。
Namespace string // Namespace 命名空间名称。
Environment string // Environment 环境名称，必须显式指定。
Queue string // Queue 初始队列名称。
WorkerAccess bool // WorkerAccess 显式增加开发用 Worker 接入及消费权限。
ScheduleAccess bool // ScheduleAccess 显式增加开发用计划查询及管理权限。
ResourceAccess bool // ResourceAccess 显式授予整个应用的额度查询及管理权限。
ExistingApplicationID core.ID // ExistingApplicationID 在指定现有应用下创建命名空间，与 Application 二选一。
}
// BootstrapResult 包含一次性交付的原始凭据，禁止记录到日志或提交到版本库。
@@ -80,6 +85,24 @@ func Bootstrap(ctx context.Context, db *sql.DB, in BootstrapInput, deliver func(
if in.WorkflowAccess {
permissions.Actions = append(permissions.Actions, "workflows:definitions:read", "workflows:definitions:publish")
}
if in.WorkflowAuditAccess {
permissions.Actions = append(permissions.Actions, "workflows:audit:read")
}
if in.WorkflowRunAccess || in.WorkflowControlAccess || in.WorkflowSignalAccess || in.WorkflowRecoveryAccess {
permissions.Actions = append(permissions.Actions, "workflows:read")
}
if in.WorkflowRunAccess {
permissions.Actions = append(permissions.Actions, "workflows:start")
}
if in.WorkflowControlAccess {
permissions.Actions = append(permissions.Actions, "workflows:control")
}
if in.WorkflowSignalAccess {
permissions.Actions = append(permissions.Actions, "workflows:signal")
}
if in.WorkflowRecoveryAccess {
permissions.Actions = append(permissions.Actions, "workflows:recover")
}
if in.ScheduleAccess {
permissions.Actions = append(permissions.Actions, "schedules:read", "schedules:manage")
}
@@ -92,6 +115,9 @@ func Bootstrap(ctx context.Context, db *sql.DB, in BootstrapInput, deliver func(
if in.RecoveryAccess {
permissions.Actions = append(permissions.Actions, "tasks:review", "tasks:retry", "tasks:audit:read")
}
if err := permissions.Validate(); err != nil {
return err
}
raw, err := json.Marshal(permissions)
if err != nil {
return err

1
internal/storage/mysql/workflow_apply.go
View File

@@ -214,6 +214,7 @@ func activateWorkflowStep(ctx context.Context, tx *sql.Tx, g *workflowGraph, row
return err
}
row.view.ActivationID = &id
row.view.ActivationSequence = 1
row.view.TaskID = direct
deadline := g.view.DeadlineAt
if c.Kind == "POLL" {

68
internal/storage/mysql/workflow_bootstrap_integration_test.go Normal file
View File

@@ -0,0 +1,68 @@
package mysql
import (
"context"
"encoding/json"
"git.yuntuo.hk/k2safe/orbit/platform/auth"
"os"
"testing"
"time"
)
func TestIntegrationWorkflowBootstrapScopes(t *testing.T) {
dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN")
if dsn == "" {
t.Skip("requires disposable MySQL")
}
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
db, err := Open(ctx, dsn, 8, 2)
if err != nil {
t.Fatal(err)
}
defer db.Close()
if err = Migrate(ctx, db); err != nil {
t.Fatal(err)
}
for _, tc := range []struct {
name string
in BootstrapInput
actions []string
}{
{"definitions", BootstrapInput{WorkflowAccess: true}, []string{"workflows:definitions:read", "workflows:definitions:publish"}},
{"run", BootstrapInput{WorkflowRunAccess: true}, []string{"workflows:start", "workflows:read"}},
{"control", BootstrapInput{WorkflowControlAccess: true}, []string{"workflows:control", "workflows:read"}},
{"signal", BootstrapInput{WorkflowSignalAccess: true}, []string{"workflows:signal", "workflows:read"}},
{"recover", BootstrapInput{WorkflowRecoveryAccess: true}, []string{"workflows:recover", "workflows:read"}},
{"audit", BootstrapInput{WorkflowAuditAccess: true}, []string{"workflows:audit:read"}},
} {
t.Run(tc.name, func(t *testing.T) {
in := tc.in
in.Application = "scopes-" + string(executionID(t))
in.Namespace = "default"
in.Environment = "test"
in.Queue = "work"
var out BootstrapResult
if err := Bootstrap(ctx, db, in, func(v BootstrapResult) error { out = v; return nil }); err != nil {
t.Fatal(err)
}
var raw []byte
if err := db.QueryRowContext(ctx, `SELECT permissions FROM orbit_credentials WHERE namespace_id=? AND id=?`, out.NamespaceID, out.CredentialID).Scan(&raw); err != nil {
t.Fatal(err)
}
var p auth.Permissions
if json.Unmarshal(raw, &p) != nil || p.Validate() != nil {
t.Fatal("bootstrap issued unusable scope")
}
expected := map[string]bool{}
for _, v := range tc.actions {
expected[v] = true
}
for _, v := range []string{"workflows:definitions:read", "workflows:definitions:publish", "workflows:start", "workflows:read", "workflows:control", "workflows:signal", "workflows:recover", "workflows:audit:read"} {
if p.Allows(v) != expected[v] {
t.Fatal("scope widened or missing", v)
}
}
})
}
}

79
internal/storage/mysql/workflow_details.go Normal file
View File

@@ -0,0 +1,79 @@
package mysql
import (
"context"
"database/sql"
"errors"
"git.yuntuo.hk/k2safe/orbit/internal/fault"
core "git.yuntuo.hk/k2safe/orbit/model"
"git.yuntuo.hk/k2safe/orbit/platform/auth"
"git.yuntuo.hk/k2safe/orbit/weave/engine"
"git.yuntuo.hk/k2safe/orbit/weave/instance"
)
func (s *WorkflowInstanceStore) Step(ctx context.Context, t auth.Token, id core.ID, key string) (instance.StepDetail, error) {
out := instance.StepDetail{InstanceID: id}
err := s.transaction(ctx, t, instance.ReadAction, func(tx *sql.Tx, p auth.Permissions) error {
v, gate, version, err := readWorkflowForQuery(ctx, tx, t.NamespaceID, id, p)
if err != nil {
return err
}
out.Revision = v.Revision
steps, err := readWorkflowStepViews(ctx, tx, t.NamespaceID, id, gate, version, key)
if err != nil {
return err
}
if len(steps) != 1 {
return fault.NotFound
}
out.Step = steps[0]
a := out.Step.ActivationID
if a == nil {
return nil
}
var input, output []byte
if err = tx.QueryRowContext(ctx, `SELECT input,output,created_at,completed_at FROM weave_activations WHERE namespace_id=? AND workflow_id=? AND step_id=? AND id=?`, t.NamespaceID, id, out.Step.ID, a).Scan(&input, &output, &out.ActivatedAt, &out.CompletedAt); err != nil {
return err
}
out.Input, out.Output = input, output
program, err := engine.Compile(version.Specification)
if err != nil {
return err
}
node, _ := program.Step(key)
if node.Poll != nil {
q := &instance.PollView{}
var owned, queue *core.ID
err = tx.QueryRowContext(ctx, "SELECT p.used_rounds,p.first_round,p.first_round_attempts_used,p.max_rounds,p.max_attempts_per_round,p.interval_ms,p.deadline_at,p.next_poll_at,p.current_task_id,r.task_id,p.resolved_at,t.id,t.queue_id FROM weave_poll_states p JOIN weave_poll_rounds r ON r.namespace_id=p.namespace_id AND r.workflow_id=p.workflow_id AND r.activation_id=p.activation_id AND r.`round`=p.used_rounds LEFT JOIN pulse_tasks t ON t.namespace_id=r.namespace_id AND t.id=r.task_id AND t.owner_kind='workflow_activation' AND t.owner_id=r.activation_id AND t.gate_id=? WHERE p.namespace_id=? AND p.workflow_id=? AND p.activation_id=?", gate, t.NamespaceID, id, a).Scan(&q.UsedRounds, &q.FirstRound, &q.FirstRoundAttemptsUsed, &q.MaxRounds, &q.MaxAttemptsPerRound, &q.IntervalMS, &q.DeadlineAt, &q.NextPollAt, &q.CurrentTaskID, &q.LastTaskID, &q.ResolvedAt, &owned, &queue)
if err != nil {
return err
}
if owned == nil || *owned != q.LastTaskID || queue == nil || *queue != node.Poll.Task.QueueID || q.CurrentTaskID != nil && *q.CurrentTaskID != q.LastTaskID || q.MaxRounds != node.Poll.MaxRounds || q.MaxAttemptsPerRound != node.Poll.Task.Retry.MaxAttempts || q.IntervalMS != node.Poll.IntervalMS {
return errors.New("workflow poll detail ownership or policy invalid")
}
out.Poll = q
}
if node.Signal != nil || node.Timer != nil || node.Poll != nil && node.Poll.FinalSignal != nil {
w := &instance.WaitView{}
if err = tx.QueryRowContext(ctx, `SELECT id,kind,state,signal_name,payload_version,correlation_key,deadline_at,winner_signal_id,resolved_at FROM weave_waits WHERE namespace_id=? AND workflow_id=? AND activation_id=?`, t.NamespaceID, id, a).Scan(&w.ID, &w.Kind, &w.State, &w.Name, &w.PayloadVersion, &w.CorrelationKey, &w.DeadlineAt, &w.WinnerSignalID, &w.ResolvedAt); err != nil {
return err
}
if w.Kind != string(out.Step.Kind) {
return errors.New("workflow wait kind mismatch")
}
if w.WinnerSignalID != nil {
var n int
if err = tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM weave_signals WHERE namespace_id=? AND workflow_id=? AND id=? AND consumed_wait_id=? AND target_step_key=? AND target_activation_sequence=? AND state='CONSUMED'`, t.NamespaceID, id, w.WinnerSignalID, w.ID, key, out.Step.ActivationSequence).Scan(&n); err != nil {
return err
}
if n != 1 {
return errors.New("workflow wait winner association invalid")
}
}
out.Wait = w
}
return nil
})
return out, err
}

102
internal/storage/mysql/workflow_details_integration_test.go Normal file
View File

@@ -0,0 +1,102 @@
package mysql
import (
"context"
"os"
"reflect"
"testing"
"time"
"git.yuntuo.hk/k2safe/orbit/internal/fault"
"git.yuntuo.hk/k2safe/orbit/pulse/service"
"git.yuntuo.hk/k2safe/orbit/weave/instance"
)
func TestIntegrationWorkflowPublicDetails(t *testing.T) {
dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN")
if dsn == "" {
t.Skip("requires disposable MySQL")
}
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
db, err := Open(ctx, dsn, 16, 4)
if err != nil {
t.Fatal(err)
}
defer db.Close()
if err = Migrate(ctx, db); err != nil {
t.Fatal(err)
}
f := newExecutionFixture(t, ctx, db)
grantWorkflowInstance(t, f)
api := instance.New(NewWorkflowInstanceStore(db))
input := publishInstanceSpec(t, f, allInitialNodes(f.resources.QueueID))
started, err := api.Start(ctx, f.token, "public-details", input)
if err != nil {
t.Fatal(err)
}
id := started.Instance.ID
if _, err = api.ListOperations(ctx, f.token, id, service.HistoryPageInput{}); fault.Code(err) != "PERMISSION_DENIED" {
t.Fatal("ordinary workflow read gained audit", err)
}
f.exec(`UPDATE orbit_credentials SET permissions=JSON_ARRAY_APPEND(permissions,'$.actions','workflows:audit:read') WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, f.resources.CredentialID)
history, err := api.ListOperations(ctx, f.token, id, service.HistoryPageInput{Limit: 1})
if err != nil || len(history.Items) != 1 {
t.Fatal("workflow audit", err)
}
audit, err := api.GetOperation(ctx, f.token, id, history.Items[0].ID)
if err != nil || audit.WorkflowID != id || audit.Action != "workflow.start" || audit.ActorID != string(f.resources.CredentialID) {
t.Fatal("audit scope", err)
}
list, err := api.Steps(ctx, f.token, id)
if err != nil {
t.Fatal(err)
}
for _, meta := range list.Steps {
detail, err := api.Step(ctx, f.token, id, meta.Key)
if err != nil {
t.Fatal(meta.Key, err)
}
if detail.Revision != list.Revision || !reflect.DeepEqual(detail.Step, meta) {
t.Fatal("inconsistent snapshot", meta.Key)
}
if meta.ActivationID != nil && (meta.ActivationSequence != 1 || detail.ActivatedAt == nil || len(detail.Input) == 0) {
t.Fatal("activation not described", meta.Key)
}
if meta.ActivationID == nil && (meta.ActivationSequence != 0 || len(detail.Input) != 0) {
t.Fatal("inactive step fabricated activation")
}
if meta.Kind == "POLL" && (detail.Poll == nil || detail.Poll.LastTaskID == "" || detail.Poll.CurrentTaskID == nil || *detail.Poll.CurrentTaskID != detail.Poll.LastTaskID || detail.Wait == nil) {
t.Fatal("poll source missing")
}
}
if _, err = api.Step(ctx, f.token, id, "missing"); fault.Code(err) != "NOT_FOUND" {
t.Fatal("missing step", err)
}
if _, err = api.Step(ctx, f.token, id, "$root"); fault.Code(err) != "INVALID_ARGUMENT" {
t.Fatal("internal root exposed", err)
}
poll, err := api.Step(ctx, f.token, id, "query")
if err != nil {
t.Fatal(err)
}
f.exec(`UPDATE pulse_tasks SET owner_id=? WHERE namespace_id=? AND id=?`, executionID(t), f.resources.NamespaceID, poll.Poll.LastTaskID)
if _, err = api.Step(ctx, f.token, id, "query"); err == nil {
t.Fatal("orphan poll task accepted without foreign key")
}
f.exec(`UPDATE pulse_tasks SET owner_id=? WHERE namespace_id=? AND id=?`, poll.Step.ActivationID, f.resources.NamespaceID, poll.Poll.LastTaskID)
grantWorkflowInstance(t, f, string(executionID(t)))
if _, err = api.Step(ctx, f.token, id, "query"); fault.Code(err) != "NOT_FOUND" {
t.Fatal("queue restriction bypassed", err)
}
other := newExecutionFixture(t, ctx, db)
grantWorkflowInstance(t, other)
other.exec(`UPDATE orbit_credentials SET permissions=JSON_ARRAY_APPEND(permissions,'$.actions','workflows:audit:read') WHERE namespace_id=? AND id=?`, other.resources.NamespaceID, other.resources.CredentialID)
if _, err = api.GetOperation(ctx, other.token, id, audit.ID); fault.Code(err) != "NOT_FOUND" {
t.Fatal("cross-namespace audit", err)
}
if _, err = api.Step(ctx, other.token, id, "query"); fault.Code(err) != "NOT_FOUND" {
t.Fatal("namespace isolation", err)
}
}

120
internal/storage/mysql/workflow_instances.go
View File

@@ -17,7 +17,7 @@ import (
)
// WorkflowInstanceStore 以应用额度锁串行受理，在一个事务内建立全部归属及初始工作。
// 当前为内部存储适配层，公共HTTP和SDK在消费运行器接通后开放。
// 公共HTTP与内部运行器复用同一套固定版本、幂等与事务规则。
type WorkflowInstanceStore struct{ *TaskStore }
func NewWorkflowInstanceStore(db *sql.DB) *WorkflowInstanceStore {
@@ -226,67 +226,77 @@ func (s *WorkflowInstanceStore) Get(ctx context.Context, t auth.Token, id core.I
})
return out, err
}
// readWorkflowForQuery 固定定义授权全部队列后再锁实例，保持与运行器相同的父子锁顺序。
func readWorkflowForQuery(ctx context.Context, tx *sql.Tx, ns, id core.ID, p auth.Permissions) (instance.View, core.ID, definition.VersionView, error) {
v, gate, err := readWorkflowInstance(ctx, tx, ns, id, false)
var version definition.VersionView
if err != nil {
return v, gate, version, err
}
version, err = instanceVersion(ctx, tx, v)
if err != nil {
return v, gate, version, err
}
if err = authorizeWorkflowQueues(ctx, tx, ns, p, version.Analysis.Queues); err != nil {
return v, gate, version, err
}
v, gate, err = readWorkflowInstance(ctx, tx, ns, id, true)
return v, gate, version, err
}
func (s *WorkflowInstanceStore) Steps(ctx context.Context, t auth.Token, id core.ID) (instance.StepsView, error) {
out := instance.StepsView{InstanceID: id, Steps: []instance.StepView{}}
err := s.transaction(ctx, t, instance.ReadAction, func(tx *sql.Tx, p auth.Permissions) error {
// 先核对队列，后锁实例；未来运行器遵守应用→队列→实例，读取不得逆序锁队列。
v, _, err := readWorkflowInstance(ctx, tx, t.NamespaceID, id, false)
if err != nil {
return err
}
version, err := instanceVersion(ctx, tx, v)
if err != nil {
return err
}
if err := authorizeWorkflowQueues(ctx, tx, t.NamespaceID, p, version.Analysis.Queues); err != nil {
return err
}
var gate core.ID
v, gate, err = readWorkflowInstance(ctx, tx, t.NamespaceID, id, true)
v, gate, version, err := readWorkflowForQuery(ctx, tx, t.NamespaceID, id, p)
if err != nil {
return err
}
out.Revision = v.Revision
program, err := engine.Compile(version.Specification)
if err != nil {
return err
}
rows, err := tx.QueryContext(ctx, "SELECT s.id,s.`key`,s.kind,s.state,s.revision,s.current_activation_id,a.task_id,a.id,a.state,task.id,task.queue_id FROM weave_steps s LEFT JOIN weave_activations a ON a.namespace_id=s.namespace_id AND a.id=s.current_activation_id AND a.workflow_id=s.workflow_id AND a.step_id=s.id LEFT JOIN pulse_tasks task ON task.namespace_id=s.namespace_id AND task.id=a.task_id AND task.owner_kind='workflow_activation' AND task.owner_id=a.id AND task.gate_id=? WHERE s.namespace_id=? AND s.workflow_id=? AND s.`key`<>? ORDER BY s.`key` LIMIT 129", gate, t.NamespaceID, id, workflowRootKey)
if err != nil {
return err
}
defer rows.Close()
for rows.Next() {
var step instance.StepView
var activation *core.ID
var task, queue *core.ID
var state sql.NullString
if err := rows.Scan(&step.ID, &step.Key, &step.Kind, &step.State, &step.Revision, &step.ActivationID, &step.TaskID, &activation, &state, &task, &queue); err != nil {
return err
}
node, ok := program.Step(step.Key)
if !ok || node.Kind != string(step.Kind) {
return errors.New("workflow step definition mismatch")
}
inactive := step.State == "BLOCKED" || step.State == "SKIPPED"
if inactive && step.ActivationID != nil || !inactive && (step.ActivationID == nil || activation == nil || *activation != *step.ActivationID || state.String != string(step.State)) {
return errors.New("workflow current activation mismatch")
}
if !inactive && (step.Kind == "TASK") != (step.TaskID != nil) {
return errors.New("workflow direct task ownership mismatch")
}
if step.TaskID != nil && (task == nil || queue == nil || *task != *step.TaskID || node.Task == nil || *queue != node.Task.QueueID) {
return errors.New("workflow task association invalid")
}
out.Steps = append(out.Steps, step)
}
if err := rows.Err(); err != nil {
return err
}
if len(out.Steps) != int(version.Analysis.Steps) {
return errors.New("workflow step set incomplete")
}
return nil
out.Steps, err = readWorkflowStepViews(ctx, tx, t.NamespaceID, id, gate, version, "")
return err
})
return out, err
}
func readWorkflowStepViews(ctx context.Context, tx *sql.Tx, ns, id, gate core.ID, version definition.VersionView, key string) ([]instance.StepView, error) {
items := []instance.StepView{}
program, err := engine.Compile(version.Specification)
if err != nil {
return nil, err
}
rows, err := tx.QueryContext(ctx, "SELECT s.id,s.`key`,s.kind,s.state,s.revision,s.current_activation_id,a.task_id,COALESCE(a.sequence,0),a.id,a.state,task.id,task.queue_id FROM weave_steps s LEFT JOIN weave_activations a ON a.namespace_id=s.namespace_id AND a.id=s.current_activation_id AND a.workflow_id=s.workflow_id AND a.step_id=s.id LEFT JOIN pulse_tasks task ON task.namespace_id=s.namespace_id AND task.id=a.task_id AND task.owner_kind='workflow_activation' AND task.owner_id=a.id AND task.gate_id=? WHERE s.namespace_id=? AND s.workflow_id=? AND s.`key`<>? AND (?='' OR s.`key`=?) ORDER BY s.`key` LIMIT 129", gate, ns, id, workflowRootKey, key, key)
if err != nil {
return nil, err
}
defer rows.Close()
for rows.Next() {
var step instance.StepView
var activation *core.ID
var task, queue *core.ID
var state sql.NullString
if err := rows.Scan(&step.ID, &step.Key, &step.Kind, &step.State, &step.Revision, &step.ActivationID, &step.TaskID, &step.ActivationSequence, &activation, &state, &task, &queue); err != nil {
return nil, err
}
node, ok := program.Step(step.Key)
if !ok || node.Kind != string(step.Kind) {
return nil, errors.New("workflow step definition mismatch")
}
inactive := step.State == "BLOCKED" || step.State == "SKIPPED"
if inactive && step.ActivationID != nil || !inactive && (step.ActivationSequence == 0 || step.ActivationID == nil || activation == nil || *activation != *step.ActivationID || state.String != string(step.State)) {
return nil, errors.New("workflow current activation mismatch")
}
if !inactive && (step.Kind == "TASK") != (step.TaskID != nil) {
return nil, errors.New("workflow direct task ownership mismatch")
}
if step.TaskID != nil && (task == nil || queue == nil || *task != *step.TaskID || node.Task == nil || *queue != node.Task.QueueID) {
return nil, errors.New("workflow task association invalid")
}
items = append(items, step)
}
if err := rows.Err(); err != nil {
return nil, err
}
if key == "" && len(items) != int(version.Analysis.Steps) {
return nil, errors.New("workflow step set incomplete")
}
return items, nil
}

80
internal/storage/mysql/workflow_operations.go Normal file
View File

@@ -0,0 +1,80 @@
package mysql
import (
"context"
"database/sql"
"errors"
"git.yuntuo.hk/k2safe/orbit/internal/fault"
core "git.yuntuo.hk/k2safe/orbit/model"
"git.yuntuo.hk/k2safe/orbit/platform/auth"
"git.yuntuo.hk/k2safe/orbit/pulse/service"
"git.yuntuo.hk/k2safe/orbit/weave/instance"
)
const workflowOperationColumns = `id,resource_id,operation_id,actor_id,action,created_at`
func scanWorkflowOperation(row scanner) (instance.OperationSummary, error) {
var v instance.OperationSummary
err := row.Scan(&v.ID, &v.WorkflowID, &v.OperationID, &v.ActorID, &v.Action, &v.CreatedAt)
return v, err
}
func (s *WorkflowInstanceStore) ListOperations(ctx context.Context, t auth.Token, id core.ID, in service.HistoryPageInput) (instance.OperationPage, error) {
out := instance.OperationPage{Items: []instance.OperationSummary{}}
err := s.transaction(ctx, t, instance.AuditAction, func(tx *sql.Tx, p auth.Permissions) error {
if _, _, _, err := readWorkflowForQuery(ctx, tx, t.NamespaceID, id, p); err != nil {
return err
}
query := `SELECT ` + workflowOperationColumns + ` FROM orbit_audit_events WHERE namespace_id=? AND resource_kind='workflow' AND resource_id=?`
args := []any{t.NamespaceID, id}
if in.Cursor != "" {
c, err := decodeHistoryCursorKind(in.Cursor, t.NamespaceID, id, "", "workflow-operations")
if err != nil {
return err
}
query += ` AND (created_at<? OR (created_at=? AND id<?))`
args = append(args, *c.Created, *c.Created, c.ID)
}
query += ` ORDER BY created_at DESC,id DESC LIMIT ?`
args = append(args, in.Limit+1)
rows, err := tx.QueryContext(ctx, query, args...)
if err != nil {
return err
}
defer rows.Close()
for rows.Next() {
v, err := scanWorkflowOperation(rows)
if err != nil {
return err
}
out.Items = append(out.Items, v)
}
if err := rows.Err(); err != nil {
return err
}
if len(out.Items) > in.Limit {
out.Items = out.Items[:in.Limit]
last := out.Items[len(out.Items)-1]
out.NextCursor, err = encodeHistoryCursor(historyCursor{Version: 1, Kind: "workflow-operations", Namespace: t.NamespaceID, Task: id, ID: last.ID, Created: &last.CreatedAt})
}
return err
})
return out, err
}
func (s *WorkflowInstanceStore) GetOperation(ctx context.Context, t auth.Token, id, audit core.ID) (instance.OperationView, error) {
var out instance.OperationView
err := s.transaction(ctx, t, instance.AuditAction, func(tx *sql.Tx, p auth.Permissions) error {
if _, _, _, err := readWorkflowForQuery(ctx, tx, t.NamespaceID, id, p); err != nil {
return err
}
var before, after []byte
err := tx.QueryRowContext(ctx, `SELECT `+workflowOperationColumns+",reason,COALESCE(evidence_ref,''),`before`,`after` FROM orbit_audit_events WHERE namespace_id=? AND resource_kind='workflow' AND resource_id=? AND id=?", t.NamespaceID, id, audit).Scan(&out.ID, &out.WorkflowID, &out.OperationID, &out.ActorID, &out.Action, &out.CreatedAt, &out.Reason, &out.EvidenceRef, &before, &after)
if errors.Is(err, sql.ErrNoRows) {
return fault.NotFound
}
out.Before, out.After = before, after
return err
})
return out, err
}

1
internal/storage/mysql/workflow_poll_retry.go
View File

@@ -128,6 +128,7 @@ func retryWorkflowPoll(ctx context.Context, tx *sql.Tx, g *workflowGraph, row *w
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 {

6
internal/storage/mysql/workflow_runtime_graph.go
View File

@@ -54,7 +54,7 @@ func lockWorkflowGraph(ctx context.Context, tx *sql.Tx, ns, id core.ID, skip boo
return nil, false, err
}
g := &workflowGraph{view: v, gate: gate, version: version, program: program, steps: map[string]*workflowStep{}}
rows, err := tx.QueryContext(ctx, "SELECT s.id,s.`key`,s.kind,s.state,s.revision,s.current_activation_id,a.task_id,a.id,a.state,a.input,a.output,t.id,t.queue_id FROM weave_steps s LEFT JOIN weave_activations a ON a.namespace_id=s.namespace_id AND a.id=s.current_activation_id AND a.workflow_id=s.workflow_id AND a.step_id=s.id LEFT JOIN pulse_tasks t ON t.namespace_id=s.namespace_id AND t.id=a.task_id AND t.owner_kind='workflow_activation' AND t.owner_id=a.id AND t.gate_id=? WHERE s.namespace_id=? AND s.workflow_id=? ORDER BY s.`key` LIMIT 130", gate, ns, id)
rows, err := tx.QueryContext(ctx, "SELECT s.id,s.`key`,s.kind,s.state,s.revision,s.current_activation_id,a.task_id,COALESCE(a.sequence,0),a.id,a.state,a.input,a.output,t.id,t.queue_id FROM weave_steps s LEFT JOIN weave_activations a ON a.namespace_id=s.namespace_id AND a.id=s.current_activation_id AND a.workflow_id=s.workflow_id AND a.step_id=s.id LEFT JOIN pulse_tasks t ON t.namespace_id=s.namespace_id AND t.id=a.task_id AND t.owner_kind='workflow_activation' AND t.owner_id=a.id AND t.gate_id=? WHERE s.namespace_id=? AND s.workflow_id=? ORDER BY s.`key` LIMIT 130", gate, ns, id)
if err != nil {
return nil, false, err
}
@@ -64,7 +64,7 @@ func lockWorkflowGraph(ctx context.Context, tx *sql.Tx, ns, id core.ID, skip boo
var activation, task, queue *core.ID
var state sql.NullString
var input, output []byte
if err := rows.Scan(&row.view.ID, &row.view.Key, &row.view.Kind, &row.view.State, &row.view.Revision, &row.view.ActivationID, &row.view.TaskID, &activation, &state, &input, &output, &task, &queue); err != nil {
if err := rows.Scan(&row.view.ID, &row.view.Key, &row.view.Kind, &row.view.State, &row.view.Revision, &row.view.ActivationID, &row.view.TaskID, &row.view.ActivationSequence, &activation, &state, &input, &output, &task, &queue); err != nil {
return nil, false, err
}
row.input, row.output = input, output
@@ -78,7 +78,7 @@ func lockWorkflowGraph(ctx context.Context, tx *sql.Tx, ns, id core.ID, skip boo
return nil, false, errors.New("workflow step definition mismatch")
}
inactive := row.view.State == weave.StepBlocked || row.view.State == weave.StepSkipped
if inactive && row.view.ActivationID != nil || !inactive && (row.view.ActivationID == nil || activation == nil || *activation != *row.view.ActivationID || state.String != string(row.view.State)) {
if inactive && row.view.ActivationID != nil || !inactive && (row.view.ActivationSequence == 0 || row.view.ActivationID == nil || activation == nil || *activation != *row.view.ActivationID || state.String != string(row.view.State)) {
return nil, false, errors.New("workflow current activation mismatch")
}
if !inactive && (row.view.Kind == weave.NodeTask) != (row.view.TaskID != nil) {

1
internal/storage/mysql/workflow_step_retry.go
View File

@@ -142,6 +142,7 @@ func (s *WorkflowRecoveryStore) RetryStep(ctx context.Context, t auth.Token, id
return nil, err
}
row.view.State, row.view.ActivationID, row.view.TaskID = weave.StepActive, &activation, &taskID
row.view.ActivationSequence = sequence + 1
row.view.Revision++
row.output = nil
if err := deleteWorkflowClock(ctx, tx, g, row); err != nil {

1
internal/storage/mysql/workflow_wait_rearm.go
View File

@@ -109,6 +109,7 @@ func (s *WorkflowRecoveryStore) RearmWait(ctx context.Context, t auth.Token, id
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

28
internal/transport/httpserver/tasks.go
View File

@@ -233,7 +233,7 @@ func writeError(w http.ResponseWriter, id string, err error) {
if !errors.As(err, &known) {
known = fault.Unavailable
}
status := map[string]int{"INVALID_ARGUMENT": 400, "UNAUTHENTICATED": 401, "PERMISSION_DENIED": 403, "NOT_FOUND": 404, "STATE_CONFLICT": 409, "IDEMPOTENCY_CONFLICT": 409, "BODY_TOO_LARGE": 413, "UNSUPPORTED_MEDIA_TYPE": 415, "DELIVERY_LEASE_LOST": 409, "EVENT_HISTORY_EXPIRED": 410, "SUBSCRIPTION_LIMIT": 409, "WORKFLOW_DEFINITION_LIMIT": 409, "WORKFLOW_VERSION_LIMIT": 409}[known.Code]
status := map[string]int{"INVALID_ARGUMENT": 400, "UNAUTHENTICATED": 401, "PERMISSION_DENIED": 403, "NOT_FOUND": 404, "STATE_CONFLICT": 409, "IDEMPOTENCY_CONFLICT": 409, "BODY_TOO_LARGE": 413, "UNSUPPORTED_MEDIA_TYPE": 415, "DELIVERY_LEASE_LOST": 409, "EVENT_HISTORY_EXPIRED": 410, "SUBSCRIPTION_LIMIT": 409, "WORKFLOW_DEFINITION_LIMIT": 409, "WORKFLOW_VERSION_LIMIT": 409, "WORKFLOW_INSTANCE_LIMIT": 409, "WORKFLOW_SIGNAL_LIMIT": 409}[known.Code]
if status == 0 {
status = 503
}
@@ -320,10 +320,23 @@ func exactFields(raw []byte, t reflect.Type) error {
}
names := map[string]reflect.Type{}
nullable := map[string]bool{}
for i := 0; i < t.NumField(); i++ {
field := t.Field(i)
name := strings.Split(field.Tag.Get("json"), ",")[0]
if name != "" && name != "-" {
var collect func(reflect.Type) error
collect = func(typ reflect.Type) error {
for i := 0; i < typ.NumField(); i++ {
field := typ.Field(i)
name := strings.Split(field.Tag.Get("json"), ",")[0]
if field.Anonymous && name == "" && field.Type.Kind() == reflect.Struct {
if err := collect(field.Type); err != nil {
return err
}
continue
}
if name == "" || name == "-" {
continue
}
if _, exists := names[name]; exists {
return errors.New("ambiguous embedded JSON field")
}
names[name] = field.Type
nullable[name] = field.Tag.Get("nullable") == "true"
if field.Tag.Get("required") == "true" {
@@ -332,7 +345,12 @@ func exactFields(raw []byte, t reflect.Type) error {
}
}
}
return nil
}
if err := collect(t); err != nil {
return err
}
for name, value := range object {
field, ok := names[name]
if !ok {

102
internal/transport/httpserver/workflows.go Normal file
View File

@@ -0,0 +1,102 @@
package httpserver
import (
"context"
"net/http"
core "git.yuntuo.hk/k2safe/orbit/model"
"git.yuntuo.hk/k2safe/orbit/platform/auth"
"git.yuntuo.hk/k2safe/orbit/weave/control"
"git.yuntuo.hk/k2safe/orbit/weave/instance"
"git.yuntuo.hk/k2safe/orbit/weave/recovery"
"git.yuntuo.hk/k2safe/orbit/weave/signal"
)
// RegisterWorkflows 只适配协议，实际授权、并发版本、原键回执及审计由事务服务校验。
func RegisterWorkflows(server *http.Server, instances *instance.Service, controls *control.Service, signals *signal.Service, reviews *recovery.Service) {
mux := server.Handler.(*http.ServeMux)
mux.HandleFunc("POST /v1/workflows", wrap(func(r *http.Request, t auth.Token) (any, int, error) {
if _, err := historyQuery(r, false); err != nil {
return nil, 0, err
}
var in instance.StartInput
if err := decodeBody(r, &in); err != nil {
return nil, 0, err
}
key, err := idempotencyKey(r)
if err != nil {
return nil, 0, err
}
out, err := instances.Start(r.Context(), t, key, in)
return out, createdStatus(out.Deduplicated), err
}))
mux.HandleFunc("GET /v1/workflows/{id}", wrap(func(r *http.Request, t auth.Token) (any, int, error) {
if _, err := historyQuery(r, false); err != nil {
return nil, 0, err
}
out, err := instances.Get(r.Context(), t, core.ID(r.PathValue("id")))
return out, 200, err
}))
mux.HandleFunc("GET /v1/workflows/{id}/steps", wrap(func(r *http.Request, t auth.Token) (any, int, error) {
if _, err := historyQuery(r, false); err != nil {
return nil, 0, err
}
out, err := instances.Steps(r.Context(), t, core.ID(r.PathValue("id")))
return out, 200, err
}))
mux.HandleFunc("GET /v1/workflows/{id}/steps/{key}", wrap(func(r *http.Request, t auth.Token) (any, int, error) {
if _, err := historyQuery(r, false); err != nil {
return nil, 0, err
}
out, err := instances.Step(r.Context(), t, core.ID(r.PathValue("id")), r.PathValue("key"))
return out, 200, err
}))
mux.HandleFunc("GET /v1/workflows/{id}/operations", wrap(func(r *http.Request, t auth.Token) (any, int, error) {
in, err := historyQuery(r, true)
if err != nil {
return nil, 0, err
}
out, err := instances.ListOperations(r.Context(), t, core.ID(r.PathValue("id")), in)
return out, 200, err
}))
mux.HandleFunc("GET /v1/workflows/{id}/operations/{operation}", wrap(func(r *http.Request, t auth.Token) (any, int, error) {
if _, err := historyQuery(r, false); err != nil {
return nil, 0, err
}
out, err := instances.GetOperation(r.Context(), t, core.ID(r.PathValue("id")), core.ID(r.PathValue("operation")))
return out, 200, err
}))
mux.HandleFunc("POST /v1/workflows/{id}/control", workflowMutation(controls.Apply, func(control.Result) int { return 200 }))
mux.HandleFunc("POST /v1/workflows/{id}/signals", workflowMutation(signals.Send, func(v signal.Result) int { return createdStatus(v.Deduplicated) }))
mux.HandleFunc("GET /v1/workflows/{id}/signals/{signal}", wrap(func(r *http.Request, t auth.Token) (any, int, error) {
if _, err := historyQuery(r, false); err != nil {
return nil, 0, err
}
out, err := signals.Get(r.Context(), t, core.ID(r.PathValue("id")), core.ID(r.PathValue("signal")))
return out, 200, err
}))
mux.HandleFunc("POST /v1/workflows/{id}/recovery/task-review", workflowMutation(reviews.ReviewTask, recoveryStatus))
mux.HandleFunc("POST /v1/workflows/{id}/recovery/step-resolve", workflowMutation(reviews.ResolveStep, recoveryStatus))
mux.HandleFunc("POST /v1/workflows/{id}/recovery/continue", workflowMutation(reviews.Continue, recoveryStatus))
mux.HandleFunc("POST /v1/workflows/{id}/recovery/step-retry", workflowMutation(reviews.RetryStep, recoveryStatus))
mux.HandleFunc("POST /v1/workflows/{id}/recovery/poll-continue", workflowMutation(reviews.ContinuePoll, recoveryStatus))
mux.HandleFunc("POST /v1/workflows/{id}/recovery/wait-rearm", workflowMutation(reviews.RearmWait, recoveryStatus))
}
func recoveryStatus(recovery.Result) int { return 200 }
func workflowMutation[I, O any](call func(context.Context, auth.Token, core.ID, string, I) (O, error), status func(O) int) http.HandlerFunc {
return wrap(func(r *http.Request, t auth.Token) (any, int, error) {
if _, err := historyQuery(r, false); err != nil {
return nil, 0, err
}
var in I
if err := decodeBody(r, &in); err != nil {
return nil, 0, err
}
key, err := idempotencyKey(r)
if err != nil {
return nil, 0, err
}
out, err := call(r.Context(), t, core.ID(r.PathValue("id")), key, in)
return out, status(out), err
})
}

69
internal/transport/httpserver/workflows_test.go Normal file
View File

@@ -0,0 +1,69 @@
package httpserver
import (
"encoding/json"
"net/http/httptest"
"strings"
"testing"
"git.yuntuo.hk/k2safe/orbit/internal/fault"
"git.yuntuo.hk/k2safe/orbit/weave/instance"
"git.yuntuo.hk/k2safe/orbit/weave/recovery"
)
func TestWorkflowRecoveryEmbeddedJSON(t *testing.T) {
valid := `{"expected_revision":"9007199254740993","reason":"核查","evidence_ref":"evidence:1","step_key":"query","activation_id":"11111111111111111111111111111111","expected_step_revision":"3","pending_confirmed":true}`
for _, tc := range []struct {
name, body string
valid bool
}{
{"flat embedded fields", valid, true},
{"unknown parent", strings.Replace(valid, `"reason"`, `"Reason"`, 1), false},
{"unknown step", strings.Replace(valid, `"step_key"`, `"StepKey"`, 1), false},
{"nested decision", `{"Decision":{"expected_revision":"1"},"pending_confirmed":true}`, false},
{"missing confirmation", strings.Replace(valid, `,"pending_confirmed":true`, "", 1), false},
{"null confirmation", strings.Replace(valid, `"pending_confirmed":true`, `"pending_confirmed":null`, 1), false},
{"numeric revision", strings.Replace(valid, `"9007199254740993"`, `9007199254740993`, 1), false},
{"duplicate inherited field", strings.Replace(valid, `"reason":"核查"`, `"reason":"核查","reason":"改写"`, 1), false},
} {
t.Run(tc.name, func(t *testing.T) {
r := httptest.NewRequest("POST", "/", strings.NewReader(tc.body))
r.Header.Set("Content-Type", "application/json")
var in recovery.ContinuePollInput
err := decodeBody(r, &in)
if (err == nil) != tc.valid {
t.Fatal(err)
}
if tc.valid && (in.ExpectedRevision != 9007199254740993 || in.ExpectedStepRevision != 3 || in.StepKey != "query") {
t.Fatal("embedded fields lost")
}
})
}
for _, raw := range []string{`{"execution_settled":true,"result":null}`, `{"execution_settled":true,"result":{"n":9007199254740993}}`} {
r := httptest.NewRequest("POST", "/", strings.NewReader(raw))
r.Header.Set("Content-Type", "application/json")
var in recovery.ResolveStepInput
if err := decodeBody(r, &in); err != nil {
t.Fatal(err)
}
if !json.Valid(in.Result) {
t.Fatal("result lost")
}
}
r := httptest.NewRequest("POST", "/", strings.NewReader(`{"definition_id":"11111111111111111111111111111111","version":"v1","input":null}`))
r.Header.Set("Content-Type", "application/json")
var in instance.StartInput
if err := decodeBody(r, &in); err != nil || string(in.Input) != "null" {
t.Fatal("explicit null input", err)
}
}
func TestWorkflowLimitsAreExplicitConflicts(t *testing.T) {
for _, code := range []string{"WORKFLOW_INSTANCE_LIMIT", "WORKFLOW_SIGNAL_LIMIT"} {
w := httptest.NewRecorder()
writeError(w, "request", fault.New(code, "limit reached"))
if w.Code != 409 {
t.Fatal(code, w.Code)
}
}
}