184 lines
7.2 KiB
Go
184 lines
7.2 KiB
Go
package orbit
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"io"
|
||
"net/http"
|
||
"net/http/httptest"
|
||
"net/http/httputil"
|
||
"net/url"
|
||
"reflect"
|
||
"sync/atomic"
|
||
"testing"
|
||
"time"
|
||
)
|
||
|
||
func TestIntegrationTaskRecoverySDK(t *testing.T) {
|
||
api, queue, token := integrationWorkerClient(t)
|
||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||
defer cancel()
|
||
handler := "recovery." + workerTestID(t)
|
||
accepted, err := api.SubmitTask(ctx, workerTestID(t), SubmitTaskRequest{QueueID: queue, Handler: handler, HandlerVersion: "v1", PayloadVersion: "v1", BusinessKey: "object:recovery-example", Payload: json.RawMessage(`{"sequence":9007199254740993}`), TaskTimeoutMS: 60000, Retry: RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: 10000}})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
reg, err := NewWorkerRegistration(queue, 1, []WorkerCapability{{Handler: handler, HandlerVersion: "v1", PayloadVersion: "v1"}})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
w, err := api.DialWorker(ctx, reg)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
defer w.Close()
|
||
if _, err = w.Capacity(ctx, workerTestID(t), WorkerCapacity{Sequence: 1, Capacity: 1}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
a, err := w.NextAssignment(ctx)
|
||
if err != nil || a.Task.ID != accepted.Task.ID {
|
||
t.Fatal(a, err)
|
||
}
|
||
if _, err = w.Start(ctx, workerTestID(t), a.Attempt()); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err = w.Finish(ctx, workerTestID(t), WorkerFinish{Ref: a.Attempt(), Outcome: WorkerUnknown, Result: json.RawMessage(`{"status":"unknown"}`), ErrorCode: "EXAMPLE_UNCERTAIN"}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
unknown := waitRuntimeState(t, ctx, api, a.Task.ID, TaskNeedsReview)
|
||
oldAttempt, err := api.GetTaskAttempt(ctx, a.Task.ID, a.AttemptID)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
// 后端完整提交后主动丢失HTTP响应。禁用连接复用，排除net/http对复用连接的隐式重发。
|
||
target, _ := url.Parse(api.baseURL)
|
||
proxy := httputil.NewSingleHostReverseProxy(target)
|
||
var drop atomic.Bool
|
||
var calls atomic.Int32
|
||
var committed atomic.Value
|
||
proxy.ModifyResponse = func(r *http.Response) error {
|
||
calls.Add(1)
|
||
if !drop.CompareAndSwap(true, false) {
|
||
return nil
|
||
}
|
||
raw, e := io.ReadAll(r.Body)
|
||
r.Body.Close()
|
||
if e != nil {
|
||
return e
|
||
}
|
||
if r.StatusCode != 200 && r.StatusCode != 201 {
|
||
t.Errorf("unexpected response before drop: %s", raw)
|
||
}
|
||
committed.Store(raw)
|
||
return io.ErrUnexpectedEOF
|
||
}
|
||
proxy.ErrorHandler = func(w http.ResponseWriter, r *http.Request, err error) {
|
||
conn, _, e := w.(http.Hijacker).Hijack()
|
||
if e == nil {
|
||
conn.Close()
|
||
}
|
||
}
|
||
server := httptest.NewServer(proxy)
|
||
defer server.Close()
|
||
transport := &http.Transport{DisableKeepAlives: true}
|
||
defer transport.CloseIdleConnections()
|
||
lost, err := NewClient(server.URL, Options{Token: token, HTTPClient: &http.Client{Transport: transport}})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
review := ReviewTaskRequest{ExpectedRevision: unknown.Revision, Reason: "确认模拟执行已结束且未完成", EvidenceRef: "evidence:demo-reconciliation", ExecutionSettled: true, Outcome: TaskFailed, Result: json.RawMessage(`{"confirmed":true,"sequence":9007199254740993}`)}
|
||
reviewKey := workerTestID(t)
|
||
drop.Store(true)
|
||
if _, err = lost.ReviewTask(ctx, unknown.ID, reviewKey, review); err == nil || calls.Load() != 1 {
|
||
t.Fatal("review response loss was hidden", err)
|
||
}
|
||
var reviewEnvelope struct {
|
||
Data ReviewTaskResult `json:"data"`
|
||
}
|
||
if err = json.Unmarshal(committed.Load().([]byte), &reviewEnvelope); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
reviewed, err := api.ReviewTask(ctx, unknown.ID, reviewKey, review)
|
||
if err != nil || !reviewed.Deduplicated || reviewed.AuditID != reviewEnvelope.Data.AuditID || reviewed.Task.State != TaskFailed || reviewed.Task.EffectUncertain {
|
||
t.Fatal(reviewed, err)
|
||
}
|
||
attempt, err := api.GetTaskAttempt(ctx, a.Task.ID, a.AttemptID)
|
||
if err != nil || !reflect.DeepEqual(attempt, oldAttempt) {
|
||
t.Fatal("review changed immutable attempt", err)
|
||
}
|
||
audit, err := api.GetTaskOperation(ctx, unknown.ID, reviewed.AuditID)
|
||
if err != nil || audit.Action != "task.review" || audit.EvidenceRef != review.EvidenceRef || audit.Reason != review.Reason {
|
||
t.Fatal(audit, err)
|
||
}
|
||
retry := RetryTaskRequest{ExpectedRevision: reviewed.Task.Revision, Reason: "已核实模拟旧执行不会再产生效果，允许重新执行", EvidenceRef: "evidence:demo-safe-retry", SafeToRetry: true, TaskTimeoutMS: 60000}
|
||
retryKey := workerTestID(t)
|
||
drop.Store(true)
|
||
if _, err = lost.RetryTask(ctx, unknown.ID, retryKey, retry); err == nil || calls.Load() != 2 {
|
||
t.Fatal("retry response loss was hidden", err)
|
||
}
|
||
var retryEnvelope struct {
|
||
Data RetryTaskResult `json:"data"`
|
||
}
|
||
if err = json.Unmarshal(committed.Load().([]byte), &retryEnvelope); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
child, err := api.RetryTask(ctx, unknown.ID, retryKey, retry)
|
||
if err != nil || !child.Deduplicated || child.Task.ID != retryEnvelope.Data.Task.ID || child.AuditID != retryEnvelope.Data.AuditID || child.Task.RetryOf == nil || *child.Task.RetryOf != unknown.ID {
|
||
t.Fatal(child, err)
|
||
}
|
||
lineage, err := api.GetTaskSource(ctx, child.Task.ID)
|
||
if err != nil || lineage.Kind != "MANUAL_RETRY" || lineage.RetryOf == nil || *lineage.RetryOf != unknown.ID || lineage.Schedule != nil {
|
||
t.Fatal("manual retry source", lineage, err)
|
||
}
|
||
originalSource, err := api.GetTaskSource(ctx, unknown.ID)
|
||
if err != nil || originalSource.Kind != "DIRECT" || originalSource.RetryOf != nil {
|
||
t.Fatal("direct source", originalSource, err)
|
||
}
|
||
assigned, err := w.NextAssignment(ctx)
|
||
if err != nil || assigned.Task.ID != child.Task.ID || assigned.Task.RetryOf == nil || *assigned.Task.RetryOf != unknown.ID {
|
||
t.Fatal("wire lost retry lineage", assigned, err)
|
||
}
|
||
if _, err = w.Start(ctx, workerTestID(t), assigned.Attempt()); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err = w.Finish(ctx, workerTestID(t), WorkerFinish{Ref: assigned.Attempt(), Outcome: WorkerSucceeded, Result: json.RawMessage(`{"object_ref":"examples/output"}`)}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
waitRuntimeState(t, ctx, api, child.Task.ID, TaskSucceeded)
|
||
late, err := w.Finish(ctx, workerTestID(t), WorkerFinish{Ref: a.Attempt(), Outcome: WorkerSucceeded, Result: json.RawMessage(`{"late":true}`)})
|
||
if err != nil || late.Receipt.Disposition != "EVIDENCE_ONLY" {
|
||
t.Fatal(late, err)
|
||
}
|
||
source, err := api.GetTask(ctx, unknown.ID)
|
||
if err != nil || !reflect.DeepEqual(source, reviewed.Task) {
|
||
t.Fatal("late source rewrote review", err)
|
||
}
|
||
seen := map[string]bool{}
|
||
actions := map[string]int{}
|
||
cursor := ""
|
||
for {
|
||
page, e := api.ListTaskOperations(ctx, unknown.ID, HistoryPageRequest{Limit: 1, Cursor: cursor})
|
||
if e != nil {
|
||
t.Fatal(e)
|
||
}
|
||
for _, v := range page.Items {
|
||
if seen[v.ID] || v.TaskID != unknown.ID {
|
||
t.Fatal("duplicate/unscoped audit")
|
||
}
|
||
seen[v.ID] = true
|
||
actions[v.Action]++
|
||
}
|
||
if page.NextCursor == "" {
|
||
break
|
||
}
|
||
cursor = page.NextCursor
|
||
}
|
||
if actions["task.review"] != 1 || actions["task.retry"] != 1 {
|
||
t.Fatal("response loss created duplicate operations", actions)
|
||
}
|
||
if _, err = w.Capacity(ctx, workerTestID(t), WorkerCapacity{Sequence: 2, Drain: true}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
t.Log("real UNKNOWN -> reviewed FAILED -> linked new task SUCCEEDED; both committed HTTP receipts lost/recovered; old attempt and audit preserved; late report evidence only")
|
||
}
|