98 lines
3.6 KiB
Go
98 lines
3.6 KiB
Go
package mysql
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"encoding/json"
|
||
"errors"
|
||
"strings"
|
||
"time"
|
||
|
||
core "git.yuntuo.hk/k2safe/orbit/model"
|
||
"git.yuntuo.hk/k2safe/orbit/weave/signal"
|
||
)
|
||
|
||
type signalTransition struct {
|
||
WorkflowID core.ID `json:"workflow_id"` // WorkflowID 与信号归属同事务保存的流程身份。
|
||
SignalID core.ID `json:"signal_id"` // SignalID 被修改的信号身份，旧记录保留。
|
||
From string `json:"from"` // From 本次转换前的真实状态，不以当前状态推测过去。
|
||
State string `json:"state"` // State 本次转换提交的目标状态。
|
||
Disposition string `json:"disposition"` // Disposition 消费、过期或拒绝的稳定原因。
|
||
ConsumedWaitID *core.ID `json:"consumed_wait_id,omitempty"` // ConsumedWaitID 消费赢家等待，其他转换省略。
|
||
}
|
||
|
||
func emitWorkflowSignalTransition(ctx context.Context, tx *sql.Tx, ns, workflow, id core.ID, revision uint64, state, reason string, wait *core.ID) error {
|
||
return emitEvent(ctx, tx, ns, id, revision, "workflow_signal", "workflow.signal.changed", signalTransition{workflow, id, "PENDING", state, reason, wait})
|
||
}
|
||
|
||
// rejectPendingWorkflowSignals 的filter只由内部固定SQL构造。调用方已持有实例写锁，
|
||
// 一次读取最多1025条检测破坏上限的存储；更新和每条事件同事务，按32条批量追加事件。
|
||
func rejectPendingWorkflowSignals(ctx context.Context, tx *sql.Tx, g *workflowGraph, filter string, args []any, reason string, now time.Time) error {
|
||
where := ` WHERE s.namespace_id=? AND s.workflow_id=? AND s.state='PENDING'` + filter
|
||
params := append([]any{g.view.NamespaceID, g.view.ID}, args...)
|
||
rows, err := tx.QueryContext(ctx, `SELECT s.id,s.revision FROM weave_signals s`+where+` ORDER BY s.id LIMIT ? FOR UPDATE`, append(params, signal.MaxPerWorkflow+1)...)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
type entry struct {
|
||
id core.ID
|
||
revision uint64
|
||
}
|
||
var items []entry
|
||
for rows.Next() {
|
||
var v entry
|
||
if err = rows.Scan(&v.id, &v.revision); err != nil {
|
||
rows.Close()
|
||
return err
|
||
}
|
||
items = append(items, v)
|
||
}
|
||
err = rows.Err()
|
||
rows.Close()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if len(items) > signal.MaxPerWorkflow {
|
||
return errors.New("workflow signal limit violated")
|
||
}
|
||
if len(items) == 0 {
|
||
return nil
|
||
}
|
||
update := append([]any{reason, now}, params...)
|
||
result, err := tx.ExecContext(ctx, `UPDATE weave_signals s SET s.state='REJECTED',s.disposition=?,s.revision=s.revision+1,s.updated_at=?`+where, update...)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
n, err := result.RowsAffected()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if n != int64(len(items)) {
|
||
return errors.New("workflow signal rejection set changed")
|
||
}
|
||
for start := 0; start < len(items); start += 32 {
|
||
end := start + 32
|
||
if end > len(items) {
|
||
end = len(items)
|
||
}
|
||
var values []string
|
||
var insert []any
|
||
for _, v := range items[start:end] {
|
||
event, err := core.NewID()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
raw, err := json.Marshal(signalTransition{WorkflowID: g.view.ID, SignalID: v.id, From: "PENDING", State: "REJECTED", Disposition: reason})
|
||
if err != nil {
|
||
return err
|
||
}
|
||
values = append(values, "(?,?,?,?,?,?,?,UTC_TIMESTAMP(6))")
|
||
insert = append(insert, g.view.NamespaceID, event, "workflow_signal", v.id, v.revision+1, "workflow.signal.changed", string(raw))
|
||
}
|
||
if _, err = tx.ExecContext(ctx, `INSERT INTO orbit_events(namespace_id,id,aggregate_kind,aggregate_id,aggregate_revision,type,payload,occurred_at) VALUES `+strings.Join(values, ","), insert...); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|