feat: add bounded polling and evidence-based workflow recovery examples
This commit is contained in:
119
sdk/go/examples/internal/flowdemo/runtime.go
Normal file
119
sdk/go/examples/internal/flowdemo/runtime.go
Normal file
@@ -0,0 +1,119 @@
|
||||
// Package flowdemo 仅为可运行流程示例提供初始化和有界等待，不属于公开SDK或业务恢复策略。
|
||||
package flowdemo
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
orbit "git.yuntuo.hk/k2safe/orbit/sdk/go"
|
||||
)
|
||||
|
||||
func Client() (*orbit.Client, string, string, error) {
|
||||
raw, err := os.ReadFile(os.Getenv("ORBIT_CREDENTIALS_FILE"))
|
||||
if err != nil {
|
||||
return nil, "", "", err
|
||||
}
|
||||
var credentials struct {
|
||||
Token string `json:"token"`
|
||||
QueueID string `json:"queue_id"`
|
||||
}
|
||||
if json.Unmarshal(raw, &credentials) != nil || credentials.Token == "" || credentials.QueueID == "" {
|
||||
return nil, "", "", errors.New("invalid private credentials file")
|
||||
}
|
||||
client, err := orbit.NewClient(os.Getenv("ORBIT_URL"), orbit.Options{Token: credentials.Token})
|
||||
if err != nil {
|
||||
return nil, "", "", err
|
||||
}
|
||||
operation, err := orbit.NewWorkerMessageID()
|
||||
return client, credentials.QueueID, operation, err
|
||||
}
|
||||
|
||||
// PublishStart 每次示例运行创建独立身份；不将重新运行命令当作未知响应的原键恢复。
|
||||
func PublishStart(ctx context.Context, api *orbit.Client, operation, name string, spec orbit.WorkflowSpecification, input json.RawMessage) (orbit.WorkflowInstance, error) {
|
||||
var empty orbit.WorkflowInstance
|
||||
fmt.Printf("scenario=%s operation=%s\n", name, operation)
|
||||
created, err := api.CreateWorkflowDefinition(ctx, operation, orbit.CreateWorkflowDefinitionRequest{Name: name + "." + operation, Description: "controlled standalone workflow example"})
|
||||
if err != nil {
|
||||
return empty, err
|
||||
}
|
||||
if _, err = api.PublishWorkflowVersion(ctx, created.Definition.ID, operation, orbit.PublishWorkflowVersionRequest{Version: "v1", Specification: spec, Reason: "publish controlled SDK example"}); err != nil {
|
||||
return empty, err
|
||||
}
|
||||
result, err := api.StartWorkflow(ctx, operation, orbit.StartWorkflowRequest{DefinitionID: created.Definition.ID, Version: "v1", BusinessKey: operation, Input: input})
|
||||
if err != nil {
|
||||
return empty, err
|
||||
}
|
||||
fmt.Printf("workflow=%s version=%s\n", result.Instance.ID, result.Instance.Version)
|
||||
return result.Instance, nil
|
||||
}
|
||||
|
||||
// Runner 持有一个真实Worker及其结束通知，等待时同时检测Worker异常退出。
|
||||
type Runner struct {
|
||||
Worker *orbit.Worker
|
||||
stop context.CancelFunc
|
||||
done chan struct{}
|
||||
err error // err 仅在done关闭后读取，关闭通道建立同步关系。
|
||||
}
|
||||
|
||||
func StartWorker(ctx context.Context, api *orbit.Client, queue string, handlers ...orbit.WorkerHandler) (*Runner, error) {
|
||||
w, err := orbit.NewWorker(api, orbit.WorkerOptions{QueueID: queue, Capacity: 1, ShutdownTimeout: 5 * time.Second, Handlers: handlers})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
workerCtx, stop := context.WithCancel(ctx)
|
||||
r := &Runner{Worker: w, stop: stop, done: make(chan struct{})}
|
||||
go func() { r.err = w.Run(workerCtx); close(r.done) }()
|
||||
return r, nil
|
||||
}
|
||||
|
||||
func (r *Runner) Close() error {
|
||||
r.stop()
|
||||
<-r.done
|
||||
return r.err
|
||||
}
|
||||
|
||||
func (r *Runner) Wait(ctx context.Context, check func() (bool, error)) error {
|
||||
for {
|
||||
ready, err := check()
|
||||
if err != nil || ready {
|
||||
return err
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-r.done:
|
||||
return fmt.Errorf("example Worker stopped before expected state: %v", r.err)
|
||||
case <-time.After(100 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (r *Runner) Step(ctx context.Context, api *orbit.Client, workflow, key string, predicate func(orbit.WorkflowStepDetail) bool) (orbit.WorkflowStepDetail, error) {
|
||||
var out orbit.WorkflowStepDetail
|
||||
err := r.Wait(ctx, func() (bool, error) {
|
||||
var err error
|
||||
out, err = api.GetWorkflowStep(ctx, workflow, key)
|
||||
return err == nil && predicate(out), err
|
||||
})
|
||||
return out, err
|
||||
}
|
||||
|
||||
func (r *Runner) Workflow(ctx context.Context, api *orbit.Client, workflow, state string) (orbit.WorkflowInstance, error) {
|
||||
var out orbit.WorkflowInstance
|
||||
err := r.Wait(ctx, func() (bool, error) {
|
||||
var err error
|
||||
out, err = api.GetWorkflow(ctx, workflow)
|
||||
if err != nil || out.State == state {
|
||||
return err == nil, err
|
||||
}
|
||||
if out.State == "NEEDS_REVIEW" || out.State == "FAILED" || out.State == "CANCELLED" || out.State == "SUCCEEDED" {
|
||||
return false, fmt.Errorf("unexpected workflow state %s; wanted %s", out.State, state)
|
||||
}
|
||||
return false, nil
|
||||
})
|
||||
return out, err
|
||||
}
|
||||
249
sdk/go/examples/workflow-poll/main.go
Normal file
249
sdk/go/examples/workflow-poll/main.go
Normal file
@@ -0,0 +1,249 @@
|
||||
// workflow-poll 区分请求失败重试与成功PENDING后的业务轮次，并演示轮次耗尽进入核查。
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
orbit "git.yuntuo.hk/k2safe/orbit/sdk/go"
|
||||
"git.yuntuo.hk/k2safe/orbit/sdk/go/examples/internal/flowdemo"
|
||||
)
|
||||
|
||||
func main() {
|
||||
if err := run(); err != nil {
|
||||
fmt.Fprintln(os.Stderr, err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
func run() error {
|
||||
api, queue, operation, err := flowdemo.Client()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
||||
defer cancel()
|
||||
for _, exhausted := range []bool{false, true} {
|
||||
if err := scenario(ctx, api, queue, operation, exhausted); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func scenario(ctx context.Context, api *orbit.Client, queue, operation string, exhausted bool) error {
|
||||
mode, expected := "success", "SUCCEEDED"
|
||||
if exhausted {
|
||||
mode, expected = "exhausted", "NEEDS_REVIEW"
|
||||
}
|
||||
operation += "." + mode
|
||||
var requests, submissions atomic.Int32
|
||||
// 这是本地外部渠道夹具。真实业务应查询已存在的外部作业，不能在轮询里重新提交作业。
|
||||
remote := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.URL.Path != "/jobs/"+operation {
|
||||
http.NotFound(w, r)
|
||||
return
|
||||
}
|
||||
if r.Method == "POST" {
|
||||
raw, err := io.ReadAll(io.LimitReader(r.Body, 1025))
|
||||
var in struct {
|
||||
OperationID string `json:"operation_id"`
|
||||
}
|
||||
if err != nil || len(raw) > 1024 || json.Unmarshal(raw, &in) != nil || in.OperationID != operation {
|
||||
http.Error(w, "invalid operation", 400)
|
||||
return
|
||||
}
|
||||
if submissions.Add(1) != 1 {
|
||||
http.Error(w, "unexpected repeated submission", 409)
|
||||
return
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(http.StatusAccepted)
|
||||
_ = json.NewEncoder(w).Encode(in)
|
||||
return
|
||||
}
|
||||
if r.Method != "GET" || submissions.Load() != 1 {
|
||||
http.NotFound(w, r)
|
||||
return
|
||||
}
|
||||
n := requests.Add(1)
|
||||
if !exhausted && n == 1 {
|
||||
http.Error(w, "temporary query failure", http.StatusServiceUnavailable)
|
||||
return
|
||||
}
|
||||
state := "PENDING"
|
||||
if !exhausted && n >= 3 {
|
||||
state = "DONE"
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_ = json.NewEncoder(w).Encode(state)
|
||||
}))
|
||||
defer remote.Close()
|
||||
input, _ := json.Marshal(struct {
|
||||
OperationID string `json:"operation_id"`
|
||||
}{operation})
|
||||
typ := orbit.WorkflowType{Kind: "OBJECT", Fields: map[string]orbit.WorkflowType{"operation_id": {Kind: "STRING"}}}
|
||||
status := orbit.WorkflowType{Kind: "STRING", Enum: []string{"PENDING", "DONE"}}
|
||||
handler := "job.query." + operation
|
||||
task := orbit.WorkflowTask{QueueID: queue, Handler: handler, HandlerVersion: "v1", PayloadVersion: "v1", InputType: typ, Input: orbit.WorkflowRef("input"), ResultType: status, TimeoutMS: 20000, Retry: orbit.RetryPolicy{MaxAttempts: 2, InitialDelayMS: 1000, MaxDelayMS: 1000, AttemptTimeoutMS: 5000}, RecoverySafety: orbit.RecoveryIdempotent}
|
||||
submit := task
|
||||
submit.Handler, submit.ResultType, submit.RecoverySafety = "job.submit."+operation, typ, orbit.RecoveryReconcileOnly
|
||||
submit.Retry = orbit.RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: 5000}
|
||||
task.Input = orbit.WorkflowRef("submit")
|
||||
spec := orbit.WorkflowSpecification{SchemaVersion: 1, InputType: typ, TimeoutMS: 45000, FailurePolicy: "REVIEW", Body: orbit.WorkflowBlock{Steps: []orbit.WorkflowStep{{Key: "submit", Kind: "TASK", Task: &submit}, {Key: "query", Kind: "POLL", Poll: &orbit.WorkflowPoll{Task: task, MaxRounds: 2, TimeoutMS: 30000, IntervalMS: 1000, Pending: []string{"PENDING"}, Succeeded: []string{"DONE"}}}}, OutputType: status, Output: orbit.WorkflowRef("query")}}
|
||||
v, err := flowdemo.PublishStart(ctx, api, operation, "poll."+mode, spec, input)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
httpClient := remote.Client()
|
||||
httpClient.Timeout = 3 * time.Second
|
||||
runner, err := flowdemo.StartWorker(ctx, api, queue, orbit.WorkerHandler{Capability: orbit.WorkerCapability{Handler: handler, HandlerVersion: "v1", PayloadVersion: "v1"}, Handle: func(ctx context.Context, e orbit.WorkerExecution) (json.RawMessage, error) {
|
||||
var in struct {
|
||||
OperationID string `json:"operation_id"`
|
||||
}
|
||||
if json.Unmarshal(e.Task.Payload, &in) != nil || in.OperationID != operation {
|
||||
return nil, &orbit.WorkerFailure{Code: "INVALID_JOB"}
|
||||
}
|
||||
req, err := http.NewRequestWithContext(ctx, "GET", remote.URL+"/jobs/"+in.OperationID, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
resp, err := httpClient.Do(req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
} // 未知传输错误保留UNKNOWN，由冻结安全策略处理。
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode == http.StatusServiceUnavailable {
|
||||
// 本例渠道明确拒绝只读查询，可以声明请求失败且允许Pulse重试。
|
||||
return nil, &orbit.WorkerFailure{Code: "QUERY_TEMPORARY", Retryable: true}
|
||||
}
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return nil, errors.New("unexpected query response")
|
||||
}
|
||||
raw, err := io.ReadAll(io.LimitReader(resp.Body, 1025))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var state string
|
||||
if len(raw) > 1024 || json.Unmarshal(raw, &state) != nil || state != "PENDING" && state != "DONE" {
|
||||
return nil, errors.New("invalid query result")
|
||||
}
|
||||
return json.RawMessage(raw), nil // PENDING是成功查询结果，不能转为可重试错误。
|
||||
}}, orbit.WorkerHandler{Capability: orbit.WorkerCapability{Handler: submit.Handler, HandlerVersion: "v1", PayloadVersion: "v1"}, Handle: func(ctx context.Context, e orbit.WorkerExecution) (json.RawMessage, error) {
|
||||
req, err := http.NewRequestWithContext(ctx, "POST", remote.URL+"/jobs/"+operation, bytes.NewReader(e.Task.Payload))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
resp, err := httpClient.Do(req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
} // 提交结果未知必须核查，不能交给POLL重新提交。
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusAccepted {
|
||||
return nil, errors.New("submission receipt unavailable")
|
||||
}
|
||||
raw, err := io.ReadAll(io.LimitReader(resp.Body, 1025))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var accepted struct {
|
||||
OperationID string `json:"operation_id"`
|
||||
}
|
||||
if len(raw) > 1024 || json.Unmarshal(raw, &accepted) != nil || accepted.OperationID != operation {
|
||||
return nil, errors.New("submission identity mismatch")
|
||||
}
|
||||
return json.RawMessage(raw), nil
|
||||
}})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer runner.Close()
|
||||
initial, err := runner.Step(ctx, api, v.ID, "query", func(d orbit.WorkflowStepDetail) bool { return d.Poll != nil && d.Step.ActivationID != nil })
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
final, err := runner.Workflow(ctx, api, v.ID, expected)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
step, err := api.GetWorkflowStep(ctx, v.ID, "query")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if step.Poll == nil || step.Step.ActivationID == nil || step.Poll.UsedRounds != 2 || step.Poll.MaxAttemptsPerRound != 2 || !step.Poll.DeadlineAt.Equal(initial.Poll.DeadlineAt) || !final.DeadlineAt.Equal(v.DeadlineAt) || *step.Step.ActivationID != *initial.Step.ActivationID {
|
||||
return errors.New("poll budget, deadline or activation changed")
|
||||
}
|
||||
// 历史按业务轮倒序；每轮再独立读取Pulse尝试，二者不可混为同一个计数。
|
||||
cursor, rounds, totalAttempts := "", 0, 0
|
||||
for {
|
||||
page, err := api.ListWorkflowPollRounds(ctx, v.ID, "query", *step.Step.ActivationID, orbit.HistoryPageRequest{Limit: 1, Cursor: cursor})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, round := range page.Items {
|
||||
rounds++
|
||||
source, err := api.GetTaskSource(ctx, round.TaskID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if source.Workflow == nil || source.Workflow.WorkflowID != v.ID || source.Workflow.PollRound != round.Round {
|
||||
return errors.New("query round source mismatch")
|
||||
}
|
||||
attempts, err := api.ListTaskAttempts(ctx, round.TaskID, orbit.HistoryPageRequest{Limit: 10})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
want := 1
|
||||
if !exhausted && round.Round == 1 {
|
||||
want = 2
|
||||
}
|
||||
if len(attempts.Items) != want || attempts.NextCursor != "" {
|
||||
return errors.New("request retries consumed wrong business rounds")
|
||||
}
|
||||
for i, a := range attempts.Items {
|
||||
expect := "SUCCEEDED"
|
||||
if want == 2 && i == 1 {
|
||||
expect = "FAILED"
|
||||
}
|
||||
if string(a.State) != expect || a.EndedAt == nil {
|
||||
return errors.New("unexpected request attempt history")
|
||||
}
|
||||
}
|
||||
totalAttempts += len(attempts.Items)
|
||||
}
|
||||
if page.NextCursor == "" {
|
||||
break
|
||||
}
|
||||
cursor = page.NextCursor
|
||||
}
|
||||
expectedRequests := 3
|
||||
if exhausted {
|
||||
expectedRequests = 2
|
||||
}
|
||||
if rounds != 2 || totalAttempts != expectedRequests || int(requests.Load()) != expectedRequests || submissions.Load() != 1 {
|
||||
return errors.New("unexpected query count")
|
||||
}
|
||||
if !exhausted && string(final.Output) != `"DONE"` {
|
||||
return errors.New("final business result missing")
|
||||
}
|
||||
if exhausted && (step.Step.State != "NEEDS_REVIEW" || step.Poll.CurrentTaskID != nil || step.Poll.NextPollAt != nil) {
|
||||
return errors.New("exhausted poll remained runnable")
|
||||
}
|
||||
if err := runner.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
stats := runner.Worker.Stats()
|
||||
if int(stats.Started) != expectedRequests+1 || int(stats.ReportsConfirmed) != expectedRequests+1 || stats.Running != 0 {
|
||||
return errors.New("Worker did not drain all requests")
|
||||
}
|
||||
fmt.Printf("poll_scenario=%s state=%s submissions=%d rounds=%d attempts=%d requests=%d deadline_preserved=true worker_drained=true\n", mode, final.State, submissions.Load(), rounds, totalAttempts, requests.Load())
|
||||
return nil
|
||||
}
|
||||
284
sdk/go/examples/workflow-recovery/main.go
Normal file
284
sdk/go/examples/workflow-recovery/main.go
Normal file
@@ -0,0 +1,284 @@
|
||||
// workflow-recovery 注入本地渠道提交后断线，使用明确证据恢复，保留UNKNOWN尝试和暂停门控。
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
orbit "git.yuntuo.hk/k2safe/orbit/sdk/go"
|
||||
"git.yuntuo.hk/k2safe/orbit/sdk/go/examples/internal/flowdemo"
|
||||
)
|
||||
|
||||
const exactSequence uint64 = 9007199254740993
|
||||
|
||||
type request struct {
|
||||
OperationID string `json:"operation_id"` // OperationID 原外部操作身份，核查时保持不变。
|
||||
Sequence uint64 `json:"sequence"` // Sequence 精确业务序号，不使用float64解析。
|
||||
}
|
||||
type receipt struct {
|
||||
OperationID string `json:"operation_id"` // OperationID 已完成操作的稳定身份。
|
||||
Sequence uint64 `json:"sequence"` // Sequence 渠道已保存的精确序号。
|
||||
State string `json:"state"` // State 本夹具唯一可核实的成功状态CONFIRMED。
|
||||
}
|
||||
|
||||
func main() {
|
||||
if err := run(); err != nil {
|
||||
fmt.Fprintln(os.Stderr, err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
func run() error {
|
||||
api, queue, operation, err := flowdemo.Client()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second)
|
||||
defer cancel()
|
||||
var mu sync.Mutex
|
||||
var committed json.RawMessage
|
||||
var submissions atomic.Int32
|
||||
var dropped atomic.Bool
|
||||
// 本地夹具代表唯一受控外部系统：先保存结果，再关闭HTTP连接而不发送回执。
|
||||
// 真实人工核查需要持久业务证据及执行已结束的证明，不能只凭GET成功推断所有执行已停止。
|
||||
remote := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.URL.Path != "/objects/"+operation {
|
||||
http.NotFound(w, r)
|
||||
return
|
||||
}
|
||||
if r.Method == "GET" {
|
||||
mu.Lock()
|
||||
raw := append([]byte(nil), committed...)
|
||||
mu.Unlock()
|
||||
if len(raw) == 0 {
|
||||
http.NotFound(w, r)
|
||||
return
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_, _ = w.Write(raw)
|
||||
return
|
||||
}
|
||||
if r.Method != "POST" {
|
||||
w.WriteHeader(http.StatusMethodNotAllowed)
|
||||
return
|
||||
}
|
||||
raw, err := io.ReadAll(io.LimitReader(r.Body, 4097))
|
||||
var in request
|
||||
if err != nil || len(raw) > 4096 || json.Unmarshal(raw, &in) != nil || in.OperationID != operation || in.Sequence != exactSequence {
|
||||
http.Error(w, "invalid command", 400)
|
||||
return
|
||||
}
|
||||
if submissions.Add(1) != 1 {
|
||||
http.Error(w, "unexpected resubmission", 409)
|
||||
return
|
||||
}
|
||||
evidence, _ := json.Marshal(receipt{operation, exactSequence, "CONFIRMED"})
|
||||
mu.Lock()
|
||||
committed = evidence
|
||||
mu.Unlock()
|
||||
conn, _, err := w.(http.Hijacker).Hijack()
|
||||
if err == nil {
|
||||
dropped.Store(true)
|
||||
_ = conn.Close()
|
||||
}
|
||||
}))
|
||||
defer remote.Close()
|
||||
httpClient := remote.Client()
|
||||
httpClient.Timeout = 3 * time.Second
|
||||
requestType := orbit.WorkflowType{Kind: "OBJECT", Fields: map[string]orbit.WorkflowType{"operation_id": {Kind: "STRING"}, "sequence": {Kind: "NUMBER"}}}
|
||||
receiptType := orbit.WorkflowType{Kind: "OBJECT", Fields: map[string]orbit.WorkflowType{"operation_id": {Kind: "STRING"}, "sequence": {Kind: "NUMBER"}, "state": {Kind: "STRING", Enum: []string{"CONFIRMED"}}}}
|
||||
publishHandler, recordHandler := "object.publish."+operation, "object.record."+operation
|
||||
publish := orbit.WorkflowTask{QueueID: queue, Handler: publishHandler, HandlerVersion: "v1", PayloadVersion: "v1", InputType: requestType, Input: orbit.WorkflowRef("input"), ResultType: receiptType, TimeoutMS: 30000, Retry: orbit.RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: 5000}, RecoverySafety: orbit.RecoveryReconcileOnly}
|
||||
record := publish
|
||||
record.Handler = recordHandler
|
||||
record.InputType = receiptType
|
||||
record.Input = orbit.WorkflowRef("publish")
|
||||
record.RecoverySafety = orbit.RecoveryIdempotent
|
||||
spec := orbit.WorkflowSpecification{SchemaVersion: 1, InputType: requestType, TimeoutMS: 60000, FailurePolicy: "REVIEW", Body: orbit.WorkflowBlock{Steps: []orbit.WorkflowStep{{Key: "publish", Kind: "TASK", Task: &publish}, {Key: "record", Kind: "TASK", Task: &record}}, OutputType: receiptType, Output: orbit.WorkflowRef("record")}}
|
||||
input, _ := json.Marshal(request{operation, exactSequence})
|
||||
v, err := flowdemo.PublishStart(ctx, api, operation, "recovery", spec, input)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
runner, err := flowdemo.StartWorker(ctx, api, queue,
|
||||
orbit.WorkerHandler{Capability: orbit.WorkerCapability{Handler: publishHandler, HandlerVersion: "v1", PayloadVersion: "v1"}, Handle: func(ctx context.Context, e orbit.WorkerExecution) (json.RawMessage, error) {
|
||||
req, err := http.NewRequestWithContext(ctx, "POST", remote.URL+"/objects/"+operation, bytes.NewReader(e.Task.Payload))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
resp, err := httpClient.Do(req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
} // 回执丢失只能报UNKNOWN，不能声明明确失败以触发重发。
|
||||
resp.Body.Close()
|
||||
return nil, errors.New("fixture expected a lost response; inspect remote operation")
|
||||
}},
|
||||
orbit.WorkerHandler{Capability: orbit.WorkerCapability{Handler: recordHandler, HandlerVersion: "v1", PayloadVersion: "v1"}, Handle: func(ctx context.Context, e orbit.WorkerExecution) (json.RawMessage, error) {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := verifyReceipt(e.Task.Payload, operation); err != nil {
|
||||
return nil, &orbit.WorkerFailure{Code: "INVALID_RECEIPT"}
|
||||
}
|
||||
return e.Task.Payload, nil
|
||||
}})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer runner.Close()
|
||||
step, err := runner.Step(ctx, api, v.ID, "publish", func(d orbit.WorkflowStepDetail) bool { return d.Step.State == "NEEDS_REVIEW" })
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if step.Step.TaskID == nil || step.Step.ActivationID == nil {
|
||||
return errors.New("review source missing")
|
||||
}
|
||||
taskID, activation := *step.Step.TaskID, *step.Step.ActivationID
|
||||
task, err := api.GetTask(ctx, taskID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
attempts, err := api.ListTaskAttempts(ctx, taskID, orbit.HistoryPageRequest{Limit: 2})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if task.State != orbit.TaskNeedsReview || len(attempts.Items) != 1 || attempts.Items[0].State != "UNKNOWN" || attempts.Items[0].EndedAt == nil || submissions.Load() != 1 || !dropped.Load() {
|
||||
return errors.New("lost response did not retain one UNKNOWN execution")
|
||||
}
|
||||
if err := runner.Wait(ctx, func() (bool, error) {
|
||||
s := runner.Worker.Stats()
|
||||
return s.Running == 0 && s.ReportsConfirmed == 1, nil
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
current, err := api.GetWorkflow(ctx, v.ID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err = api.ControlWorkflow(ctx, v.ID, operation+".pause", orbit.WorkflowControlRequest{Operation: "PAUSE", ExpectedControlRevision: current.ControlRevision, Reason: "hold successor while operator verifies evidence"}); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// 这里只在受控夹具证明“唯一提交已经结束、操作身份和结果一致”后模拟管理员决定。
|
||||
req, err := http.NewRequestWithContext(ctx, "GET", remote.URL+"/objects/"+operation, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
resp, err := httpClient.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
evidence, readErr := io.ReadAll(io.LimitReader(resp.Body, 4097))
|
||||
resp.Body.Close()
|
||||
if readErr != nil {
|
||||
return readErr
|
||||
}
|
||||
if resp.StatusCode != 200 || len(evidence) > 4096 {
|
||||
return errors.New("verified external evidence unavailable")
|
||||
}
|
||||
if err := verifyReceipt(evidence, operation); err != nil {
|
||||
return err
|
||||
}
|
||||
step, err = api.GetWorkflowStep(ctx, v.ID, "publish")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
evidenceRef := "evidence:local-fixture/" + operation
|
||||
decision := orbit.WorkflowRecoveryDecision{ExpectedRevision: step.Revision, Reason: "verified fixture commit, original operation and settled execution", EvidenceRef: evidenceRef}
|
||||
command := orbit.ReviewWorkflowTaskRequest{WorkflowRecoveryDecision: decision, TaskID: taskID, ExpectedTaskRevision: task.Revision, ExecutionSettled: true, Outcome: orbit.TaskSucceeded, Result: evidence}
|
||||
reviewed, err := api.ReviewWorkflowTask(ctx, v.ID, operation+".review", command)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
again, err := api.ReviewWorkflowTask(ctx, v.ID, operation+".review", command)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !again.Deduplicated || again.AuditID != reviewed.AuditID || reviewed.Task == nil || reviewed.TaskAuditID == nil {
|
||||
return errors.New("review identity or dual audit missing")
|
||||
}
|
||||
step, err = runner.Step(ctx, api, v.ID, "publish", func(d orbit.WorkflowStepDetail) bool { return d.Revision > reviewed.Instance.Revision })
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if step.Step.ActivationID == nil || *step.Step.ActivationID != activation {
|
||||
return errors.New("review changed activation")
|
||||
}
|
||||
decision.ExpectedRevision = step.Revision
|
||||
resolved, err := api.ResolveWorkflowStep(ctx, v.ID, operation+".resolve", orbit.ResolveWorkflowStepRequest{WorkflowStepDecision: orbit.WorkflowStepDecision{WorkflowRecoveryDecision: decision, StepKey: "publish", ActivationID: activation, ExpectedStepRevision: step.Step.Revision}, ExecutionSettled: true, Outcome: "SUCCEEDED", Result: evidence})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
current, err = api.GetWorkflow(ctx, v.ID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
decision.ExpectedRevision = current.Revision
|
||||
continued, err := api.ContinueWorkflow(ctx, v.ID, operation+".continue", orbit.ContinueWorkflowRequest{WorkflowRecoveryDecision: decision, ReviewResolved: true})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
next, err := api.GetWorkflowStep(ctx, v.ID, "record")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !continued.Instance.Paused || next.Step.ActivationID != nil || next.Step.TaskID != nil || next.Step.State != "BLOCKED" {
|
||||
return errors.New("manual recovery bypassed pause gate")
|
||||
}
|
||||
for action, id := range map[string]string{"workflow.task.review": reviewed.AuditID, "workflow.step.resolve": resolved.AuditID, "workflow.review.continue": continued.AuditID} {
|
||||
audit, err := api.GetWorkflowOperation(ctx, v.ID, id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if audit.Action != action || audit.EvidenceRef != evidenceRef || audit.Reason != decision.Reason || audit.ActorID == "" {
|
||||
return errors.New("management evidence or actor missing")
|
||||
}
|
||||
}
|
||||
current, err = api.GetWorkflow(ctx, v.ID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err = api.ControlWorkflow(ctx, v.ID, operation+".resume", orbit.WorkflowControlRequest{Operation: "RESUME", ExpectedControlRevision: current.ControlRevision, Reason: "verified review complete; explicitly release pause"}); err != nil {
|
||||
return err
|
||||
}
|
||||
final, err := runner.Workflow(ctx, api, v.ID, "SUCCEEDED")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := verifyReceipt(final.Output, operation); err != nil {
|
||||
return err
|
||||
}
|
||||
old, err := api.GetTaskAttempt(ctx, taskID, attempts.Items[0].ID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if old.State != "UNKNOWN" || old.EndedAt == nil || !old.EndedAt.Equal(*attempts.Items[0].EndedAt) || submissions.Load() != 1 || !final.DeadlineAt.Equal(v.DeadlineAt) {
|
||||
return errors.New("recovery rewrote history, repeated submission or extended deadline")
|
||||
}
|
||||
if err := runner.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
stats := runner.Worker.Stats()
|
||||
if stats.Started != 2 || stats.ReportsConfirmed != 2 || stats.Running != 0 {
|
||||
return errors.New("unexpected executions or incomplete drain")
|
||||
}
|
||||
fmt.Printf("recovery_verified=true state=%s submissions=%d unknown_history_preserved=true review_deduplicated=true pause_preserved=true audits=3 sequence=%d worker_drained=true\n", final.State, submissions.Load(), exactSequence)
|
||||
return nil
|
||||
}
|
||||
|
||||
func verifyReceipt(raw json.RawMessage, operation string) error {
|
||||
var out receipt
|
||||
if json.Unmarshal(raw, &out) != nil || out.OperationID != operation || out.Sequence != exactSequence || out.State != "CONFIRMED" {
|
||||
return errors.New("external receipt identity or exact sequence mismatch")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
Reference in New Issue
Block a user