feat: add bounded workflow decisions and durable result inbox
This commit is contained in:
@@ -64,6 +64,7 @@ func TestIntegrationApplicationResourcesUpgrade(t *testing.T) {
|
||||
retryFinish(t, g, finished, "SUCCEEDED", false)
|
||||
assertApplicationCounts(t, f, 4, 2, 1)
|
||||
// 还原后续新增表和字段，得到真实 v9 结构；现有任务内容保持不变。
|
||||
f.exec(`ALTER TABLE weave_inbox DROP CHECK ck_weave_inbox_source, DROP INDEX uq_weave_task_result, DROP COLUMN source_task_id, DROP COLUMN source_activation_id, DROP COLUMN source_revision, DROP COLUMN snapshot`)
|
||||
f.exec(`ALTER TABLE pulse_tasks DROP INDEX ix_task_queue_business_created, DROP INDEX ix_task_queue_business_state_created, MODIFY COLUMN business_key VARCHAR(191) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_as_cs NULL COMMENT '业务侧关联键，用于检索追踪，不自动提供业务幂等'`)
|
||||
f.exec(`ALTER TABLE pulse_tasks DROP CHECK ck_task_retry_not_self, DROP INDEX uq_task_retry_parent, ADD INDEX fk_pulse_tasks_3(namespace_id,retry_of)`)
|
||||
f.exec(`DROP TABLE orbit_application_resources`)
|
||||
|
||||
@@ -0,0 +1,18 @@
|
||||
-- Weave内部结果保存独立快照，避免外部结果日志清理或任务后续核查改变历史输入。
|
||||
ALTER TABLE weave_inbox
|
||||
ADD COLUMN source_task_id BINARY(16) NULL COMMENT '来源Pulse任务标识；仅迁移前无法核实的历史记录允许为空',
|
||||
ADD COLUMN source_activation_id BINARY(16) NULL COMMENT '事件发生时的Weave激活标识，保留旧激活归属；仅未核实历史记录可空',
|
||||
ADD COLUMN source_revision BIGINT UNSIGNED NULL COMMENT '来源任务的状态版本，必须大于零；仅未核实历史记录可空',
|
||||
ADD COLUMN snapshot MEDIUMTEXT NULL COMMENT 'schema_version版本化结果JSON文本，保留精确数字与当时状态；仅未核实历史记录可空';
|
||||
|
||||
-- 旧记录缺少冻结结果，不能用当前任务状态倒推历史并自动推进。
|
||||
UPDATE weave_inbox SET state='QUARANTINED',lease_owner=NULL,lease_until=NULL,
|
||||
error_code='LEGACY_RESULT_UNVERIFIED',revision=revision+1,updated_at=UTC_TIMESTAMP(6)
|
||||
WHERE state<>'PROCESSED';
|
||||
|
||||
ALTER TABLE weave_inbox
|
||||
ADD UNIQUE KEY uq_weave_task_result (namespace_id,source_task_id,source_revision),
|
||||
ADD CONSTRAINT ck_weave_inbox_source CHECK (
|
||||
(state IN ('PROCESSED','QUARANTINED') AND source_task_id IS NULL AND source_activation_id IS NULL AND source_revision IS NULL AND snapshot IS NULL)
|
||||
OR (source_task_id IS NOT NULL AND source_activation_id IS NOT NULL AND source_revision IS NOT NULL AND source_revision>0 AND snapshot IS NOT NULL AND OCTET_LENGTH(snapshot)<=524288 AND JSON_VALID(snapshot))
|
||||
);
|
||||
@@ -88,8 +88,8 @@ func TestIntegrationNoForeignKeysUpgrade(t *testing.T) {
|
||||
}
|
||||
indexes := func() []string {
|
||||
t.Helper()
|
||||
// v15 将原 retry_of 非唯一索引替换为相同列的唯一索引，下方单独验证其定义。
|
||||
rows, err := db.QueryContext(ctx, `SELECT CONCAT_WS(':',TABLE_NAME,INDEX_NAME,NON_UNIQUE,SEQ_IN_INDEX,COLUMN_NAME,COALESCE(SUB_PART,0),INDEX_TYPE,IS_VISIBLE) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME NOT IN ('orbit_application_resources','orbit_event_streams','orbit_result_events') AND INDEX_NAME NOT IN ('ix_task_priority_dispatch','ix_delivery_subscription_due','ix_delivery_subscription_lease','ix_delivery_history','ix_delivery_state_history','uq_task_retry_parent','fk_pulse_tasks_3','ix_task_queue_business_created','ix_task_queue_business_state_created') ORDER BY TABLE_NAME,INDEX_NAME,SEQ_IN_INDEX`)
|
||||
// v15替换retry_of索引、v17增加内部结果唯一索引，下方单独验证定义。
|
||||
rows, err := db.QueryContext(ctx, `SELECT CONCAT_WS(':',TABLE_NAME,INDEX_NAME,NON_UNIQUE,SEQ_IN_INDEX,COLUMN_NAME,COALESCE(SUB_PART,0),INDEX_TYPE,IS_VISIBLE) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME NOT IN ('orbit_application_resources','orbit_event_streams','orbit_result_events') AND INDEX_NAME NOT IN ('ix_task_priority_dispatch','ix_delivery_subscription_due','ix_delivery_subscription_lease','ix_delivery_history','ix_delivery_state_history','uq_task_retry_parent','fk_pulse_tasks_3','ix_task_queue_business_created','ix_task_queue_business_state_created','uq_weave_task_result') ORDER BY TABLE_NAME,INDEX_NAME,SEQ_IN_INDEX`)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -173,6 +173,10 @@ func TestIntegrationNoForeignKeysUpgrade(t *testing.T) {
|
||||
if err := db.QueryRowContext(ctx, `SELECT GROUP_CONCAT(CONCAT(COLUMN_NAME,':',NON_UNIQUE) ORDER BY SEQ_IN_INDEX) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='pulse_tasks' AND INDEX_NAME='uq_task_retry_parent'`).Scan(&retryIndex); err != nil || retryIndex != "namespace_id:0,retry_of:0" {
|
||||
t.Fatalf("retry uniqueness replacement=%q err=%v", retryIndex, err)
|
||||
}
|
||||
var resultIndex string
|
||||
if err := db.QueryRowContext(ctx, `SELECT GROUP_CONCAT(CONCAT(COLUMN_NAME,':',NON_UNIQUE) ORDER BY SEQ_IN_INDEX) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='weave_inbox' AND INDEX_NAME='uq_weave_task_result'`).Scan(&resultIndex); err != nil || resultIndex != "namespace_id:0,source_task_id:0,source_revision:0" {
|
||||
t.Fatalf("workflow result uniqueness=%q err=%v", resultIndex, err)
|
||||
}
|
||||
var collation string
|
||||
if err := db.QueryRowContext(ctx, `SELECT COLLATION_NAME FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='pulse_tasks' AND COLUMN_NAME='business_key'`).Scan(&collation); err != nil || collation != "utf8mb4_0900_bin" {
|
||||
t.Fatalf("business key collation=%q err=%v", collation, err)
|
||||
|
||||
@@ -17,8 +17,13 @@ import (
|
||||
func appendResultEvent(ctx context.Context, tx *sql.Tx, ns, eventID, taskID core.ID, revision uint64, eventType string) error {
|
||||
var snapshot events.TaskResult
|
||||
var raw []byte
|
||||
var link workflowTaskLink
|
||||
snapshot.SchemaVersion = 1
|
||||
err := tx.QueryRowContext(ctx, `SELECT id,queue_id,state,revision,result,effect_uncertain,current_attempt_id FROM pulse_tasks WHERE namespace_id=? AND id=?`, ns, taskID).Scan(&snapshot.TaskID, &snapshot.QueueID, &snapshot.State, &snapshot.Revision, &raw, &snapshot.EffectUncertain, &snapshot.CurrentAttemptID)
|
||||
err := tx.QueryRowContext(ctx, `SELECT t.id,t.queue_id,t.state,t.revision,t.result,t.effect_uncertain,t.current_attempt_id,t.owner_kind,t.owner_id,t.gate_id,
|
||||
EXISTS(SELECT 1 FROM weave_activations a WHERE a.namespace_id=t.namespace_id AND a.task_id=t.id)
|
||||
OR EXISTS(SELECT 1 FROM weave_poll_rounds r WHERE r.namespace_id=t.namespace_id AND r.task_id=t.id)
|
||||
OR EXISTS(SELECT 1 FROM weave_instances i WHERE i.namespace_id=t.namespace_id AND i.gate_id=t.gate_id)
|
||||
FROM pulse_tasks t WHERE t.namespace_id=? AND t.id=?`, ns, taskID).Scan(&snapshot.TaskID, &snapshot.QueueID, &snapshot.State, &snapshot.Revision, &raw, &snapshot.EffectUncertain, &snapshot.CurrentAttemptID, &link.Owner, &link.Activation, &link.Gate, &link.Linked)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -30,6 +35,10 @@ func appendResultEvent(ctx context.Context, tx *sql.Tx, ns, eventID, taskID core
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// 内部待处理记录和结果同事务，外部结果流锁仍最后取得。
|
||||
if err := appendWorkflowResult(ctx, tx, ns, eventID, taskID, revision, body, link); err != nil {
|
||||
return err
|
||||
}
|
||||
head, _, err := lockEventStream(ctx, tx, ns)
|
||||
if err != nil {
|
||||
return err
|
||||
|
||||
80
internal/storage/mysql/workflow_events.go
Normal file
80
internal/storage/mysql/workflow_events.go
Normal file
@@ -0,0 +1,80 @@
|
||||
package mysql
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
|
||||
core "git.yuntuo.hk/k2safe/orbit/model"
|
||||
"git.yuntuo.hk/k2safe/orbit/weave/engine"
|
||||
)
|
||||
|
||||
type workflowTaskLink struct {
|
||||
Owner sql.NullString
|
||||
Activation *core.ID
|
||||
Gate *core.ID
|
||||
Linked bool
|
||||
}
|
||||
|
||||
// appendWorkflowResult 在原Pulse结果事务内创建内部交付，必须先于外部结果流行锁。
|
||||
// 这里只核验稳定归属并写入队列，不锁实例/激活、不推进流程、不锁对外订阅。
|
||||
func appendWorkflowResult(ctx context.Context, tx *sql.Tx, ns, eventID, taskID core.ID, revision uint64, snapshot []byte, link workflowTaskLink) error {
|
||||
owner, ownerID, gate := link.Owner, link.Activation, link.Gate
|
||||
if !owner.Valid || owner.String != engine.TaskOwnerKind {
|
||||
if owner.Valid && owner.String != "schedule" {
|
||||
return errors.New("unrecognized task owner")
|
||||
}
|
||||
// 无外键情况下，丢失owner不能把已有流程关联静默降级成普通任务。
|
||||
if link.Linked {
|
||||
return errors.New("workflow result ownership missing")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
if ownerID == nil || gate == nil {
|
||||
return errors.New("workflow result owner or gate missing")
|
||||
}
|
||||
var workflow core.ID
|
||||
var kind string
|
||||
var activationTask *core.ID
|
||||
// 即使旧激活不再是current，也要保存内部事件，消费方再核对当前激活并留痕。
|
||||
err := tx.QueryRowContext(ctx, `SELECT a.workflow_id,s.kind,a.task_id
|
||||
FROM weave_activations a
|
||||
JOIN weave_steps s ON s.namespace_id=a.namespace_id AND s.id=a.step_id AND s.workflow_id=a.workflow_id
|
||||
JOIN weave_instances i ON i.namespace_id=a.namespace_id AND i.id=a.workflow_id AND i.gate_id=?
|
||||
JOIN pulse_execution_gates g ON g.namespace_id=i.namespace_id AND g.id=i.gate_id
|
||||
JOIN weave_definition_versions v ON v.namespace_id=i.namespace_id AND v.id=i.definition_version_id AND v.definition_id=i.definition_id
|
||||
JOIN weave_definitions d ON d.namespace_id=i.namespace_id AND d.id=i.definition_id
|
||||
WHERE a.namespace_id=? AND a.id=?`, gate, ns, ownerID).Scan(&workflow, &kind, &activationTask)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return errors.New("workflow result associations invalid")
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
switch kind {
|
||||
case "TASK":
|
||||
if activationTask == nil || *activationTask != taskID {
|
||||
return errors.New("workflow activation task mismatch")
|
||||
}
|
||||
case "POLL":
|
||||
if activationTask != nil {
|
||||
return errors.New("poll activation cannot own a direct task")
|
||||
}
|
||||
var round uint32
|
||||
err = tx.QueryRowContext(ctx, `SELECT r.round FROM weave_poll_rounds r JOIN weave_poll_states p ON p.namespace_id=r.namespace_id AND p.workflow_id=r.workflow_id AND p.activation_id=r.activation_id WHERE r.namespace_id=? AND r.workflow_id=? AND r.activation_id=? AND r.task_id=? AND r.round BETWEEN 1 AND p.used_rounds`, ns, workflow, ownerID, taskID).Scan(&round)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return errors.New("workflow poll task association invalid")
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
default:
|
||||
return errors.New("workflow task owner kind invalid")
|
||||
}
|
||||
id, err := core.NewID()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
_, err = tx.ExecContext(ctx, `INSERT INTO weave_inbox(namespace_id,id,workflow_id,event_id,available_at,source_task_id,source_activation_id,source_revision,snapshot) VALUES (?,?,?,?,UTC_TIMESTAMP(6),?,?,?,?)`, ns, id, workflow, eventID, taskID, ownerID, revision, string(snapshot))
|
||||
return err
|
||||
}
|
||||
277
internal/storage/mysql/workflow_events_integration_test.go
Normal file
277
internal/storage/mysql/workflow_events_integration_test.go
Normal file
@@ -0,0 +1,277 @@
|
||||
package mysql
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"os"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
core "git.yuntuo.hk/k2safe/orbit/model"
|
||||
"git.yuntuo.hk/k2safe/orbit/platform/auth"
|
||||
"git.yuntuo.hk/k2safe/orbit/platform/events"
|
||||
pulse "git.yuntuo.hk/k2safe/orbit/pulse/model"
|
||||
"git.yuntuo.hk/k2safe/orbit/pulse/service"
|
||||
"git.yuntuo.hk/k2safe/orbit/pulse/worker"
|
||||
"git.yuntuo.hk/k2safe/orbit/weave/definition"
|
||||
"git.yuntuo.hk/k2safe/orbit/weave/engine"
|
||||
)
|
||||
|
||||
// 当前尚无实例API；此夹具只为验证Pulse结果交付边界，不能充当已实现的实例运行器。
|
||||
type workflowOwnerFixture struct{ workflow, step, activation, gate core.ID }
|
||||
|
||||
func bindWorkflowOwner(t *testing.T, f *executionFixture, task core.ID, poll bool) workflowOwnerFixture {
|
||||
t.Helper()
|
||||
ns := f.resources.NamespaceID
|
||||
permissions := auth.Permissions{Actions: []string{"tasks:read", "tasks:submit", "tasks:cancel", "workers:connect", "workers:consume", definition.ReadAction, definition.PublishAction}, Queues: []string{"*"}}
|
||||
raw, _ := json.Marshal(permissions)
|
||||
f.exec(`UPDATE orbit_credentials SET permissions=? WHERE namespace_id=? AND id=?`, string(raw), ns, f.resources.CredentialID)
|
||||
api := definition.New(NewWorkflowDefinitionStore(f.db))
|
||||
created, err := api.Create(f.ctx, f.token, string(executionID(t)), definition.CreateInput{Name: "flow-" + string(executionID(t))})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
spec := workflowTestSpec(f.resources.QueueID)
|
||||
kind, key := "TASK", "prepare"
|
||||
if poll {
|
||||
kind, key = "POLL", "query"
|
||||
query := *spec.Body.Steps[0].Task
|
||||
status := definition.Type{Kind: "OBJECT", Fields: map[string]definition.Type{"status": {Kind: "STRING", Enum: []string{"PENDING", "DONE"}}}}
|
||||
query.ResultType = status
|
||||
spec.Body = definition.Block{Steps: []definition.Step{{Key: key, Kind: kind, Poll: &definition.Poll{Task: query, MaxRounds: 2, TimeoutMS: 10000, IntervalMS: 1000, StatusPath: []string{"status"}, Pending: []string{"PENDING"}, Succeeded: []string{"DONE"}}}}, OutputType: status, Output: definition.Value{Kind: "REFERENCE", Reference: &definition.Reference{Source: key}}}
|
||||
}
|
||||
version, err := api.Publish(f.ctx, f.token, created.Definition.ID, string(executionID(t)), definition.PublishInput{Version: "v1", Reason: "internal result delivery regression", Specification: spec})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
out := workflowOwnerFixture{executionID(t), executionID(t), executionID(t), executionID(t)}
|
||||
f.exec(`INSERT INTO pulse_execution_gates(namespace_id,id) VALUES (?,?)`, ns, out.gate)
|
||||
f.exec(`INSERT INTO weave_instances(namespace_id,id,definition_id,definition_version_id,gate_id,input,policy_snapshot,deadline_at) VALUES (?,?,?,?,?,'{"n":9007199254740993}','{}',DATE_ADD(UTC_TIMESTAMP(6),INTERVAL 1 HOUR))`, ns, out.workflow, created.Definition.ID, version.Version.ID, out.gate)
|
||||
f.exec("INSERT INTO weave_steps(namespace_id,id,workflow_id,`key`,kind,state) VALUES (?,?,?,?,?,'ACTIVE')", ns, out.step, out.workflow, key, kind)
|
||||
var direct *core.ID
|
||||
if !poll {
|
||||
direct = &task
|
||||
}
|
||||
f.exec(`INSERT INTO weave_activations(namespace_id,id,workflow_id,step_id,sequence,state,task_id,input) VALUES (?,?,?,?,1,'ACTIVE',?,'{"n":9007199254740993}')`, ns, out.activation, out.workflow, out.step, direct)
|
||||
f.exec(`UPDATE weave_steps SET current_activation_id=? WHERE namespace_id=? AND id=?`, out.activation, ns, out.step)
|
||||
f.exec(`UPDATE pulse_tasks SET owner_kind=?,owner_id=?,gate_id=? WHERE namespace_id=? AND id=?`, engine.TaskOwnerKind, out.activation, out.gate, ns, task)
|
||||
if poll {
|
||||
f.exec(`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) VALUES (?,?,?,?,2,1,1,1000,DATE_ADD(UTC_TIMESTAMP(6),INTERVAL 1 HOUR),?)`, ns, executionID(t), out.workflow, out.activation, task)
|
||||
f.exec("INSERT INTO weave_poll_rounds(namespace_id,id,workflow_id,activation_id,`round`,task_id) VALUES (?,?,?,?,1,?)", ns, executionID(t), out.workflow, out.activation, task)
|
||||
}
|
||||
return out
|
||||
}
|
||||
func assertWorkflowInbox(t *testing.T, f *executionFixture, owner workflowOwnerFixture, task core.ID, state string, raw string) {
|
||||
t.Helper()
|
||||
var taskID, activationID, workflowID core.ID
|
||||
var revision uint64
|
||||
var snapshot []byte
|
||||
var status string
|
||||
if err := f.db.QueryRowContext(f.ctx, `SELECT source_task_id,source_activation_id,workflow_id,source_revision,snapshot,state FROM weave_inbox WHERE namespace_id=? AND workflow_id=?`, f.resources.NamespaceID, owner.workflow).Scan(&taskID, &activationID, &workflowID, &revision, &snapshot, &status); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var result events.TaskResult
|
||||
if err := json.Unmarshal(snapshot, &result); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if taskID != task || activationID != owner.activation || workflowID != owner.workflow || revision != result.Revision || status != "PENDING" || result.State != state || result.TaskID != task || string(result.Result) != raw {
|
||||
t.Fatal("internal result snapshot mismatch", result)
|
||||
}
|
||||
if f.count("weave_inbox", "workflow_id=?", owner.workflow) != 1 {
|
||||
t.Fatal("duplicate workflow delivery")
|
||||
}
|
||||
var n int
|
||||
if err := f.db.QueryRowContext(f.ctx, `SELECT COUNT(*) FROM weave_inbox i JOIN orbit_events e ON e.namespace_id=i.namespace_id AND e.id=i.event_id JOIN orbit_result_events r ON r.namespace_id=i.namespace_id AND r.id=i.event_id WHERE i.namespace_id=? AND i.workflow_id=? AND e.aggregate_id=i.source_task_id AND e.aggregate_revision=i.source_revision`, f.resources.NamespaceID, owner.workflow).Scan(&n); err != nil || n != 1 {
|
||||
t.Fatal("source events not atomic", err)
|
||||
}
|
||||
}
|
||||
func TestIntegrationWorkflowResultDelivery(t *testing.T) {
|
||||
dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN")
|
||||
if dsn == "" {
|
||||
t.Skip("requires disposable MySQL")
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
|
||||
defer cancel()
|
||||
db, err := Open(ctx, dsn, 24, 8)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
if err := Migrate(ctx, db); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, outcome := range []string{"SUCCEEDED", "FAILED", "UNKNOWN", "CANCELLED"} {
|
||||
t.Run(outcome, func(t *testing.T) {
|
||||
f := newExecutionFixture(t, ctx, db)
|
||||
task := f.submit(1)
|
||||
owner := bindWorkflowOwner(t, f, task.ID, false)
|
||||
_, ref := f.claim()
|
||||
if _, err := f.a.Start(ctx, f.token, f.binding, executionID(t), ref); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// 暂停后在途结果仍需持久化；同一完成回执并发不能产生第二条内部事件。
|
||||
f.exec(`UPDATE pulse_execution_gates SET paused=TRUE WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, owner.gate)
|
||||
in := worker.FinishInput{Ref: ref, Outcome: outcome, Result: json.RawMessage(`{"n":9007199254740993}`)}
|
||||
if outcome != "SUCCEEDED" {
|
||||
in.ErrorCode = "CONTROLLED_RESULT"
|
||||
}
|
||||
if outcome == "CANCELLED" {
|
||||
f.exec(`UPDATE pulse_tasks SET cancel_requested_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, task.ID)
|
||||
}
|
||||
message := executionID(t)
|
||||
var wg sync.WaitGroup
|
||||
results := make(chan error, 8)
|
||||
for i := 0; i < 8; i++ {
|
||||
wg.Add(1)
|
||||
go func() { defer wg.Done(); _, err := f.a.Finish(ctx, f.token, f.binding, message, in); results <- err }()
|
||||
}
|
||||
wg.Wait()
|
||||
close(results)
|
||||
for err := range results {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
state := outcome
|
||||
if outcome == "UNKNOWN" {
|
||||
state = "NEEDS_REVIEW"
|
||||
}
|
||||
assertWorkflowInbox(t, f, owner, task.ID, state, string(in.Result))
|
||||
// 外部结果日志可独立保留/清理，不能改变内部接续输入。
|
||||
f.exec(`DELETE FROM orbit_result_events WHERE namespace_id=?`, f.resources.NamespaceID)
|
||||
var frozen string
|
||||
if err := db.QueryRowContext(ctx, `SELECT snapshot FROM weave_inbox WHERE namespace_id=? AND workflow_id=?`, f.resources.NamespaceID, owner.workflow).Scan(&frozen); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var result events.TaskResult
|
||||
if json.Unmarshal([]byte(frozen), &result) != nil || string(result.Result) != string(in.Result) {
|
||||
t.Fatal("internal result depended on external journal")
|
||||
}
|
||||
})
|
||||
}
|
||||
t.Run("poll_and_old_activation_keep_original_identity", func(t *testing.T) {
|
||||
f := newExecutionFixture(t, ctx, db)
|
||||
task := f.submit(1)
|
||||
owner := bindWorkflowOwner(t, f, task.ID, true)
|
||||
_, err := f.tasks.CancelTask(ctx, f.token, task.ID, "forbidden-direct", service.CancelInput{ExpectedRevision: task.Revision, Reason: "owner controls cancellation"})
|
||||
expectWorkerCode(t, err, "STATE_CONFLICT")
|
||||
_, ref := f.claim()
|
||||
if _, err := f.a.Start(ctx, f.token, f.binding, executionID(t), ref); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
next := executionID(t)
|
||||
f.exec(`INSERT INTO weave_activations(namespace_id,id,workflow_id,step_id,sequence,input) VALUES (?,?,?,?,2,'{}')`, f.resources.NamespaceID, next, owner.workflow, owner.step)
|
||||
f.exec(`UPDATE weave_steps SET current_activation_id=? WHERE namespace_id=? AND id=?`, next, f.resources.NamespaceID, owner.step)
|
||||
raw := `{"status":"DONE"}`
|
||||
if _, err := f.a.Finish(ctx, f.token, f.binding, executionID(t), worker.FinishInput{Ref: ref, Outcome: "SUCCEEDED", Result: json.RawMessage(raw)}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
assertWorkflowInbox(t, f, owner, task.ID, "SUCCEEDED", raw)
|
||||
})
|
||||
|
||||
t.Run("natural_lease_expiry_also_delivers", func(t *testing.T) {
|
||||
f := newExecutionFixture(t, ctx, db)
|
||||
task, err := f.tasks.Submit(ctx, f.token, string(executionID(t)), service.SubmitInput{QueueID: f.resources.QueueID, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"n":1}`), TaskTimeoutMS: 1500, Retry: pulse.RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: 1000}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
owner := bindWorkflowOwner(t, f, task.Task.ID, false)
|
||||
_, ref := f.claim()
|
||||
if _, err := f.a.Start(ctx, f.token, f.binding, executionID(t), ref); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for f.get(task.Task.ID).State == pulse.TaskRunning {
|
||||
if _, err := f.storeB.Recover(ctx, 100); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
t.Fatal(ctx.Err())
|
||||
case <-time.After(50 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
assertWorkflowInbox(t, f, owner, task.Task.ID, "NEEDS_REVIEW", "")
|
||||
if _, err := f.storeB.Recover(ctx, 100); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if f.count("weave_inbox", "workflow_id=?", owner.workflow) != 1 {
|
||||
t.Fatal("recovery duplicated result")
|
||||
}
|
||||
})
|
||||
|
||||
for _, table := range []string{"weave_inbox", "orbit_result_events"} {
|
||||
t.Run("rollback_"+table, func(t *testing.T) {
|
||||
f := newExecutionFixture(t, ctx, db)
|
||||
task := f.submit(1)
|
||||
owner := bindWorkflowOwner(t, f, task.ID, false)
|
||||
_, ref := f.claim()
|
||||
if _, err := f.a.Start(ctx, f.token, f.binding, executionID(t), ref); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
trigger := "orbit_test_workflow_delivery_failure"
|
||||
f.exec(`CREATE TRIGGER ` + trigger + ` BEFORE INSERT ON ` + table + ` FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='controlled workflow delivery failure'`)
|
||||
message := executionID(t)
|
||||
in := worker.FinishInput{Ref: ref, Outcome: "SUCCEEDED", Result: json.RawMessage(`{"n":9007199254740993}`)}
|
||||
_, err := f.a.Finish(ctx, f.token, f.binding, message, in)
|
||||
f.exec(`DROP TRIGGER ` + trigger)
|
||||
if err == nil || f.get(task.ID).State != pulse.TaskRunning || f.used() != 1 || f.count("weave_inbox", "workflow_id=?", owner.workflow) != 0 || f.count("orbit_events", "aggregate_id=? AND type='task.succeeded'", task.ID) != 0 || f.count("pulse_attempt_receipts", "attempt_id=? AND operation='finish'", ref.AttemptID) != 0 {
|
||||
t.Fatal("partial result committed", err)
|
||||
}
|
||||
out, err := f.a.Finish(ctx, f.token, f.binding, message, in)
|
||||
if err != nil || out.Deduplicated {
|
||||
t.Fatal("original retry failed", err)
|
||||
}
|
||||
assertWorkflowInbox(t, f, owner, task.ID, "SUCCEEDED", string(in.Result))
|
||||
})
|
||||
}
|
||||
for _, broken := range []string{"owner", "gate", "activation", "definition"} {
|
||||
t.Run("reject_broken_"+broken, func(t *testing.T) {
|
||||
f := newExecutionFixture(t, ctx, db)
|
||||
task := f.submit(1)
|
||||
owner := bindWorkflowOwner(t, f, task.ID, false)
|
||||
_, ref := f.claim()
|
||||
if _, err := f.a.Start(ctx, f.token, f.binding, executionID(t), ref); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var restore func()
|
||||
switch broken {
|
||||
case "owner":
|
||||
f.exec(`UPDATE pulse_tasks SET owner_kind=NULL,owner_id=NULL WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, task.ID)
|
||||
restore = func() {
|
||||
f.exec(`UPDATE pulse_tasks SET owner_kind=?,owner_id=? WHERE namespace_id=? AND id=?`, engine.TaskOwnerKind, owner.activation, f.resources.NamespaceID, task.ID)
|
||||
}
|
||||
case "gate":
|
||||
f.exec(`UPDATE pulse_tasks SET gate_id=NULL WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, task.ID)
|
||||
restore = func() {
|
||||
f.exec(`UPDATE pulse_tasks SET gate_id=? WHERE namespace_id=? AND id=?`, owner.gate, f.resources.NamespaceID, task.ID)
|
||||
}
|
||||
case "activation":
|
||||
f.exec(`UPDATE weave_activations SET task_id=NULL WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, owner.activation)
|
||||
restore = func() {
|
||||
f.exec(`UPDATE weave_activations SET task_id=? WHERE namespace_id=? AND id=?`, task.ID, f.resources.NamespaceID, owner.activation)
|
||||
}
|
||||
case "definition":
|
||||
var old core.ID
|
||||
if err := db.QueryRowContext(ctx, `SELECT definition_version_id FROM weave_instances WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, owner.workflow).Scan(&old); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
f.exec(`UPDATE weave_instances SET definition_version_id=? WHERE namespace_id=? AND id=?`, executionID(t), f.resources.NamespaceID, owner.workflow)
|
||||
restore = func() {
|
||||
f.exec(`UPDATE weave_instances SET definition_version_id=? WHERE namespace_id=? AND id=?`, old, f.resources.NamespaceID, owner.workflow)
|
||||
}
|
||||
}
|
||||
message := executionID(t)
|
||||
in := worker.FinishInput{Ref: ref, Outcome: "SUCCEEDED", Result: json.RawMessage(`{"n":1}`)}
|
||||
_, err := f.a.Finish(ctx, f.token, f.binding, message, in)
|
||||
restore()
|
||||
if err == nil || f.get(task.ID).State != pulse.TaskRunning {
|
||||
t.Fatal("broken association accepted", err)
|
||||
}
|
||||
if _, err := f.a.Finish(ctx, f.token, f.binding, message, in); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
assertWorkflowInbox(t, f, owner, task.ID, "SUCCEEDED", string(in.Result))
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,120 @@
|
||||
package mysql
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
core "git.yuntuo.hk/k2safe/orbit/model"
|
||||
driver "github.com/go-sql-driver/mysql"
|
||||
)
|
||||
|
||||
func TestIntegrationWorkflowInboxUpgrade(t *testing.T) {
|
||||
dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN")
|
||||
if dsn == "" {
|
||||
t.Skip("requires disposable MySQL")
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
||||
defer cancel()
|
||||
admin, err := Open(ctx, dsn, 2, 1)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer admin.Close()
|
||||
name := "orbit_workflow_upgrade_" + string(executionID(t))
|
||||
if _, err := admin.ExecContext(ctx, "CREATE DATABASE `"+name+"`"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer func() {
|
||||
cleanup, stop := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer stop()
|
||||
if _, err := admin.ExecContext(cleanup, "DROP DATABASE `"+name+"`"); err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
}()
|
||||
cfg, err := driver.ParseDSN(dsn)
|
||||
if err != nil {
|
||||
t.Fatal("invalid test DSN")
|
||||
}
|
||||
cfg.DBName = name
|
||||
db, err := Open(ctx, cfg.FormatDSN(), 4, 2)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
exec := func(q string, args ...any) {
|
||||
t.Helper()
|
||||
if _, err := db.ExecContext(ctx, q, args...); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
exec(`CREATE TABLE orbit_schema_migrations (version BIGINT UNSIGNED PRIMARY KEY,name VARCHAR(191) NOT NULL,checksum CHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,dirty BOOLEAN NOT NULL,applied_at DATETIME(6) NULL) ENGINE=InnoDB`)
|
||||
migrations, err := Migrations()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, m := range migrations {
|
||||
if m.Version > 16 {
|
||||
break
|
||||
}
|
||||
statements, err := splitSQL(m.SQL)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, q := range statements {
|
||||
exec(q)
|
||||
}
|
||||
exec(`INSERT INTO orbit_schema_migrations(version,name,checksum,dirty,applied_at) VALUES (?,?,?,FALSE,UTC_TIMESTAMP(6))`, m.Version, m.Name, m.Checksum)
|
||||
}
|
||||
ns, workflow := executionID(t), executionID(t)
|
||||
rows := map[string]core.ID{}
|
||||
for _, state := range []string{"PENDING", "LEASED", "PROCESSED", "QUARANTINED"} {
|
||||
id := executionID(t)
|
||||
rows[state] = id
|
||||
var lease *core.ID
|
||||
var until *time.Time
|
||||
if state == "LEASED" {
|
||||
v, tm := executionID(t), time.Now().UTC().Add(time.Minute)
|
||||
lease, until = &v, &tm
|
||||
}
|
||||
exec(`INSERT INTO weave_inbox(namespace_id,id,workflow_id,event_id,state,available_at,lease_owner,lease_until,lease_epoch) VALUES (?,?,?,?,?,UTC_TIMESTAMP(6),?,?,7)`, ns, id, workflow, executionID(t), state, lease, until)
|
||||
}
|
||||
if err := Migrate(ctx, db); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := CheckSchema(ctx, db); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for original, id := range rows {
|
||||
var state string
|
||||
var source, activation, owner *core.ID
|
||||
var version *uint64
|
||||
var snapshot []byte
|
||||
var code sql.NullString
|
||||
var epoch, revision uint64
|
||||
if err := db.QueryRowContext(ctx, `SELECT state,source_task_id,source_activation_id,source_revision,snapshot,lease_owner,error_code,lease_epoch,revision FROM weave_inbox WHERE namespace_id=? AND id=?`, ns, id).Scan(&state, &source, &activation, &version, &snapshot, &owner, &code, &epoch, &revision); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
expected := "QUARANTINED"
|
||||
expectedRevision := uint64(2)
|
||||
if original == "PROCESSED" {
|
||||
expected = "PROCESSED"
|
||||
expectedRevision = 1
|
||||
}
|
||||
if state != expected || source != nil || activation != nil || version != nil || snapshot != nil || owner != nil || epoch != 7 || revision != expectedRevision {
|
||||
t.Fatal("legacy record was reinterpreted", state)
|
||||
}
|
||||
if original != "PROCESSED" && code.String != "LEGACY_RESULT_UNVERIFIED" {
|
||||
t.Fatal("missing quarantine reason")
|
||||
}
|
||||
}
|
||||
if _, err := db.ExecContext(ctx, `INSERT INTO weave_inbox(namespace_id,id,workflow_id,event_id,available_at) VALUES (?,?,?,?,UTC_TIMESTAMP(6))`, ns, executionID(t), workflow, executionID(t)); err == nil {
|
||||
t.Fatal("new unverified pending input accepted")
|
||||
}
|
||||
if err := Migrate(ctx, db); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Log("v16 legacy inputs quarantined without reconstructed results or restored leases; processed history retained; current schema has zero foreign keys")
|
||||
}
|
||||
Reference in New Issue
Block a user