feat: trace task sources to fixed workflow activations and rounds
This commit is contained in:
@@ -10,6 +10,7 @@ import (
|
||||
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/engine"
|
||||
)
|
||||
|
||||
func (s *TaskStore) GetTaskSource(ctx context.Context, t auth.Token, id core.ID) (service.TaskSourceView, error) {
|
||||
@@ -18,7 +19,12 @@ func (s *TaskStore) GetTaskSource(ctx context.Context, t auth.Token, id core.ID)
|
||||
var queue core.ID
|
||||
var owner sql.NullString
|
||||
var ownerID, gate, retry *core.ID
|
||||
err := tx.QueryRowContext(ctx, `SELECT queue_id,owner_kind,owner_id,gate_id,retry_of FROM pulse_tasks WHERE namespace_id=? AND id=?`, t.NamespaceID, id).Scan(&queue, &owner, &ownerID, &gate, &retry)
|
||||
var scheduled, activated, polled bool
|
||||
err := tx.QueryRowContext(ctx, `SELECT queue_id,owner_kind,owner_id,gate_id,retry_of,
|
||||
EXISTS(SELECT 1 FROM pulse_schedule_occurrences o WHERE o.namespace_id=t.namespace_id AND o.task_id=t.id),
|
||||
EXISTS(SELECT 1 FROM weave_activations a WHERE a.namespace_id=t.namespace_id AND a.task_id=t.id),
|
||||
EXISTS(SELECT 1 FROM weave_poll_rounds r WHERE r.namespace_id=t.namespace_id AND r.task_id=t.id)
|
||||
FROM pulse_tasks t WHERE namespace_id=? AND id=?`, t.NamespaceID, id).Scan(&queue, &owner, &ownerID, &gate, &retry, &scheduled, &activated, &polled)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return fault.NotFound
|
||||
}
|
||||
@@ -28,8 +34,22 @@ func (s *TaskStore) GetTaskSource(ctx context.Context, t auth.Token, id core.ID)
|
||||
if !p.AllowsQueue(queue) {
|
||||
return fault.NotFound
|
||||
}
|
||||
// 来源字段由创建事务固定；当前未实现的门控/所有者不能被误判为普通受理。
|
||||
if gate != nil || owner.Valid != (ownerID != nil) || owner.Valid && retry != nil {
|
||||
// 反向关联同样必须存在且种类唯一；丢失所有者不能把流程任务降级为普通受理。
|
||||
if owner.Valid != (ownerID != nil) {
|
||||
return fault.Unavailable
|
||||
}
|
||||
if owner.Valid && owner.String == engine.TaskOwnerKind {
|
||||
if gate == nil || scheduled || activated == polled {
|
||||
return fault.Unavailable
|
||||
}
|
||||
source, err := readTaskWorkflowSource(ctx, tx, t.NamespaceID, id, queue, *ownerID, *gate, activated)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
out.Kind, out.Workflow = "WORKFLOW", &source
|
||||
return nil
|
||||
}
|
||||
if gate != nil || activated || polled || owner.Valid && retry != nil {
|
||||
return fault.Unavailable
|
||||
}
|
||||
if owner.Valid {
|
||||
@@ -56,11 +76,7 @@ func (s *TaskStore) GetTaskSource(ctx context.Context, t auth.Token, id core.ID)
|
||||
out.Schedule = &source
|
||||
return nil
|
||||
}
|
||||
var occurrences int
|
||||
if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM pulse_schedule_occurrences WHERE namespace_id=? AND task_id=?`, t.NamespaceID, id).Scan(&occurrences); err != nil {
|
||||
return err
|
||||
}
|
||||
if occurrences != 0 {
|
||||
if scheduled {
|
||||
return fault.Unavailable
|
||||
}
|
||||
if retry == nil {
|
||||
@@ -68,7 +84,10 @@ func (s *TaskStore) GetTaskSource(ctx context.Context, t auth.Token, id core.ID)
|
||||
return nil
|
||||
}
|
||||
var parent core.ID
|
||||
err = tx.QueryRowContext(ctx, `SELECT id FROM pulse_tasks WHERE namespace_id=? AND id=? AND id<>? AND queue_id=? AND owner_kind IS NULL AND owner_id IS NULL AND gate_id IS NULL AND state IN ('FAILED','CANCELLED')`, t.NamespaceID, *retry, id, queue).Scan(&parent)
|
||||
err = tx.QueryRowContext(ctx, `SELECT id FROM pulse_tasks t WHERE namespace_id=? AND id=? AND id<>? AND queue_id=? AND owner_kind IS NULL AND owner_id IS NULL AND gate_id IS NULL AND state IN ('FAILED','CANCELLED')
|
||||
AND NOT EXISTS(SELECT 1 FROM pulse_schedule_occurrences o WHERE o.namespace_id=t.namespace_id AND o.task_id=t.id)
|
||||
AND NOT EXISTS(SELECT 1 FROM weave_activations a WHERE a.namespace_id=t.namespace_id AND a.task_id=t.id)
|
||||
AND NOT EXISTS(SELECT 1 FROM weave_poll_rounds r WHERE r.namespace_id=t.namespace_id AND r.task_id=t.id)`, t.NamespaceID, *retry, id, queue).Scan(&parent)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return fault.Unavailable
|
||||
}
|
||||
|
||||
@@ -70,6 +70,12 @@ func TestIntegrationTaskSource(t *testing.T) {
|
||||
expectWorkerCode(t, err, "UNAVAILABLE")
|
||||
}
|
||||
f.exec(`UPDATE pulse_tasks SET retry_of=? WHERE namespace_id=? AND id=?`, original.ID, f.resources.NamespaceID, child.Task.ID)
|
||||
// 没有外键时，父任务即使丢失owner，反向流程关联仍能揭示其不是独立任务。
|
||||
activation := executionID(t)
|
||||
f.exec(`INSERT INTO weave_activations(namespace_id,id,workflow_id,step_id,sequence,task_id,input) VALUES (?,?,?,?,1,?,'{}')`, f.resources.NamespaceID, activation, executionID(t), executionID(t), original.ID)
|
||||
_, err = f.tasks.GetTaskSource(ctx, f.token, child.Task.ID)
|
||||
expectWorkerCode(t, err, "UNAVAILABLE")
|
||||
f.exec(`DELETE FROM weave_activations WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, activation)
|
||||
f.exec(`UPDATE orbit_credentials SET revoked_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, f.resources.CredentialID)
|
||||
_, err = f.tasks.GetTaskSource(ctx, f.token, child.Task.ID)
|
||||
expectWorkerCode(t, err, "UNAUTHENTICATED")
|
||||
|
||||
99
internal/storage/mysql/task_workflow_source.go
Normal file
99
internal/storage/mysql/task_workflow_source.go
Normal file
@@ -0,0 +1,99 @@
|
||||
package mysql
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
|
||||
"git.yuntuo.hk/k2safe/orbit/internal/fault"
|
||||
core "git.yuntuo.hk/k2safe/orbit/model"
|
||||
"git.yuntuo.hk/k2safe/orbit/pulse/service"
|
||||
"git.yuntuo.hk/k2safe/orbit/weave/engine"
|
||||
)
|
||||
|
||||
// readTaskWorkflowSource 只核验创建时固定的来源，不按当前激活替换历史，也不取得运行器锁。
|
||||
// 这里只向已授权任务暴露有限引用；完整流程及定义继续使用各自权限。
|
||||
func readTaskWorkflowSource(ctx context.Context, tx *sql.Tx, ns, task, queue, activation, gate core.ID, direct bool) (service.TaskWorkflowSource, error) {
|
||||
var out service.TaskWorkflowSource
|
||||
var activationTask *core.ID
|
||||
var policyRaw []byte
|
||||
var handler, handlerVersion, payloadVersion string
|
||||
err := tx.QueryRowContext(ctx, "SELECT a.workflow_id,i.definition_id,i.definition_version_id,v.version,s.id,s.`key`,s.kind,a.id,a.sequence,a.task_id,i.policy_snapshot,t.retry_of,t.handler,t.handler_version,t.payload_version FROM weave_activations a JOIN weave_steps s ON s.namespace_id=a.namespace_id AND s.workflow_id=a.workflow_id AND s.id=a.step_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_definitions d ON d.namespace_id=i.namespace_id AND d.id=i.definition_id JOIN weave_definition_versions v ON v.namespace_id=i.namespace_id AND v.definition_id=i.definition_id AND v.id=i.definition_version_id JOIN pulse_tasks t ON t.namespace_id=a.namespace_id AND t.id=? AND t.owner_kind='workflow_activation' AND t.owner_id=a.id AND t.gate_id=i.gate_id AND t.queue_id=? JOIN pulse_queues q ON q.namespace_id=t.namespace_id AND q.id=t.queue_id WHERE a.namespace_id=? AND a.id=?", gate, task, queue, ns, activation).Scan(&out.WorkflowID, &out.DefinitionID, &out.DefinitionVersionID, &out.Version, &out.StepID, &out.StepKey, &out.StepKind, &out.ActivationID, &out.ActivationSequence, &activationTask, &policyRaw, &out.RetryOf, &handler, &handlerVersion, &payloadVersion)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return out, fault.Unavailable
|
||||
}
|
||||
if err != nil {
|
||||
return out, err
|
||||
}
|
||||
version, err := readWorkflowVersion(ctx, tx, ns, out.DefinitionID, out.Version)
|
||||
if err != nil {
|
||||
return out, err
|
||||
}
|
||||
var policy workflowPolicy
|
||||
if version.ID != out.DefinitionVersionID || json.Unmarshal(policyRaw, &policy) != nil || policy != (workflowPolicy{1, version.ID, version.ContentDigest, version.Specification.FailurePolicy, version.Analysis.MaxTaskAttempts}) {
|
||||
return out, fault.Unavailable
|
||||
}
|
||||
program, err := engine.Compile(version.Specification)
|
||||
if err != nil {
|
||||
return out, err
|
||||
}
|
||||
node, ok := program.Step(out.StepKey)
|
||||
if !ok || node.Kind != out.StepKind || out.ActivationSequence == 0 {
|
||||
return out, fault.Unavailable
|
||||
}
|
||||
contract := node.Task
|
||||
wantsParent := out.ActivationSequence > 1
|
||||
switch out.StepKind {
|
||||
case "TASK":
|
||||
if !direct || activationTask == nil || *activationTask != task || contract == nil || out.ActivationSequence > contract.Retry.MaxAttempts {
|
||||
return out, fault.Unavailable
|
||||
}
|
||||
case "POLL":
|
||||
if direct || activationTask != nil || node.Poll == nil {
|
||||
return out, fault.Unavailable
|
||||
}
|
||||
contract = &node.Poll.Task
|
||||
var first, used, maximum uint32
|
||||
var roundID core.ID
|
||||
err = tx.QueryRowContext(ctx, "SELECT r.id,r.`round`,p.first_round,p.used_rounds,p.max_rounds 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=?", ns, out.WorkflowID, activation, task).Scan(&roundID, &out.PollRound, &first, &used, &maximum)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return out, fault.Unavailable
|
||||
}
|
||||
if err != nil {
|
||||
return out, err
|
||||
}
|
||||
if first == 0 || first > out.PollRound || out.PollRound > used || used > maximum || maximum != node.Poll.MaxRounds || out.ActivationSequence == 1 && first != 1 || out.ActivationSequence > maximum*contract.Retry.MaxAttempts {
|
||||
return out, fault.Unavailable
|
||||
}
|
||||
out.PollRoundID = &roundID
|
||||
wantsParent = wantsParent && out.PollRound == first
|
||||
default:
|
||||
return out, fault.Unavailable
|
||||
}
|
||||
if contract.QueueID != queue || contract.Handler != handler || contract.HandlerVersion != handlerVersion || contract.PayloadVersion != payloadVersion || wantsParent != (out.RetryOf != nil) {
|
||||
return out, fault.Unavailable
|
||||
}
|
||||
if wantsParent {
|
||||
if *out.RetryOf == task {
|
||||
return out, fault.Unavailable
|
||||
}
|
||||
// 只读直接父任务，使用步骤/代次唯一索引；不扫描或递归整个恢复历史。
|
||||
q := `SELECT COUNT(*) FROM weave_activations a JOIN pulse_tasks t ON t.namespace_id=a.namespace_id AND t.owner_kind='workflow_activation' AND t.owner_id=a.id AND t.gate_id=? AND t.queue_id=? AND t.id=? WHERE a.namespace_id=? AND a.workflow_id=? AND a.step_id=? AND a.sequence=? AND t.effect_uncertain=FALSE`
|
||||
args := []any{gate, queue, *out.RetryOf, ns, out.WorkflowID, out.StepID, out.ActivationSequence - 1}
|
||||
if out.StepKind == "TASK" {
|
||||
q += ` AND a.task_id=t.id AND t.state IN ('FAILED','CANCELLED')`
|
||||
} else {
|
||||
q += " AND a.task_id IS NULL AND t.state IN ('SUCCEEDED','FAILED','CANCELLED') AND EXISTS(SELECT 1 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=a.namespace_id AND r.workflow_id=a.workflow_id AND r.activation_id=a.id AND r.task_id=t.id AND r.`round`=? AND p.used_rounds=r.`round` AND p.first_round<=r.`round`)"
|
||||
args = append(args, out.PollRound)
|
||||
}
|
||||
var n int
|
||||
if err = tx.QueryRowContext(ctx, q, args...).Scan(&n); err != nil {
|
||||
return out, err
|
||||
}
|
||||
if n != 1 {
|
||||
return out, fault.Unavailable
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
172
internal/storage/mysql/task_workflow_source_integration_test.go
Normal file
172
internal/storage/mysql/task_workflow_source_integration_test.go
Normal file
@@ -0,0 +1,172 @@
|
||||
package mysql
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"reflect"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.yuntuo.hk/k2safe/orbit/internal/fault"
|
||||
core "git.yuntuo.hk/k2safe/orbit/model"
|
||||
"git.yuntuo.hk/k2safe/orbit/pulse/service"
|
||||
"git.yuntuo.hk/k2safe/orbit/weave/definition"
|
||||
"git.yuntuo.hk/k2safe/orbit/weave/instance"
|
||||
)
|
||||
|
||||
func assertWorkflowTaskSource(t *testing.T, f *executionFixture, v instance.View, task core.ID, sequence, round uint32, parent *core.ID) service.TaskSourceView {
|
||||
t.Helper()
|
||||
out, err := f.tasks.GetTaskSource(f.ctx, f.token, task)
|
||||
if err != nil || out.Kind != "WORKFLOW" || out.Workflow == nil || out.RetryOf != nil || out.Schedule != nil {
|
||||
t.Fatal("workflow source missing", out, err)
|
||||
}
|
||||
w := out.Workflow
|
||||
if w.WorkflowID != v.ID || w.DefinitionID != v.DefinitionID || w.DefinitionVersionID != v.DefinitionVersionID || w.Version != v.Version || w.ActivationSequence != sequence || w.PollRound != round || !reflect.DeepEqual(w.RetryOf, parent) {
|
||||
t.Fatal("wrong fixed source", w)
|
||||
}
|
||||
if round == 0 && (w.StepKind != "TASK" || w.PollRoundID != nil) || round > 0 && (w.StepKind != "POLL" || w.PollRoundID == nil) {
|
||||
t.Fatal("wrong step source kind", w)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func TestIntegrationTaskWorkflowSource(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()
|
||||
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)
|
||||
}
|
||||
t.Run("fixed_version_and_task_only_authorization", func(t *testing.T) {
|
||||
f := newExecutionFixture(t, ctx, db)
|
||||
otherQueue := executionID(t)
|
||||
f.exec(`INSERT INTO pulse_queues(namespace_id,id,name,max_concurrency,rate_per_second,policy) SELECT namespace_id,?,?,max_concurrency,rate_per_second,policy FROM pulse_queues WHERE namespace_id=? AND id=?`, otherQueue, string(otherQueue), f.resources.NamespaceID, f.resources.QueueID)
|
||||
spec := runtimeSpec(f.resources.QueueID, 2)
|
||||
spec.Body.Steps[1].Task.QueueID = otherQueue
|
||||
v := runtimeStart(t, f, spec)
|
||||
steps, err := instance.New(NewWorkflowInstanceStore(db)).Steps(ctx, f.token, v.ID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
task := *steps.Steps[0].TaskID
|
||||
before := f.get(task)
|
||||
original := assertWorkflowTaskSource(t, f, v, task, 1, 0, nil)
|
||||
published := spec
|
||||
published.TimeoutMS++
|
||||
if _, err = definition.New(NewWorkflowDefinitionStore(db)).Publish(ctx, f.token, v.DefinitionID, "v2", definition.PublishInput{Version: "v2", Specification: published, Reason: "new publication must not retarget source"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
f.exec(`UPDATE orbit_credentials SET permissions=JSON_OBJECT('actions',JSON_ARRAY('tasks:read'),'queues',JSON_ARRAY(?)) WHERE namespace_id=? AND id=?`, string(f.resources.QueueID), f.resources.NamespaceID, f.resources.CredentialID)
|
||||
after := assertWorkflowTaskSource(t, f, v, task, 1, 0, nil)
|
||||
if !reflect.DeepEqual(original, after) || !reflect.DeepEqual(before, f.get(task)) {
|
||||
t.Fatal("source read or publication changed original task")
|
||||
}
|
||||
if _, err = instance.New(NewWorkflowInstanceStore(db)).Get(ctx, f.token, v.ID); fault.Code(err) != "PERMISSION_DENIED" {
|
||||
t.Fatal("source granted full workflow read", err)
|
||||
}
|
||||
if _, err = definition.New(NewWorkflowDefinitionStore(db)).Version(ctx, f.token, v.DefinitionID, "v1"); fault.Code(err) != "PERMISSION_DENIED" {
|
||||
t.Fatal("source granted definition read", err)
|
||||
}
|
||||
other := newExecutionFixture(t, ctx, db)
|
||||
if _, err = other.tasks.GetTaskSource(ctx, other.token, task); fault.Code(err) != "NOT_FOUND" {
|
||||
t.Fatal("cross namespace source", err)
|
||||
}
|
||||
f.exec(`UPDATE orbit_credentials SET permissions=JSON_SET(permissions,'$.queues',JSON_ARRAY(?)) WHERE namespace_id=? AND id=?`, string(otherQueue), f.resources.NamespaceID, f.resources.CredentialID)
|
||||
if _, err = f.tasks.GetTaskSource(ctx, f.token, task); fault.Code(err) != "NOT_FOUND" {
|
||||
t.Fatal("wrong queue source", err)
|
||||
}
|
||||
})
|
||||
t.Run("damaged_associations_never_become_direct_sources", func(t *testing.T) {
|
||||
f := newExecutionFixture(t, ctx, db)
|
||||
v := runtimeStart(t, f, runtimeSpec(f.resources.QueueID, 1))
|
||||
steps, err := instance.New(NewWorkflowInstanceStore(db)).Steps(ctx, f.token, v.ID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
step := steps.Steps[0]
|
||||
task, activation := *step.TaskID, *step.ActivationID
|
||||
original := assertWorkflowTaskSource(t, f, v, task, 1, 0, nil)
|
||||
var gate core.ID
|
||||
var policy string
|
||||
if err = db.QueryRowContext(ctx, `SELECT gate_id,policy_snapshot FROM weave_instances WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, v.ID).Scan(&gate, &policy); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
broken := func() {
|
||||
t.Helper()
|
||||
out, err := f.tasks.GetTaskSource(ctx, f.token, task)
|
||||
if fault.Code(err) != "UNAVAILABLE" || !reflect.DeepEqual(out, service.TaskSourceView{}) {
|
||||
t.Fatal("invalid source leaked", out, err)
|
||||
}
|
||||
}
|
||||
for _, tc := range []struct {
|
||||
table, field string
|
||||
value, restore any
|
||||
}{
|
||||
{"pulse_tasks", "owner_id", executionID(t), activation},
|
||||
{"pulse_tasks", "gate_id", executionID(t), gate},
|
||||
{"pulse_tasks", "handler_version", "mismatched", f.get(task).HandlerVersion},
|
||||
{"weave_activations", "workflow_id", executionID(t), v.ID},
|
||||
{"weave_activations", "step_id", executionID(t), step.ID},
|
||||
{"weave_activations", "task_id", nil, task},
|
||||
{"weave_steps", "kind", "SIGNAL", "TASK"},
|
||||
{"weave_instances", "definition_version_id", executionID(t), v.DefinitionVersionID},
|
||||
{"weave_instances", "policy_snapshot", `{}`, policy},
|
||||
} {
|
||||
t.Run(tc.table+"/"+tc.field, func(t *testing.T) {
|
||||
id := map[string]core.ID{"pulse_tasks": task, "weave_activations": activation, "weave_steps": step.ID, "weave_instances": v.ID}[tc.table]
|
||||
q := "UPDATE " + tc.table + " SET " + tc.field + "=? WHERE namespace_id=? AND id=?"
|
||||
f.exec(q, tc.value, f.resources.NamespaceID, id)
|
||||
broken()
|
||||
f.exec(q, tc.restore, f.resources.NamespaceID, id)
|
||||
})
|
||||
}
|
||||
f.exec(`UPDATE pulse_tasks SET owner_kind=NULL,owner_id=NULL,gate_id=NULL WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, task)
|
||||
broken()
|
||||
f.exec(`UPDATE pulse_tasks SET owner_kind='workflow_activation',owner_id=?,gate_id=? WHERE namespace_id=? AND id=?`, activation, gate, f.resources.NamespaceID, task)
|
||||
round := executionID(t)
|
||||
f.exec("INSERT INTO weave_poll_rounds(namespace_id,id,workflow_id,activation_id,`round`,task_id) VALUES (?,?,?,?,1,?)", f.resources.NamespaceID, round, v.ID, activation, task)
|
||||
broken()
|
||||
f.exec(`DELETE FROM weave_poll_rounds WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, round)
|
||||
if after := assertWorkflowTaskSource(t, f, v, task, 1, 0, nil); !reflect.DeepEqual(original, after) {
|
||||
t.Fatal("restored source changed")
|
||||
}
|
||||
})
|
||||
t.Run("recovered_poll_parent_must_be_same_step_and_round", func(t *testing.T) {
|
||||
f, v, ref := failedPoll(t, ctx, db, false)
|
||||
old := assertWorkflowTaskSource(t, f, v, ref.TaskID, 1, 1, nil)
|
||||
child := retryPollStep(t, f, v.ID)
|
||||
current := assertWorkflowTaskSource(t, f, v, child.Task.ID, 2, 1, &ref.TaskID)
|
||||
if current.Workflow.ActivationID == old.Workflow.ActivationID {
|
||||
t.Fatal("recovery source collapsed activation history")
|
||||
}
|
||||
if after := assertWorkflowTaskSource(t, f, v, ref.TaskID, 1, 1, nil); !reflect.DeepEqual(old, after) {
|
||||
t.Fatal("old source retargeted")
|
||||
}
|
||||
stranger := executionID(t)
|
||||
f.exec(`UPDATE pulse_tasks SET retry_of=? WHERE namespace_id=? AND id=?`, stranger, f.resources.NamespaceID, child.Task.ID)
|
||||
if _, err = f.tasks.GetTaskSource(ctx, f.token, child.Task.ID); fault.Code(err) != "UNAVAILABLE" {
|
||||
t.Fatal("unrelated retry parent accepted", err)
|
||||
}
|
||||
f.exec(`UPDATE pulse_tasks SET retry_of=? WHERE namespace_id=? AND id=?`, ref.TaskID, f.resources.NamespaceID, child.Task.ID)
|
||||
f.exec("UPDATE weave_poll_rounds SET `round`=2 WHERE namespace_id=? AND task_id=?", f.resources.NamespaceID, ref.TaskID)
|
||||
if _, err = f.tasks.GetTaskSource(ctx, f.token, child.Task.ID); fault.Code(err) != "UNAVAILABLE" {
|
||||
t.Fatal("different parent round accepted", err)
|
||||
}
|
||||
f.exec("UPDATE weave_poll_rounds SET `round`=1 WHERE namespace_id=? AND task_id=?", f.resources.NamespaceID, ref.TaskID)
|
||||
f.exec(`UPDATE pulse_tasks SET owner_kind=NULL,owner_id=NULL,gate_id=NULL,retry_of=NULL WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, child.Task.ID)
|
||||
if _, err = f.tasks.GetTaskSource(ctx, f.token, child.Task.ID); fault.Code(err) != "UNAVAILABLE" {
|
||||
t.Fatal("poll downgraded to direct", err)
|
||||
}
|
||||
// 明确保持数据库无外键；这里只回滚测试注入，绝不改写生产历史。
|
||||
f.exec(`UPDATE pulse_tasks SET owner_kind='workflow_activation',owner_id=?,gate_id=(SELECT gate_id FROM weave_instances WHERE namespace_id=? AND id=?),retry_of=? WHERE namespace_id=? AND id=?`, current.Workflow.ActivationID, f.resources.NamespaceID, v.ID, ref.TaskID, f.resources.NamespaceID, child.Task.ID)
|
||||
assertWorkflowTaskSource(t, f, v, child.Task.ID, 2, 1, &ref.TaskID)
|
||||
})
|
||||
}
|
||||
@@ -70,8 +70,10 @@ func TestIntegrationWorkflowPollRecovery(t *testing.T) {
|
||||
t.Run("retry_same_round_then_next_round_and_retry_again", func(t *testing.T) {
|
||||
f, v, ref := failedPoll(t, ctx, db, false)
|
||||
first := f.get(ref.TaskID)
|
||||
assertWorkflowTaskSource(t, f, v, first.ID, 1, 1, nil)
|
||||
original := recoveryStep(t, f, v.ID, "query")
|
||||
out := retryPollStep(t, f, v.ID)
|
||||
assertWorkflowTaskSource(t, f, v, out.Task.ID, 2, 1, &first.ID)
|
||||
if out.Task.Retry.MaxAttempts != 2 || !out.Task.DeadlineAt.Equal(first.DeadlineAt) || *out.Task.RetryOf != first.ID || *out.Step.ActivationID == original.ActivationID || out.Step.TaskID != nil || !sameWorkflowJSON(out.Task.Payload, first.Payload) {
|
||||
t.Fatal("recovery reset original identity, task deadline or budget")
|
||||
}
|
||||
@@ -91,12 +93,14 @@ func TestIntegrationWorkflowPollRecovery(t *testing.T) {
|
||||
runClock(t, f, c)
|
||||
ref = retryStart(t, f)
|
||||
second := f.get(ref.TaskID)
|
||||
assertWorkflowTaskSource(t, f, v, second.ID, 2, 2, nil)
|
||||
if second.Retry.MaxAttempts != 3 || second.RetryOf != nil || second.BusinessKey != first.BusinessKey || second.ConcurrencyKey != first.ConcurrencyKey {
|
||||
t.Fatal("next business round did not receive independent original policy")
|
||||
}
|
||||
retryFinish(t, f, ref, "FAILED", false)
|
||||
runtimeConsume(t, f, newRuntime(t, db), ref.TaskID)
|
||||
out = retryPollStep(t, f, v.ID)
|
||||
assertWorkflowTaskSource(t, f, v, out.Task.ID, 3, 2, &second.ID)
|
||||
if out.Task.Retry.MaxAttempts != 2 || !out.Task.DeadlineAt.Equal(second.DeadlineAt) || *out.Task.RetryOf != second.ID || f.count("weave_poll_states", "first_round=2 AND used_rounds=2 AND first_round_attempts_used=1") != 1 {
|
||||
t.Fatal("second round recovery reset logical round")
|
||||
}
|
||||
|
||||
@@ -132,6 +132,7 @@ func TestIntegrationWorkflowRecovery(t *testing.T) {
|
||||
retryFinish(t, f, ref, "FAILED", false)
|
||||
runtimeConsume(t, f, newRuntime(t, db), ref.TaskID)
|
||||
oldTask, _ := f.tasks.GetTask(ctx, f.token, ref.TaskID)
|
||||
oldSource := assertWorkflowTaskSource(t, f, v, ref.TaskID, 1, 0, nil)
|
||||
in := recovery.RetryStepInput{StepDecision: recoveryStep(t, f, v.ID, "task0"), SafeToRetry: true}
|
||||
key := string(executionID(t))
|
||||
out, err := api.RetryStep(ctx, f.token, v.ID, key, in)
|
||||
@@ -141,6 +142,11 @@ func TestIntegrationWorkflowRecovery(t *testing.T) {
|
||||
if out.Task.ID == ref.TaskID || out.Task.RetryOf == nil || *out.Task.RetryOf != ref.TaskID || out.Task.Retry.MaxAttempts != oldTask.Retry.MaxAttempts-1 || !out.Task.DeadlineAt.Equal(oldTask.DeadlineAt) || !sameWorkflowJSON(out.Task.Payload, oldTask.Payload) || *out.Step.ActivationID == in.ActivationID {
|
||||
t.Fatal("retry replaced identity or reset budget")
|
||||
}
|
||||
newSource := assertWorkflowTaskSource(t, f, v, out.Task.ID, 2, 0, &ref.TaskID)
|
||||
if newSource.Workflow.ActivationID == oldSource.Workflow.ActivationID {
|
||||
t.Fatal("retry source lost generation")
|
||||
}
|
||||
assertWorkflowTaskSource(t, f, v, ref.TaskID, 1, 0, nil)
|
||||
if f.count("weave_activations", "id=? AND state='FAILED'", in.ActivationID) != 1 {
|
||||
t.Fatal("old activation rewritten")
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user