feat: expose bounded workflow activation and signal history
This commit is contained in:
@@ -63,7 +63,8 @@ func TestIntegrationApplicationResourcesUpgrade(t *testing.T) {
|
||||
finished := retryStart(t, g)
|
||||
retryFinish(t, g, finished, "SUCCEEDED", false)
|
||||
assertApplicationCounts(t, f, 4, 2, 1)
|
||||
// 还原后续新增表和字段，得到真实 v9 结构；现有任务内容保持不变。
|
||||
// 还原后续新增表、字段和索引，得到真实 v9 结构；现有任务内容保持不变。
|
||||
f.exec(`ALTER TABLE weave_signals DROP INDEX ix_signal_history`)
|
||||
f.exec(`ALTER TABLE weave_poll_states DROP CHECK ck_poll_recovery_budget, DROP COLUMN first_round, DROP COLUMN first_round_attempts_used`)
|
||||
f.exec(`ALTER TABLE pulse_tasks DROP INDEX ix_task_workflow_owner`)
|
||||
f.exec(`ALTER TABLE weave_instances DROP CHECK ck_workflow_stop_intent, DROP COLUMN stop_kind, DROP COLUMN stop_requested_at`)
|
||||
|
||||
@@ -0,0 +1,3 @@
|
||||
-- 信号按实例及原受理时间分页；不改变历史内容、截止、幂等或消费状态，不增加外键。
|
||||
ALTER TABLE weave_signals
|
||||
ADD KEY ix_signal_history (namespace_id,workflow_id,accepted_at,id);
|
||||
@@ -89,7 +89,7 @@ func TestIntegrationNoForeignKeysUpgrade(t *testing.T) {
|
||||
indexes := func() []string {
|
||||
t.Helper()
|
||||
// v15替换retry_of索引、v17增加内部结果唯一索引、v20增加信号索引，下方单独验证定义。
|
||||
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','ix_signal_expiry','ix_signal_activation','uq_wait_winner_signal','ix_task_workflow_owner') ORDER BY TABLE_NAME,INDEX_NAME,SEQ_IN_INDEX`)
|
||||
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','ix_signal_expiry','ix_signal_activation','uq_wait_winner_signal','ix_task_workflow_owner','ix_signal_history') ORDER BY TABLE_NAME,INDEX_NAME,SEQ_IN_INDEX`)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -180,6 +180,7 @@ func TestIntegrationNoForeignKeysUpgrade(t *testing.T) {
|
||||
var collation string
|
||||
for _, index := range []struct{ table, name, want string }{
|
||||
{"pulse_tasks", "ix_task_workflow_owner", "namespace_id:1,owner_kind:1,owner_id:1,gate_id:1,id:1"},
|
||||
{"weave_signals", "ix_signal_history", "namespace_id:1,workflow_id:1,accepted_at:1,id:1"},
|
||||
{"weave_signals", "ix_signal_expiry", "state:1,available_at:1,namespace_id:1,id:1"},
|
||||
{"weave_signals", "ix_signal_activation", "namespace_id:1,workflow_id:1,target_step_key:1,target_activation_sequence:1,state:1,accepted_at:1,id:1"},
|
||||
{"weave_waits", "uq_wait_winner_signal", "namespace_id:0,winner_signal_id:0"},
|
||||
|
||||
29
internal/storage/mysql/no_foreign_keys_test.go
Normal file
29
internal/storage/mysql/no_foreign_keys_test.go
Normal file
@@ -0,0 +1,29 @@
|
||||
package mysql
|
||||
|
||||
import (
|
||||
"regexp"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// 已应用的历史文件不改写；移除迁移之后的DDL不能再引入数据库引用约束。
|
||||
func TestMigrationsAfterForeignKeyRemovalDoNotReintroduceReferences(t *testing.T) {
|
||||
migrations, err := Migrations()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
forbidden := regexp.MustCompile(`(?i)\b(FOREIGN\s+KEY|REFERENCES)\b`)
|
||||
for _, m := range migrations {
|
||||
if m.Version <= 8 {
|
||||
continue
|
||||
}
|
||||
statements, err := splitSQL(m.SQL)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, statement := range statements {
|
||||
if forbidden.MatchString(statement) {
|
||||
t.Errorf("migration %d introduces forbidden reference DDL", m.Version)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -61,8 +61,8 @@ func decodeHistoryCursorKind(value string, ns, task, attempt core.ID, kind strin
|
||||
if d.Decode(&c) != nil || c.Version != 1 || c.Namespace != ns || c.Task != task || c.Attempt != attempt {
|
||||
return c, fault.Invalid("cursor does not match this history query")
|
||||
}
|
||||
if kind == "attempts" {
|
||||
if c.Kind != "attempts" || c.Epoch == 0 || c.ID != "" || c.Created != nil {
|
||||
if kind == "attempts" || kind == "workflow-activations" || kind == "workflow-rounds" {
|
||||
if c.Kind != kind || c.Epoch == 0 || kind != "attempts" && c.Epoch > 1<<32-1 || c.ID != "" || c.Created != nil {
|
||||
return c, fault.Invalid("invalid attempts cursor")
|
||||
}
|
||||
} else {
|
||||
|
||||
@@ -8,6 +8,7 @@ import (
|
||||
"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/definition"
|
||||
"git.yuntuo.hk/k2safe/orbit/weave/engine"
|
||||
"git.yuntuo.hk/k2safe/orbit/weave/instance"
|
||||
)
|
||||
@@ -37,43 +38,51 @@ func (s *WorkflowInstanceStore) Step(ctx context.Context, t auth.Token, id core.
|
||||
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
|
||||
out.Poll, out.Wait, err = readWorkflowActivationAttachments(ctx, tx, t.NamespaceID, id, gate, key, string(out.Step.Kind), *a, out.Step.ActivationSequence, version)
|
||||
return err
|
||||
})
|
||||
return out, err
|
||||
}
|
||||
|
||||
// readWorkflowActivationAttachments 查询指定激活的等待与预算，不替换为当前激活。
|
||||
func readWorkflowActivationAttachments(ctx context.Context, tx *sql.Tx, ns, workflow, gate core.ID, key, kind string, activation core.ID, sequence uint32, version definition.VersionView) (*instance.PollView, *instance.WaitView, error) {
|
||||
var poll *instance.PollView
|
||||
var wait *instance.WaitView
|
||||
program, err := engine.Compile(version.Specification)
|
||||
if err != nil {
|
||||
return nil, nil, 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, ns, workflow, activation).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 nil, nil, 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 nil, nil, errors.New("workflow poll detail ownership or policy invalid")
|
||||
}
|
||||
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=?`, ns, workflow, activation).Scan(&w.ID, &w.Kind, &w.State, &w.Name, &w.PayloadVersion, &w.CorrelationKey, &w.DeadlineAt, &w.WinnerSignalID, &w.ResolvedAt); err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
if w.Kind != kind {
|
||||
return nil, nil, 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'`, ns, workflow, w.WinnerSignalID, w.ID, key, sequence).Scan(&n); err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
if n != 1 {
|
||||
return nil, nil, errors.New("workflow wait winner association invalid")
|
||||
}
|
||||
}
|
||||
wait = w
|
||||
}
|
||||
return poll, wait, nil
|
||||
}
|
||||
|
||||
206
internal/storage/mysql/workflow_history.go
Normal file
206
internal/storage/mysql/workflow_history.go
Normal file
@@ -0,0 +1,206 @@
|
||||
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/definition"
|
||||
"git.yuntuo.hk/k2safe/orbit/weave/engine"
|
||||
"git.yuntuo.hk/k2safe/orbit/weave/instance"
|
||||
)
|
||||
|
||||
type workflowHistoryParent struct {
|
||||
view instance.View
|
||||
gate core.ID
|
||||
version definition.VersionView
|
||||
step instance.StepView
|
||||
node definition.Step
|
||||
}
|
||||
|
||||
func readWorkflowHistoryParent(ctx context.Context, tx *sql.Tx, ns, id core.ID, key string, p auth.Permissions) (workflowHistoryParent, error) {
|
||||
var h workflowHistoryParent
|
||||
var err error
|
||||
h.view, h.gate, h.version, err = readWorkflowForQuery(ctx, tx, ns, id, p)
|
||||
if err != nil {
|
||||
return h, err
|
||||
}
|
||||
steps, err := readWorkflowStepViews(ctx, tx, ns, id, h.gate, h.version, key)
|
||||
if err != nil {
|
||||
return h, err
|
||||
}
|
||||
if len(steps) != 1 {
|
||||
return h, fault.NotFound
|
||||
}
|
||||
h.step = steps[0]
|
||||
program, err := engine.Compile(h.version.Specification)
|
||||
if err != nil {
|
||||
return h, err
|
||||
}
|
||||
h.node, _ = program.Step(key)
|
||||
return h, nil
|
||||
}
|
||||
|
||||
const activationHistorySelect = `SELECT a.id,a.step_id,a.sequence,a.state,a.revision,a.task_id,a.created_at,a.updated_at,a.completed_at,t.id,t.queue_id FROM weave_activations a LEFT JOIN pulse_tasks t ON t.namespace_id=a.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 a.namespace_id=? AND a.workflow_id=? AND a.step_id=?`
|
||||
|
||||
func scanWorkflowActivation(row scanner, h workflowHistoryParent) (instance.ActivationSummary, error) {
|
||||
var a instance.ActivationSummary
|
||||
var task, queue *core.ID
|
||||
err := row.Scan(&a.ID, &a.StepID, &a.Sequence, &a.State, &a.Revision, &a.TaskID, &a.CreatedAt, &a.UpdatedAt, &a.CompletedAt, &task, &queue)
|
||||
if err != nil {
|
||||
return a, err
|
||||
}
|
||||
if a.StepID != h.step.ID || a.Sequence == 0 || (h.node.Task != nil) != (a.TaskID != nil) {
|
||||
return a, fault.Unavailable
|
||||
}
|
||||
if a.TaskID != nil && (task == nil || queue == nil || *task != *a.TaskID || *queue != h.node.Task.QueueID) {
|
||||
return a, fault.Unavailable
|
||||
}
|
||||
return a, nil
|
||||
}
|
||||
func readHistoryActivation(ctx context.Context, tx *sql.Tx, h workflowHistoryParent, activation core.ID) (instance.ActivationSummary, error) {
|
||||
a, err := scanWorkflowActivation(tx.QueryRowContext(ctx, activationHistorySelect+` AND a.id=?`, h.gate, h.view.NamespaceID, h.view.ID, h.step.ID, activation), h)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return a, fault.NotFound
|
||||
}
|
||||
return a, err
|
||||
}
|
||||
|
||||
func (s *WorkflowInstanceStore) ListActivations(ctx context.Context, t auth.Token, id core.ID, key string, in service.HistoryPageInput) (instance.ActivationPage, error) {
|
||||
out := instance.ActivationPage{InstanceID: id, Items: []instance.ActivationSummary{}}
|
||||
err := s.transaction(ctx, t, instance.ReadAction, func(tx *sql.Tx, p auth.Permissions) error {
|
||||
h, err := readWorkflowHistoryParent(ctx, tx, t.NamespaceID, id, key, p)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
out.Revision, out.StepID, out.CurrentActivationID = h.view.Revision, h.step.ID, h.step.ActivationID
|
||||
q := activationHistorySelect
|
||||
args := []any{h.gate, t.NamespaceID, id, h.step.ID}
|
||||
if in.Cursor != "" {
|
||||
c, err := decodeHistoryCursorKind(in.Cursor, t.NamespaceID, id, h.step.ID, "workflow-activations")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
q += ` AND a.sequence<?`
|
||||
args = append(args, c.Epoch)
|
||||
}
|
||||
q += ` ORDER BY a.sequence DESC LIMIT ?`
|
||||
args = append(args, in.Limit+1)
|
||||
rows, err := tx.QueryContext(ctx, q, args...)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
a, err := scanWorkflowActivation(rows, h)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
out.Items = append(out.Items, a)
|
||||
}
|
||||
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-activations", Namespace: t.NamespaceID, Task: id, Attempt: h.step.ID, Epoch: uint64(last.Sequence)})
|
||||
}
|
||||
return err
|
||||
})
|
||||
if err != nil {
|
||||
return instance.ActivationPage{}, err
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
func (s *WorkflowInstanceStore) GetActivation(ctx context.Context, t auth.Token, id core.ID, key string, activation core.ID) (instance.ActivationDetail, error) {
|
||||
out := instance.ActivationDetail{InstanceID: id}
|
||||
err := s.transaction(ctx, t, instance.ReadAction, func(tx *sql.Tx, p auth.Permissions) error {
|
||||
h, err := readWorkflowHistoryParent(ctx, tx, t.NamespaceID, id, key, p)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
out.Revision, out.Step = h.view.Revision, h.step
|
||||
out.Activation, err = readHistoryActivation(ctx, tx, h, activation)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var input, output []byte
|
||||
if err = tx.QueryRowContext(ctx, `SELECT input,output FROM weave_activations WHERE namespace_id=? AND workflow_id=? AND step_id=? AND id=?`, t.NamespaceID, id, h.step.ID, activation).Scan(&input, &output); err != nil {
|
||||
return err
|
||||
}
|
||||
out.Input, out.Output = input, output
|
||||
out.Poll, out.Wait, err = readWorkflowActivationAttachments(ctx, tx, t.NamespaceID, id, h.gate, key, string(h.step.Kind), activation, out.Activation.Sequence, h.version)
|
||||
return err
|
||||
})
|
||||
if err != nil {
|
||||
return instance.ActivationDetail{}, err
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
func (s *WorkflowInstanceStore) ListPollRounds(ctx context.Context, t auth.Token, id core.ID, key string, activation core.ID, in service.HistoryPageInput) (instance.PollRoundPage, error) {
|
||||
out := instance.PollRoundPage{InstanceID: id, ActivationID: activation, Items: []instance.PollRoundSummary{}}
|
||||
err := s.transaction(ctx, t, instance.ReadAction, func(tx *sql.Tx, p auth.Permissions) error {
|
||||
h, err := readWorkflowHistoryParent(ctx, tx, t.NamespaceID, id, key, p)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
a, err := readHistoryActivation(ctx, tx, h, activation)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if h.node.Poll == nil {
|
||||
return fault.Invalid("round history requires a POLL activation")
|
||||
}
|
||||
out.Revision = h.view.Revision
|
||||
poll, _, err := readWorkflowActivationAttachments(ctx, tx, t.NamespaceID, id, h.gate, key, string(h.step.Kind), activation, a.Sequence, h.version)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
q := "SELECT r.id,r.activation_id,r.`round`,r.task_id,r.created_at,t.id,t.queue_id FROM weave_poll_rounds r 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 r.namespace_id=? AND r.workflow_id=? AND r.activation_id=?"
|
||||
args := []any{h.gate, t.NamespaceID, id, activation}
|
||||
if in.Cursor != "" {
|
||||
c, err := decodeHistoryCursorKind(in.Cursor, t.NamespaceID, id, activation, "workflow-rounds")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
q += " AND r.`round`<?"
|
||||
args = append(args, c.Epoch)
|
||||
}
|
||||
q += " ORDER BY r.`round` DESC LIMIT ?"
|
||||
args = append(args, in.Limit+1)
|
||||
rows, err := tx.QueryContext(ctx, q, args...)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var round instance.PollRoundSummary
|
||||
var task, queue *core.ID
|
||||
if err = rows.Scan(&round.ID, &round.ActivationID, &round.Round, &round.TaskID, &round.CreatedAt, &task, &queue); err != nil {
|
||||
return err
|
||||
}
|
||||
if task == nil || queue == nil || *task != round.TaskID || *queue != h.node.Poll.Task.QueueID || round.Round < poll.FirstRound || round.Round > poll.UsedRounds {
|
||||
return fault.Unavailable
|
||||
}
|
||||
out.Items = append(out.Items, round)
|
||||
}
|
||||
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-rounds", Namespace: t.NamespaceID, Task: id, Attempt: activation, Epoch: uint64(last.Round)})
|
||||
}
|
||||
return err
|
||||
})
|
||||
if err != nil {
|
||||
return instance.PollRoundPage{}, err
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
251
internal/storage/mysql/workflow_history_integration_test.go
Normal file
251
internal/storage/mysql/workflow_history_integration_test.go
Normal file
@@ -0,0 +1,251 @@
|
||||
package mysql
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"os"
|
||||
"strings"
|
||||
"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/instance"
|
||||
"git.yuntuo.hk/k2safe/orbit/weave/recovery"
|
||||
)
|
||||
|
||||
func TestIntegrationWorkflowHistory(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, 16, 4)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
if err = Migrate(ctx, db); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
api := instance.New(NewWorkflowInstanceStore(db))
|
||||
t.Run("task_generations_input_pagination_and_no_foreign_key_corruption", func(t *testing.T) {
|
||||
f := newExecutionFixture(t, ctx, db)
|
||||
spec := runtimeSpec(f.resources.QueueID, 2)
|
||||
spec.Body.Steps[0].Task.Retry.MaxAttempts = 2
|
||||
spec.Body.Steps[0].Task.Retry.InitialDelayMS = 1000
|
||||
spec.Body.Steps[0].Task.Retry.MaxDelayMS = 1000
|
||||
v := runtimeStart(t, f, spec)
|
||||
grantWorkflowRecovery(f)
|
||||
empty, err := api.ListActivations(ctx, f.token, v.ID, "task1", service.HistoryPageInput{})
|
||||
if err != nil || empty.Items == nil || len(empty.Items) != 0 || empty.CurrentActivationID != nil {
|
||||
t.Fatal("unactivated history", err)
|
||||
}
|
||||
ref := retryStart(t, f)
|
||||
retryFinish(t, f, ref, "FAILED", false)
|
||||
runtimeConsume(t, f, newRuntime(t, db), ref.TaskID)
|
||||
old := recoveryStep(t, f, v.ID, "task0")
|
||||
before, err := api.GetActivation(ctx, f.token, v.ID, "task0", old.ActivationID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
child, err := recovery.New(NewWorkflowRecoveryStore(db)).RetryStep(ctx, f.token, v.ID, string(executionID(t)), recovery.RetryStepInput{StepDecision: old, SafeToRetry: true})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
page, err := api.ListActivations(ctx, f.token, v.ID, "task0", service.HistoryPageInput{Limit: 1})
|
||||
if err != nil || len(page.Items) != 1 || page.Items[0].Sequence != 2 || page.NextCursor == "" || *page.CurrentActivationID != *child.Step.ActivationID {
|
||||
t.Fatal("latest page", err)
|
||||
}
|
||||
older, err := api.ListActivations(ctx, f.token, v.ID, "task0", service.HistoryPageInput{Limit: 1, Cursor: page.NextCursor})
|
||||
if err != nil || len(older.Items) != 1 || older.Items[0].ID != old.ActivationID || older.NextCursor != "" {
|
||||
t.Fatal("historical page", err)
|
||||
}
|
||||
detail, err := api.GetActivation(ctx, f.token, v.ID, "task0", old.ActivationID)
|
||||
if err != nil || detail.Activation.ID != old.ActivationID || detail.Activation.Sequence != 1 || detail.Step.ActivationSequence != 2 || *detail.Activation.TaskID != ref.TaskID || !sameWorkflowJSON(detail.Input, before.Input) || !bytes.Equal(detail.Output, before.Output) {
|
||||
t.Fatal("history replaced by current activation", err)
|
||||
}
|
||||
_, err = api.ListActivations(ctx, f.token, v.ID, "task1", service.HistoryPageInput{Cursor: page.NextCursor})
|
||||
expectWorkerCode(t, err, "INVALID_ARGUMENT")
|
||||
_, err = api.GetActivation(ctx, f.token, v.ID, "task1", old.ActivationID)
|
||||
expectWorkerCode(t, err, "NOT_FOUND")
|
||||
_, err = api.GetActivation(ctx, f.token, v.ID, "task0", executionID(t))
|
||||
expectWorkerCode(t, err, "NOT_FOUND")
|
||||
_, err = api.ListPollRounds(ctx, f.token, v.ID, "task0", old.ActivationID, service.HistoryPageInput{})
|
||||
expectWorkerCode(t, err, "INVALID_ARGUMENT")
|
||||
for _, in := range []service.HistoryPageInput{{Limit: -1}, {Limit: 101}, {Cursor: strings.Repeat("x", 1025)}} {
|
||||
_, err = api.ListActivations(ctx, f.token, v.ID, "task0", in)
|
||||
expectWorkerCode(t, err, "INVALID_ARGUMENT")
|
||||
}
|
||||
_, err = api.ListActivations(ctx, f.token, v.ID, "$root", service.HistoryPageInput{})
|
||||
expectWorkerCode(t, err, "INVALID_ARGUMENT")
|
||||
// 无外键时模拟旧任务归属漂移：历史读取必须失败且不返回部分页。
|
||||
f.exec(`UPDATE pulse_tasks SET owner_id=? WHERE namespace_id=? AND id=?`, executionID(t), f.resources.NamespaceID, ref.TaskID)
|
||||
bad, err := api.ListActivations(ctx, f.token, v.ID, "task0", service.HistoryPageInput{})
|
||||
if fault.Code(err) != "UNAVAILABLE" || bad.Items != nil {
|
||||
t.Fatal("foreign historical task leaked", err)
|
||||
}
|
||||
if _, err = api.GetActivation(ctx, f.token, v.ID, "task0", old.ActivationID); fault.Code(err) != "UNAVAILABLE" {
|
||||
t.Fatal("foreign activation accepted", err)
|
||||
}
|
||||
f.exec(`UPDATE pulse_tasks SET owner_id=? WHERE namespace_id=? AND id=?`, old.ActivationID, f.resources.NamespaceID, ref.TaskID)
|
||||
f.exec(`UPDATE orbit_credentials SET permissions=JSON_SET(permissions,'$.queues',JSON_ARRAY(?)) WHERE namespace_id=? AND id=?`, string(executionID(t)), f.resources.NamespaceID, f.resources.CredentialID)
|
||||
_, err = api.ListActivations(ctx, f.token, v.ID, "task0", service.HistoryPageInput{})
|
||||
expectWorkerCode(t, err, "NOT_FOUND")
|
||||
})
|
||||
t.Run("poll_same_round_recovery_then_two_round_pages", func(t *testing.T) {
|
||||
f, v, oldTask := failedPoll(t, ctx, db, false)
|
||||
old := recoveryStep(t, f, v.ID, "query")
|
||||
before, err := api.GetActivation(ctx, f.token, v.ID, "query", old.ActivationID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
child := retryPollStep(t, f, v.ID)
|
||||
continueWorkflow(t, f, v.ID)
|
||||
ref := retryStart(t, f)
|
||||
finishPoll(t, f, ref, "PENDING")
|
||||
c := clockCandidate(t, f, v.ID, "query")
|
||||
waitClockDue(t, f, c)
|
||||
runClock(t, f, c)
|
||||
last := retryStart(t, f)
|
||||
finishPoll(t, f, last, "DONE")
|
||||
runtimeView(t, f, v.ID, "SUCCEEDED")
|
||||
prior, err := api.ListPollRounds(ctx, f.token, v.ID, "query", old.ActivationID, service.HistoryPageInput{})
|
||||
if err != nil || len(prior.Items) != 1 || prior.Items[0].Round != 1 || prior.Items[0].TaskID != oldTask.TaskID {
|
||||
t.Fatal("old round lost", err)
|
||||
}
|
||||
page, err := api.ListPollRounds(ctx, f.token, v.ID, "query", *child.Step.ActivationID, service.HistoryPageInput{Limit: 1})
|
||||
if err != nil || len(page.Items) != 1 || page.Items[0].Round != 2 || page.Items[0].TaskID != last.TaskID || page.NextCursor == "" {
|
||||
t.Fatal("latest round", err)
|
||||
}
|
||||
older, err := api.ListPollRounds(ctx, f.token, v.ID, "query", *child.Step.ActivationID, service.HistoryPageInput{Limit: 1, Cursor: page.NextCursor})
|
||||
if err != nil || len(older.Items) != 1 || older.Items[0].Round != 1 || older.Items[0].TaskID != child.Task.ID || older.NextCursor != "" {
|
||||
t.Fatal("recovered round", err)
|
||||
}
|
||||
_, err = api.ListPollRounds(ctx, f.token, v.ID, "query", old.ActivationID, service.HistoryPageInput{Cursor: page.NextCursor})
|
||||
expectWorkerCode(t, err, "INVALID_ARGUMENT")
|
||||
_, err = api.ListActivations(ctx, f.token, v.ID, "query", service.HistoryPageInput{Cursor: page.NextCursor})
|
||||
expectWorkerCode(t, err, "INVALID_ARGUMENT")
|
||||
after, err := api.GetActivation(ctx, f.token, v.ID, "query", old.ActivationID)
|
||||
if err != nil || after.Poll.UsedRounds != 1 || after.Poll.DeadlineAt != before.Poll.DeadlineAt || after.Poll.LastTaskID != oldTask.TaskID || after.Activation.Sequence != 1 || after.Step.ActivationSequence != 2 {
|
||||
t.Fatal("old poll replaced", err)
|
||||
}
|
||||
// 破坏非末轮的任务所有者，必须由轮次列表本身发现。
|
||||
f.exec(`UPDATE pulse_tasks SET owner_id=? WHERE namespace_id=? AND id=?`, executionID(t), f.resources.NamespaceID, child.Task.ID)
|
||||
bad, err := api.ListPollRounds(ctx, f.token, v.ID, "query", *child.Step.ActivationID, service.HistoryPageInput{})
|
||||
if fault.Code(err) != "UNAVAILABLE" || bad.Items != nil {
|
||||
t.Fatal("round ownership bypassed", err)
|
||||
}
|
||||
})
|
||||
for _, kind := range []string{"SIGNAL", "TIMER"} {
|
||||
t.Run("historical_wait_"+kind, func(t *testing.T) {
|
||||
f, v := waitRecoveryFixture(t, ctx, db, kind, 30000)
|
||||
key := "signal"
|
||||
if kind == "TIMER" {
|
||||
key = "timer"
|
||||
}
|
||||
old, err := api.Step(ctx, f.token, v.ID, key)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
rearmWorkflowWait(t, f, v.ID, key)
|
||||
detail, err := api.GetActivation(ctx, f.token, v.ID, key, *old.Step.ActivationID)
|
||||
if err != nil || detail.Wait.ID != old.Wait.ID || detail.Wait.DeadlineAt != old.Wait.DeadlineAt || detail.Step.ActivationSequence != 2 || detail.Activation.Sequence != 1 || detail.Activation.State != "CANCELLED" {
|
||||
t.Fatal("wait history lost", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
t.Run("signal_pages_legacy_summary_and_resource_audit", func(t *testing.T) {
|
||||
f, v, signals := signalStart(t, ctx, db, true)
|
||||
empty, err := signals.List(ctx, f.token, v.ID, service.HistoryPageInput{})
|
||||
if err != nil || empty.Items == nil || len(empty.Items) != 0 {
|
||||
t.Fatal(err)
|
||||
}
|
||||
sent := make([]core.ID, 0, 3)
|
||||
for i := 0; i < 3; i++ {
|
||||
out, err := signals.Send(ctx, f.token, v.ID, string(executionID(t)), signalInput(v))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
sent = append(sent, out.Signal.ID)
|
||||
}
|
||||
_, err = signals.ListOperations(ctx, f.token, v.ID, sent[0], service.HistoryPageInput{})
|
||||
expectWorkerCode(t, err, "PERMISSION_DENIED")
|
||||
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)
|
||||
audits, err := signals.ListOperations(ctx, f.token, v.ID, sent[0], service.HistoryPageInput{Limit: 1})
|
||||
if err != nil || len(audits.Items) != 1 || audits.Items[0].Action != "workflow.signal" {
|
||||
t.Fatal("signal audit list", err)
|
||||
}
|
||||
audit, err := signals.GetOperation(ctx, f.token, v.ID, sent[0], audits.Items[0].ID)
|
||||
if err != nil || audit.SignalID != sent[0] || audit.WorkflowID != v.ID || audit.Reason != signalInput(v).Reason || audit.EvidenceRef != signalInput(v).EvidenceRef {
|
||||
t.Fatal("signal audit detail", err)
|
||||
}
|
||||
_, err = signals.GetOperation(ctx, f.token, v.ID, sent[1], audit.ID)
|
||||
expectWorkerCode(t, err, "NOT_FOUND")
|
||||
// 构造相同受理时间、历史未核验身份和大正文，验证有界摘要及稳定二级排序；不将夹具作为合法信号送入运行器。
|
||||
f.exec(`UPDATE weave_signals SET accepted_at='2026-01-01 00:00:00.000000',state='REJECTED',disposition='LEGACY_UNVERIFIED',payload_version=NULL,target_activation_sequence=NULL,payload=? WHERE namespace_id=? AND workflow_id=?`, `"`+strings.Repeat("p", 200<<10)+`"`, f.resources.NamespaceID, v.ID)
|
||||
seen := map[core.ID]bool{}
|
||||
cursor := ""
|
||||
last := core.ID("")
|
||||
for {
|
||||
page, err := signals.List(ctx, f.token, v.ID, service.HistoryPageInput{Limit: 1, Cursor: cursor})
|
||||
if err != nil || len(page.Items) != 1 {
|
||||
t.Fatal("signal page", err)
|
||||
}
|
||||
raw, _ := json.Marshal(page)
|
||||
item := page.Items[0]
|
||||
if len(raw) > 2048 || seen[item.ID] || last != "" && item.ID >= last || item.TargetActivationSequence != 0 || item.PayloadVersion != "" {
|
||||
t.Fatal("signal summary lost bounds or identity")
|
||||
}
|
||||
seen[item.ID] = true
|
||||
last = item.ID
|
||||
if page.NextCursor == "" {
|
||||
break
|
||||
}
|
||||
cursor = page.NextCursor
|
||||
}
|
||||
if len(seen) != 3 {
|
||||
t.Fatal("signal omitted")
|
||||
}
|
||||
_, err = signals.ListOperations(ctx, f.token, v.ID, sent[0], service.HistoryPageInput{Cursor: cursor})
|
||||
expectWorkerCode(t, err, "INVALID_ARGUMENT")
|
||||
other := runtimeStart(t, f, signalSpec(f.resources.QueueID, true))
|
||||
grantWorkflowSignal(f)
|
||||
_, err = signals.List(ctx, f.token, other.ID, service.HistoryPageInput{Cursor: cursor})
|
||||
expectWorkerCode(t, err, "INVALID_ARGUMENT")
|
||||
_, err = signals.List(ctx, f.token, v.ID, service.HistoryPageInput{Limit: 101})
|
||||
expectWorkerCode(t, err, "INVALID_ARGUMENT")
|
||||
_, err = signals.ListOperations(ctx, f.token, v.ID, executionID(t), service.HistoryPageInput{})
|
||||
expectWorkerCode(t, err, "PERMISSION_DENIED") // runtimeStart重签普通权限，审计权限不能随读取继承。
|
||||
|
||||
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)
|
||||
_, err = signals.ListOperations(ctx, f.token, other.ID, sent[0], service.HistoryPageInput{})
|
||||
expectWorkerCode(t, err, "NOT_FOUND")
|
||||
_, err = signals.GetOperation(ctx, f.token, other.ID, sent[0], audit.ID)
|
||||
expectWorkerCode(t, err, "NOT_FOUND")
|
||||
|
||||
})
|
||||
}
|
||||
|
||||
func TestWorkflowHistoryCursorScopeAndWidth(t *testing.T) {
|
||||
ns, id, step := executionID(t), executionID(t), executionID(t)
|
||||
for _, kind := range []string{"workflow-activations", "workflow-rounds", "attempts"} {
|
||||
cursor := historyCursor{Version: 1, Kind: kind, Namespace: ns, Task: id, Attempt: step, Epoch: 1<<32 - 1}
|
||||
raw, _ := encodeHistoryCursor(cursor)
|
||||
if _, err := decodeHistoryCursorKind(raw, ns, id, step, kind); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := decodeHistoryCursorKind(raw, ns, id, executionID(t), kind); err == nil {
|
||||
t.Fatal("cursor crossed child")
|
||||
}
|
||||
cursor.Epoch = 1 << 32
|
||||
raw, _ = encodeHistoryCursor(cursor)
|
||||
_, err := decodeHistoryCursorKind(raw, ns, id, step, kind)
|
||||
if (err == nil) != (kind == "attempts") {
|
||||
t.Fatal("sequence width or legacy epoch changed", kind, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,121 @@
|
||||
package mysql
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"os"
|
||||
"reflect"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
driver "github.com/go-sql-driver/mysql"
|
||||
)
|
||||
|
||||
// 用真实v22结构验证历史索引升级，逐字段核对信号未被改写。
|
||||
func TestIntegrationWorkflowHistoryUpgrade(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_history_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 > 22 {
|
||||
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, id := executionID(t), executionID(t), executionID(t)
|
||||
exec(`INSERT INTO weave_signals(namespace_id,id,workflow_id,key_digest,request_digest,type,target_step_key,correlation_key,payload,actor_id,state,accepted_at,expires_at,available_at,disposition) VALUES (?,?,?,UNHEX(SHA2('k',256)),UNHEX(SHA2('r',256)),'external.done','signal','关联尾部 ','{"n":9007199254740993}','legacy','REJECTED',UTC_TIMESTAMP(6),DATE_ADD(UTC_TIMESTAMP(6),INTERVAL 1 HOUR),UTC_TIMESTAMP(6),'LEGACY_UNVERIFIED')`, ns, id, workflow)
|
||||
snapshot := func() []sql.NullString {
|
||||
t.Helper()
|
||||
rows, err := db.QueryContext(ctx, `SELECT * FROM weave_signals WHERE namespace_id=? AND id=?`, ns, id)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer rows.Close()
|
||||
cols, err := rows.Columns()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !rows.Next() {
|
||||
t.Fatal("signal missing")
|
||||
}
|
||||
out := make([]sql.NullString, len(cols))
|
||||
args := make([]any, len(cols))
|
||||
for i := range out {
|
||||
args[i] = &out[i]
|
||||
}
|
||||
if err = rows.Scan(args...); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return out
|
||||
}
|
||||
before := snapshot()
|
||||
if err = Migrate(ctx, db); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !reflect.DeepEqual(before, snapshot()) {
|
||||
t.Fatal("history index changed signal fields")
|
||||
}
|
||||
if err = CheckSchema(ctx, db); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var index string
|
||||
if err = db.QueryRowContext(ctx, `SELECT GROUP_CONCAT(COLUMN_NAME ORDER BY SEQ_IN_INDEX) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='weave_signals' AND INDEX_NAME='ix_signal_history' AND NON_UNIQUE=1`).Scan(&index); err != nil || index != "namespace_id,workflow_id,accepted_at,id" {
|
||||
t.Fatal("history index", index, err)
|
||||
}
|
||||
var fks int
|
||||
if err = db.QueryRowContext(ctx, `SELECT COUNT(*) FROM information_schema.REFERENTIAL_CONSTRAINTS WHERE CONSTRAINT_SCHEMA=DATABASE()`).Scan(&fks); err != nil || fks != 0 {
|
||||
t.Fatal("foreign keys", fks, err)
|
||||
}
|
||||
if err = Migrate(ctx, db); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Log("v22 upgraded; signal values, precision, legacy identity and timestamps unchanged; history index installed; foreign keys=0")
|
||||
}
|
||||
141
internal/storage/mysql/workflow_signal_history.go
Normal file
141
internal/storage/mysql/workflow_signal_history.go
Normal file
@@ -0,0 +1,141 @@
|
||||
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/signal"
|
||||
)
|
||||
|
||||
func (s *WorkflowSignalStore) List(ctx context.Context, t auth.Token, workflow core.ID, in service.HistoryPageInput) (signal.Page, error) {
|
||||
out := signal.Page{WorkflowID: workflow, Items: []signal.Summary{}}
|
||||
err := s.transaction(ctx, t, signal.ReadAction, func(tx *sql.Tx, p auth.Permissions) error {
|
||||
v, _, _, err := readWorkflowForQuery(ctx, tx, t.NamespaceID, workflow, p)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
out.Revision = v.Revision
|
||||
q := `SELECT id,workflow_id,revision,type,COALESCE(payload_version,''),target_step_key,COALESCE(target_activation_sequence,0),state,COALESCE(disposition,''),accepted_at,expires_at FROM weave_signals WHERE namespace_id=? AND workflow_id=?`
|
||||
args := []any{t.NamespaceID, workflow}
|
||||
if in.Cursor != "" {
|
||||
c, err := decodeHistoryCursorKind(in.Cursor, t.NamespaceID, workflow, "", "workflow-signals")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
q += ` AND (accepted_at<? OR (accepted_at=? AND id<?))`
|
||||
args = append(args, *c.Created, *c.Created, c.ID)
|
||||
}
|
||||
q += ` ORDER BY accepted_at DESC,id DESC LIMIT ?`
|
||||
args = append(args, in.Limit+1)
|
||||
rows, err := tx.QueryContext(ctx, q, args...)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var v signal.Summary
|
||||
if err = rows.Scan(&v.ID, &v.WorkflowID, &v.Revision, &v.Name, &v.PayloadVersion, &v.TargetStepKey, &v.TargetActivationSequence, &v.State, &v.Disposition, &v.AcceptedAt, &v.ExpiresAt); 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-signals", Namespace: t.NamespaceID, Task: workflow, Created: &last.AcceptedAt, ID: last.ID})
|
||||
}
|
||||
return err
|
||||
})
|
||||
if err != nil {
|
||||
return signal.Page{}, err
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// authorizeSignalAudit 先授权固定流程全部队列，再验证信号属于该实例，不通过审计ID跨资源查找。
|
||||
func authorizeSignalAudit(ctx context.Context, tx *sql.Tx, ns, workflow, id core.ID, p auth.Permissions) error {
|
||||
if _, _, _, err := readWorkflowForQuery(ctx, tx, ns, workflow, p); err != nil {
|
||||
return err
|
||||
}
|
||||
var found core.ID
|
||||
err := tx.QueryRowContext(ctx, `SELECT id FROM weave_signals WHERE namespace_id=? AND workflow_id=? AND id=?`, ns, workflow, id).Scan(&found)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return fault.NotFound
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
const signalOperationColumns = `id,resource_id,operation_id,actor_id,action,created_at`
|
||||
|
||||
func (s *WorkflowSignalStore) ListOperations(ctx context.Context, t auth.Token, workflow, id core.ID, in service.HistoryPageInput) (signal.OperationPage, error) {
|
||||
out := signal.OperationPage{Items: []signal.OperationSummary{}}
|
||||
err := s.transaction(ctx, t, signal.AuditAction, func(tx *sql.Tx, p auth.Permissions) error {
|
||||
if err := authorizeSignalAudit(ctx, tx, t.NamespaceID, workflow, id, p); err != nil {
|
||||
return err
|
||||
}
|
||||
q := `SELECT ` + signalOperationColumns + ` FROM orbit_audit_events WHERE namespace_id=? AND resource_kind='workflow_signal' AND resource_id=?`
|
||||
args := []any{t.NamespaceID, id}
|
||||
if in.Cursor != "" {
|
||||
c, err := decodeHistoryCursorKind(in.Cursor, t.NamespaceID, workflow, id, "signal-operations")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
q += ` AND (created_at<? OR (created_at=? AND id<?))`
|
||||
args = append(args, *c.Created, *c.Created, c.ID)
|
||||
}
|
||||
q += ` ORDER BY created_at DESC,id DESC LIMIT ?`
|
||||
args = append(args, in.Limit+1)
|
||||
rows, err := tx.QueryContext(ctx, q, args...)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
v := signal.OperationSummary{WorkflowID: workflow}
|
||||
if err = rows.Scan(&v.ID, &v.SignalID, &v.OperationID, &v.ActorID, &v.Action, &v.CreatedAt); 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: "signal-operations", Namespace: t.NamespaceID, Task: workflow, Attempt: id, Created: &last.CreatedAt, ID: last.ID})
|
||||
}
|
||||
return err
|
||||
})
|
||||
if err != nil {
|
||||
return signal.OperationPage{}, err
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
func (s *WorkflowSignalStore) GetOperation(ctx context.Context, t auth.Token, workflow, id, operation core.ID) (signal.OperationView, error) {
|
||||
out := signal.OperationView{OperationSummary: signal.OperationSummary{WorkflowID: workflow}}
|
||||
err := s.transaction(ctx, t, signal.AuditAction, func(tx *sql.Tx, p auth.Permissions) error {
|
||||
if err := authorizeSignalAudit(ctx, tx, t.NamespaceID, workflow, id, p); err != nil {
|
||||
return err
|
||||
}
|
||||
var before, after []byte
|
||||
err := tx.QueryRowContext(ctx, `SELECT `+signalOperationColumns+",reason,COALESCE(evidence_ref,''),`before`,`after` FROM orbit_audit_events WHERE namespace_id=? AND resource_kind='workflow_signal' AND resource_id=? AND id=?", t.NamespaceID, id, operation).Scan(&out.ID, &out.SignalID, &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
|
||||
})
|
||||
if err != nil {
|
||||
return signal.OperationView{}, err
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
58
internal/transport/httpserver/workflow_history.go
Normal file
58
internal/transport/httpserver/workflow_history.go
Normal file
@@ -0,0 +1,58 @@
|
||||
package httpserver
|
||||
|
||||
import (
|
||||
core "git.yuntuo.hk/k2safe/orbit/model"
|
||||
"git.yuntuo.hk/k2safe/orbit/platform/auth"
|
||||
"git.yuntuo.hk/k2safe/orbit/weave/instance"
|
||||
"git.yuntuo.hk/k2safe/orbit/weave/signal"
|
||||
"net/http"
|
||||
)
|
||||
|
||||
func registerWorkflowHistory(mux *http.ServeMux, instances *instance.Service, signals *signal.Service) {
|
||||
mux.HandleFunc("GET /v1/workflows/{id}/steps/{key}/activations", 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.ListActivations(r.Context(), t, core.ID(r.PathValue("id")), r.PathValue("key"), in)
|
||||
return out, 200, err
|
||||
}))
|
||||
mux.HandleFunc("GET /v1/workflows/{id}/steps/{key}/activations/{activation}", 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.GetActivation(r.Context(), t, core.ID(r.PathValue("id")), r.PathValue("key"), core.ID(r.PathValue("activation")))
|
||||
return out, 200, err
|
||||
}))
|
||||
mux.HandleFunc("GET /v1/workflows/{id}/steps/{key}/activations/{activation}/rounds", 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.ListPollRounds(r.Context(), t, core.ID(r.PathValue("id")), r.PathValue("key"), core.ID(r.PathValue("activation")), in)
|
||||
return out, 200, err
|
||||
}))
|
||||
mux.HandleFunc("GET /v1/workflows/{id}/signals", 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 := signals.List(r.Context(), t, core.ID(r.PathValue("id")), in)
|
||||
return out, 200, err
|
||||
}))
|
||||
mux.HandleFunc("GET /v1/workflows/{id}/signals/{signal}/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 := signals.ListOperations(r.Context(), t, core.ID(r.PathValue("id")), core.ID(r.PathValue("signal")), in)
|
||||
return out, 200, err
|
||||
}))
|
||||
mux.HandleFunc("GET /v1/workflows/{id}/signals/{signal}/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 := signals.GetOperation(r.Context(), t, core.ID(r.PathValue("id")), core.ID(r.PathValue("signal")), core.ID(r.PathValue("operation")))
|
||||
return out, 200, err
|
||||
}))
|
||||
}
|
||||
@@ -15,6 +15,7 @@ import (
|
||||
// RegisterWorkflows 只适配协议，实际授权、并发版本、原键回执及审计由事务服务校验。
|
||||
func RegisterWorkflows(server *http.Server, instances *instance.Service, controls *control.Service, signals *signal.Service, reviews *recovery.Service) {
|
||||
mux := server.Handler.(*http.ServeMux)
|
||||
registerWorkflowHistory(mux, instances, signals)
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user