253 lines
9.9 KiB
Go
253 lines
9.9 KiB
Go
// workflow 通过公共SDK运行并行处理、计时等待、外部回执与后续发布；业务处理器在Worker中执行。
|
||
package main
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"os"
|
||
"strings"
|
||
"time"
|
||
|
||
orbit "git.yuntuo.hk/k2safe/orbit/sdk/go"
|
||
)
|
||
|
||
func main() {
|
||
if err := run(); err != nil {
|
||
fmt.Fprintln(os.Stderr, err)
|
||
os.Exit(1)
|
||
}
|
||
}
|
||
func run() error {
|
||
raw, err := os.ReadFile(os.Getenv("ORBIT_CREDENTIALS_FILE"))
|
||
if err != nil {
|
||
return err
|
||
}
|
||
var credentials struct {
|
||
Token string `json:"token"`
|
||
QueueID string `json:"queue_id"`
|
||
}
|
||
if json.Unmarshal(raw, &credentials) != nil || credentials.Token == "" || credentials.QueueID == "" {
|
||
return errors.New("invalid private credentials file")
|
||
}
|
||
client, err := orbit.NewClient(os.Getenv("ORBIT_URL"), orbit.Options{Token: credentials.Token})
|
||
if err != nil {
|
||
return err
|
||
}
|
||
operation, err := orbit.NewWorkerMessageID()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second)
|
||
defer cancel()
|
||
handler := "artifact.normalize." + operation
|
||
typ := orbit.WorkflowType{Kind: "OBJECT", Fields: map[string]orbit.WorkflowType{"text": {Kind: "STRING"}}}
|
||
task := orbit.WorkflowTask{QueueID: credentials.QueueID, Handler: handler, HandlerVersion: "v1", PayloadVersion: "v1", InputType: typ, Input: orbit.WorkflowRef("input"), ResultType: typ, TimeoutMS: 20000, Retry: orbit.RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: 5000}, RecoverySafety: orbit.RecoveryIdempotent}
|
||
pair := orbit.WorkflowParallel{Branches: []orbit.WorkflowParallelBranch{
|
||
{Name: "left", Body: orbit.WorkflowBlock{Steps: []orbit.WorkflowStep{{Key: "left", Kind: "TASK", Task: &task}}, OutputType: typ, Output: orbit.WorkflowRef("left")}},
|
||
{Name: "right", Body: orbit.WorkflowBlock{Steps: []orbit.WorkflowStep{{Key: "right", Kind: "TASK", Task: &task}}, OutputType: typ, Output: orbit.WorkflowRef("right")}},
|
||
}}
|
||
publish := task
|
||
publish.Input = orbit.WorkflowRef("pair", "left")
|
||
spec := orbit.WorkflowSpecification{SchemaVersion: 1, TimeoutMS: 60000, FailurePolicy: "REVIEW", InputType: typ, Body: orbit.WorkflowBlock{Steps: []orbit.WorkflowStep{
|
||
{Key: "pair", Kind: "PARALLEL", Parallel: &pair},
|
||
{Key: "cooldown", Kind: "TIMER", Timer: &orbit.WorkflowTimer{DelayMS: 500}},
|
||
{Key: "receipt", Kind: "SIGNAL", Signal: &orbit.WorkflowSignal{Name: "artifact.confirmed", PayloadVersion: "v1", Correlation: orbit.WorkflowValue{Kind: "LITERAL", Literal: json.RawMessage(`"` + operation + `"`)}, PayloadType: orbit.WorkflowType{Kind: "BOOLEAN"}, TimeoutMS: 30000}},
|
||
{Key: "publish", Kind: "TASK", Task: &publish},
|
||
}, OutputType: typ, Output: orbit.WorkflowRef("publish")}}
|
||
definition, err := client.CreateWorkflowDefinition(ctx, operation, orbit.CreateWorkflowDefinitionRequest{Name: "artifact." + operation, Description: "并行结果生成与模拟外部回执"})
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if _, err = client.PublishWorkflowVersion(ctx, definition.Definition.ID, operation, orbit.PublishWorkflowVersionRequest{Version: "v1", Specification: spec, Reason: "publish runnable workflow example"}); err != nil {
|
||
return err
|
||
}
|
||
start := orbit.StartWorkflowRequest{DefinitionID: definition.Definition.ID, Version: "v1", Input: json.RawMessage(`{"text":" hello orbit "}`), BusinessKey: operation}
|
||
accepted, err := client.StartWorkflow(ctx, operation, start)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
repeated, err := client.StartWorkflow(ctx, operation, start)
|
||
if err != nil || !repeated.Deduplicated || repeated.Instance.ID != accepted.Instance.ID {
|
||
return errors.New("start identity was not preserved")
|
||
}
|
||
worker, err := orbit.NewWorker(client, orbit.WorkerOptions{QueueID: credentials.QueueID, Capacity: 2, ShutdownTimeout: 5 * time.Second, Handlers: []orbit.WorkerHandler{{Capability: orbit.WorkerCapability{Handler: handler, HandlerVersion: "v1", PayloadVersion: "v1"}, Handle: normalize}}})
|
||
if err != nil {
|
||
return err
|
||
}
|
||
workerCtx, stop := context.WithCancel(ctx)
|
||
done := make(chan error, 1)
|
||
go func() { done <- worker.Run(workerCtx) }()
|
||
waited := false
|
||
defer func() {
|
||
stop()
|
||
if !waited {
|
||
<-done
|
||
}
|
||
}()
|
||
var wait orbit.WorkflowStepDetail
|
||
for {
|
||
wait, err = client.GetWorkflowStep(ctx, accepted.Instance.ID, "receipt")
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if wait.Step.State == "WAITING" {
|
||
break
|
||
}
|
||
current, e := client.GetWorkflow(ctx, accepted.Instance.ID)
|
||
if e != nil {
|
||
return e
|
||
}
|
||
if current.State == "NEEDS_REVIEW" || current.State == "FAILED" {
|
||
return fmt.Errorf("workflow requires attention: %s", current.State)
|
||
}
|
||
select {
|
||
case e := <-done:
|
||
waited = true
|
||
return fmt.Errorf("Worker stopped: %v", e)
|
||
case <-ctx.Done():
|
||
return ctx.Err()
|
||
case <-time.After(100 * time.Millisecond):
|
||
}
|
||
}
|
||
if wait.Wait == nil || wait.Step.ActivationSequence == 0 {
|
||
return errors.New("wait contract missing")
|
||
}
|
||
// 示例模拟外部系统的已验证回执。实际系统由独立、具备信号权限的调用者确认业务事实。
|
||
signal := orbit.SendWorkflowSignalRequest{Name: *wait.Wait.Name, PayloadVersion: *wait.Wait.PayloadVersion, TargetStepKey: wait.Step.Key, TargetActivationSequence: wait.Step.ActivationSequence, CorrelationKey: wait.Wait.CorrelationKey, Payload: json.RawMessage(`true`), Reason: "simulated verified external receipt"}
|
||
sent, err := client.SendWorkflowSignal(ctx, accepted.Instance.ID, operation, signal)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
again, err := client.SendWorkflowSignal(ctx, accepted.Instance.ID, operation, signal)
|
||
if err != nil || !again.Deduplicated || again.Signal.ID != sent.Signal.ID {
|
||
return errors.New("signal identity was not preserved")
|
||
}
|
||
for {
|
||
current, e := client.GetWorkflow(ctx, accepted.Instance.ID)
|
||
if e != nil {
|
||
return e
|
||
}
|
||
if current.State == "SUCCEEDED" {
|
||
fmt.Printf("workflow=%s state=%s result=%s start_deduplicated=%t signal_deduplicated=%t\n", current.ID, current.State, current.Output, repeated.Deduplicated, again.Deduplicated)
|
||
break
|
||
}
|
||
if current.State == "NEEDS_REVIEW" || current.State == "FAILED" || current.State == "CANCELLED" {
|
||
return fmt.Errorf("workflow requires attention: %s", current.State)
|
||
}
|
||
select {
|
||
case e := <-done:
|
||
waited = true
|
||
return fmt.Errorf("Worker stopped: %v", e)
|
||
case <-ctx.Done():
|
||
return ctx.Err()
|
||
case <-time.After(100 * time.Millisecond):
|
||
}
|
||
}
|
||
steps, err := client.GetWorkflowSteps(ctx, accepted.Instance.ID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
verified := 0
|
||
for _, step := range steps.Steps {
|
||
if step.TaskID == nil {
|
||
continue
|
||
}
|
||
source, err := client.GetTaskSource(ctx, *step.TaskID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if source.Workflow == nil || source.Workflow.WorkflowID != accepted.Instance.ID || source.Workflow.DefinitionVersionID != accepted.Instance.DefinitionVersionID || source.Workflow.StepID != step.ID || source.Workflow.ActivationID != *step.ActivationID || source.Workflow.ActivationSequence != 1 {
|
||
return errors.New("workflow task source mismatch")
|
||
}
|
||
verified++
|
||
}
|
||
if verified != 3 {
|
||
return errors.New("incomplete workflow task sources")
|
||
}
|
||
fmt.Printf("workflow_sources_verified=%d fixed_version=%s\n", verified, accepted.Instance.Version)
|
||
history, err := client.ListWorkflowActivations(ctx, accepted.Instance.ID, "receipt", orbit.HistoryPageRequest{Limit: 20})
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if len(history.Items) != 1 {
|
||
return errors.New("unexpected receipt activation history")
|
||
}
|
||
historical, err := client.GetWorkflowActivation(ctx, accepted.Instance.ID, "receipt", history.Items[0].ID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if historical.Wait == nil || historical.Wait.WinnerSignalID == nil || *historical.Wait.WinnerSignalID != sent.Signal.ID {
|
||
return errors.New("historical signal winner missing")
|
||
}
|
||
signals, err := client.ListWorkflowSignals(ctx, accepted.Instance.ID, orbit.HistoryPageRequest{Limit: 20})
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if len(signals.Items) != 1 || signals.Items[0].State != "CONSUMED" || signals.Items[0].ID != sent.Signal.ID {
|
||
return errors.New("signal history missing")
|
||
}
|
||
fmt.Println("workflow_history_verified=true signal_state=CONSUMED")
|
||
// 时间线用于审计展示，游标只是倒序翻页位置，不作为可靠消息消费水位。
|
||
cursor, eventCount, consumed := "", 0, false
|
||
for {
|
||
page, err := client.ListWorkflowEvents(ctx, accepted.Instance.ID, orbit.HistoryPageRequest{Limit: 5, Cursor: cursor})
|
||
if err != nil {
|
||
return err
|
||
}
|
||
for _, event := range page.Items {
|
||
eventCount++
|
||
if event.AggregateKind == "workflow_signal" && event.AggregateID == sent.Signal.ID && event.State == "CONSUMED" {
|
||
detail, err := client.GetWorkflowEvent(ctx, accepted.Instance.ID, event.ID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
var transition struct {
|
||
From string `json:"from"`
|
||
ConsumedWaitID string `json:"consumed_wait_id"`
|
||
}
|
||
if json.Unmarshal(detail.Payload, &transition) != nil || transition.From != "PENDING" || transition.ConsumedWaitID != historical.Wait.ID {
|
||
return errors.New("signal transition evidence mismatch")
|
||
}
|
||
consumed = true
|
||
}
|
||
}
|
||
if page.NextCursor == "" {
|
||
break
|
||
}
|
||
cursor = page.NextCursor
|
||
}
|
||
if !consumed {
|
||
return errors.New("consumed signal event missing")
|
||
}
|
||
fmt.Printf("workflow_timeline_verified=true events=%d signal_transition=PENDING->CONSUMED\n", eventCount)
|
||
|
||
stop()
|
||
err = <-done
|
||
waited = true
|
||
if err != nil {
|
||
return err
|
||
}
|
||
stats := worker.Stats()
|
||
if stats.Started != 3 || stats.ReportsConfirmed != 3 || stats.Running != 0 {
|
||
return errors.New("unexpected task count or incomplete drain")
|
||
}
|
||
fmt.Printf("worker_drained=true started=%d confirmed=%d\n", stats.Started, stats.ReportsConfirmed)
|
||
return nil
|
||
}
|
||
func normalize(ctx context.Context, execution orbit.WorkerExecution) (json.RawMessage, error) {
|
||
var in struct {
|
||
Text string `json:"text"`
|
||
}
|
||
if json.Unmarshal(execution.Task.Payload, &in) != nil {
|
||
return nil, &orbit.WorkerFailure{Code: "INVALID_TEXT"}
|
||
}
|
||
if err := ctx.Err(); err != nil {
|
||
return nil, err
|
||
}
|
||
return json.Marshal(struct {
|
||
Text string `json:"text"`
|
||
}{strings.ToUpper(strings.TrimSpace(in.Text))})
|
||
}
|