231 lines
9.6 KiB
Go
231 lines
9.6 KiB
Go
package transaction
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"io"
|
|
"log/slog"
|
|
"sync/atomic"
|
|
"testing"
|
|
)
|
|
|
|
// TestCoordinatorRecoversIntentWithoutRepeatingAppliedOperation 验证崩溃恢复时，
|
|
// 若已记录意图的外部副作用经 Inspect 确认为已生效，则直接补记成功而不再重复 Apply。
|
|
func TestCoordinatorRecoversIntentWithoutRepeatingAppliedOperation(t *testing.T) {
|
|
t.Parallel()
|
|
ctx := context.Background()
|
|
store := openTestStore(t)
|
|
record := createTestTransaction(t, store, "recover-applied")
|
|
intent := StepIntent{Key: "start-green", Name: "start green", Intent: json.RawMessage(`{"slot":"green"}`)}
|
|
if _, _, err := store.RecordStepIntent(ctx, record.ID, intent); err != nil {
|
|
t.Fatalf("record crash-window intent: %v", err)
|
|
}
|
|
operation := &fakeOperation{applied: true, result: json.RawMessage(`{"running":true}`)}
|
|
coordinator := newTestCoordinator(t, store)
|
|
step, err := coordinator.ExecuteStep(ctx, record.ID, intent, operation)
|
|
if err != nil {
|
|
t.Fatalf("recover applied operation: %v", err)
|
|
}
|
|
if step.Status != StepSucceeded || operation.applyCalls.Load() != 0 || operation.inspectCalls.Load() != 1 {
|
|
t.Fatalf("unexpected recovery result: step=%+v apply=%d inspect=%d", step, operation.applyCalls.Load(), operation.inspectCalls.Load())
|
|
}
|
|
}
|
|
|
|
// TestCoordinatorRecoversNotAppliedIntentThenExecutesOnce 验证已记录意图但 Inspect
|
|
// 确认副作用未发生时，恢复流程会重新 Apply 且只执行一次，最终补记成功。
|
|
func TestCoordinatorRecoversNotAppliedIntentThenExecutesOnce(t *testing.T) {
|
|
t.Parallel()
|
|
ctx := context.Background()
|
|
store := openTestStore(t)
|
|
record := createTestTransaction(t, store, "recover-not-applied")
|
|
intent := StepIntent{Key: "write-files", Name: "write files", Intent: json.RawMessage(`{}`)}
|
|
if _, _, err := store.RecordStepIntent(ctx, record.ID, intent); err != nil {
|
|
t.Fatalf("record crash-window intent: %v", err)
|
|
}
|
|
operation := &fakeOperation{result: json.RawMessage(`{"written":true}`)}
|
|
coordinator := newTestCoordinator(t, store)
|
|
step, err := coordinator.ExecuteStep(ctx, record.ID, intent, operation)
|
|
if err != nil {
|
|
t.Fatalf("recover not-applied operation: %v", err)
|
|
}
|
|
if step.Status != StepSucceeded || operation.applyCalls.Load() != 1 || operation.inspectCalls.Load() != 2 {
|
|
t.Fatalf("unexpected recovery result: step=%+v apply=%d inspect=%d", step, operation.applyCalls.Load(), operation.inspectCalls.Load())
|
|
}
|
|
}
|
|
|
|
// TestCoordinatorLeavesIntentPendingWhenInspectionIsUnknown 验证 Inspect 无法确认
|
|
// 副作用状态时，返回 UncertainStepError 且步骤保留 INTENT_RECORDED 等待后续恢复。
|
|
func TestCoordinatorLeavesIntentPendingWhenInspectionIsUnknown(t *testing.T) {
|
|
t.Parallel()
|
|
ctx := context.Background()
|
|
store := openTestStore(t)
|
|
record := createTestTransaction(t, store, "recover-unknown")
|
|
intent := StepIntent{Key: "reload-gateway", Name: "reload gateway", Intent: json.RawMessage(`{}`)}
|
|
operation := &fakeOperation{inspectErr: errors.New("gateway unavailable")}
|
|
coordinator := newTestCoordinator(t, store)
|
|
step, err := coordinator.ExecuteStep(ctx, record.ID, intent, operation)
|
|
var uncertain *UncertainStepError
|
|
if !errors.As(err, &uncertain) {
|
|
t.Fatalf("expected uncertain step error, got step=%+v err=%v", step, err)
|
|
}
|
|
pending, err := store.PendingSteps(ctx, record.ID)
|
|
if err != nil {
|
|
t.Fatalf("read pending steps: %v", err)
|
|
}
|
|
if len(pending) != 1 || pending[0].Status != StepIntentRecorded {
|
|
t.Fatalf("uncertain step did not remain pending: %+v", pending)
|
|
}
|
|
}
|
|
|
|
// TestCoordinatorRetriesFailedStepOnlyAfterManualResumeInspection 验证失败的步骤不会
|
|
// 自动重试，只有人工恢复后再次调用 ExecuteStep，经 Inspect 确认未生效才重新 Apply。
|
|
func TestCoordinatorRetriesFailedStepOnlyAfterManualResumeInspection(t *testing.T) {
|
|
t.Parallel()
|
|
ctx := context.Background()
|
|
store := openTestStore(t)
|
|
record := createTestTransaction(t, store, "retry-failed")
|
|
intent := StepIntent{Key: "restore-gateway", Name: "restore gateway", Intent: json.RawMessage(`{}`)}
|
|
operation := &fakeOperation{applyErr: errors.New("gateway executable unavailable"), result: json.RawMessage(`{"restored":true}`)}
|
|
coordinator := newTestCoordinator(t, store)
|
|
|
|
failed, err := coordinator.ExecuteStep(ctx, record.ID, intent, operation)
|
|
if err == nil || failed.Status != StepFailed {
|
|
t.Fatalf("unexpected initial failed step: step=%+v err=%v", failed, err)
|
|
}
|
|
if operation.applyCalls.Load() != 1 || operation.inspectCalls.Load() != 1 {
|
|
t.Fatalf("initial call retried unexpectedly: apply=%d inspect=%d", operation.applyCalls.Load(), operation.inspectCalls.Load())
|
|
}
|
|
|
|
operation.applyErr = nil
|
|
recovered, err := coordinator.ExecuteStep(ctx, record.ID, intent, operation)
|
|
if err != nil || recovered.Status != StepSucceeded {
|
|
t.Fatalf("recover failed step: step=%+v err=%v", recovered, err)
|
|
}
|
|
if operation.applyCalls.Load() != 2 || operation.inspectCalls.Load() != 3 {
|
|
t.Fatalf("unexpected manual recovery calls: apply=%d inspect=%d", operation.applyCalls.Load(), operation.inspectCalls.Load())
|
|
}
|
|
}
|
|
|
|
// TestCoordinatorCompletesFailedStepAlreadyAppliedBeforeManualResume 验证失败步骤在
|
|
// 人工恢复时经 Inspect 发现副作用其实已生效，则直接补记成功而不重复 Apply。
|
|
func TestCoordinatorCompletesFailedStepAlreadyAppliedBeforeManualResume(t *testing.T) {
|
|
t.Parallel()
|
|
ctx := context.Background()
|
|
store := openTestStore(t)
|
|
record := createTestTransaction(t, store, "recover-failed-applied")
|
|
intent := StepIntent{Key: "restore-gateway", Name: "restore gateway", Intent: json.RawMessage(`{}`)}
|
|
operation := &fakeOperation{applyErr: errors.New("gateway reload result unavailable"), result: json.RawMessage(`{"restored":true}`)}
|
|
coordinator := newTestCoordinator(t, store)
|
|
|
|
if _, err := coordinator.ExecuteStep(ctx, record.ID, intent, operation); err == nil {
|
|
t.Fatal("expected initial external step failure")
|
|
}
|
|
operation.applyErr = nil
|
|
operation.applied = true
|
|
recovered, err := coordinator.ExecuteStep(ctx, record.ID, intent, operation)
|
|
if err != nil || recovered.Status != StepSucceeded {
|
|
t.Fatalf("complete externally applied failed step: step=%+v err=%v", recovered, err)
|
|
}
|
|
if operation.applyCalls.Load() != 1 || operation.inspectCalls.Load() != 2 {
|
|
t.Fatalf("externally applied step was repeated: apply=%d inspect=%d", operation.applyCalls.Load(), operation.inspectCalls.Load())
|
|
}
|
|
}
|
|
|
|
// TestCoordinatorExclusiveExecutionHonorsContext 验证 RunExclusive 的互斥与上下文取消：
|
|
// 已有流程占用执行区时，后续取消的上下文会立即返回 context.Canceled，且不进入执行区。
|
|
func TestCoordinatorExclusiveExecutionHonorsContext(t *testing.T) {
|
|
t.Parallel()
|
|
store := openTestStore(t)
|
|
coordinator := newTestCoordinator(t, store)
|
|
firstEntered := make(chan struct{})
|
|
releaseFirst := make(chan struct{})
|
|
firstDone := make(chan error, 1)
|
|
go func() {
|
|
firstDone <- coordinator.RunExclusive(context.Background(), func(context.Context) error {
|
|
close(firstEntered)
|
|
<-releaseFirst
|
|
return nil
|
|
})
|
|
}()
|
|
<-firstEntered
|
|
cancelled, cancel := context.WithCancel(context.Background())
|
|
cancel()
|
|
err := coordinator.RunExclusive(cancelled, func(context.Context) error {
|
|
t.Fatal("cancelled update entered exclusive section")
|
|
return nil
|
|
})
|
|
if !errors.Is(err, context.Canceled) {
|
|
t.Fatalf("expected context cancellation, got %v", err)
|
|
}
|
|
close(releaseFirst)
|
|
if err := <-firstDone; err != nil {
|
|
t.Fatalf("first exclusive update failed: %v", err)
|
|
}
|
|
}
|
|
|
|
// fakeOperation 测试用的 Operation 实现，可编程地模拟 Apply 与 Inspect 行为。
|
|
//
|
|
// 字段 applied 表示副作用是否已生效，result 为核对结果，applyErr 与 inspectErr 分别
|
|
// 模拟执行与核对失败；applyCalls 与 inspectCalls 用于断言调用次数。
|
|
type fakeOperation struct {
|
|
applied bool
|
|
result json.RawMessage
|
|
applyErr error
|
|
inspectErr error
|
|
applyCalls atomic.Int32
|
|
inspectCalls atomic.Int32
|
|
}
|
|
|
|
// Apply 模拟执行外部副作用：递增 applyCalls 计数，成功时把 applied 置为 true。
|
|
func (o *fakeOperation) Apply(context.Context) error {
|
|
o.applyCalls.Add(1)
|
|
if o.applyErr == nil {
|
|
o.applied = true
|
|
}
|
|
return o.applyErr
|
|
}
|
|
|
|
// Inspect 模拟核对外部副作用：递增 inspectCalls 计数，根据 applied 返回
|
|
// APPLIED 或 NOT_APPLIED；若设置了 inspectErr 则返回该错误。
|
|
func (o *fakeOperation) Inspect(context.Context) (Inspection, error) {
|
|
o.inspectCalls.Add(1)
|
|
if o.inspectErr != nil {
|
|
return Inspection{}, o.inspectErr
|
|
}
|
|
if o.applied {
|
|
return Inspection{Status: InspectionApplied, Result: o.result}, nil
|
|
}
|
|
return Inspection{Status: InspectionNotApplied}, nil
|
|
}
|
|
|
|
// createTestTransaction 创建一条用于测试的事务记录。
|
|
//
|
|
// 参数 store 为目标存储，suffix 用于生成唯一的事务 ID 与幂等键。创建失败时以
|
|
// t.Fatalf 终止测试并返回错误说明。
|
|
func createTestTransaction(t *testing.T, store *Store, suffix string) Transaction {
|
|
t.Helper()
|
|
record, _, err := store.CreateTransaction(context.Background(), CreateRequest{
|
|
ID: "transaction-" + suffix,
|
|
IdempotencyKey: "request-" + suffix,
|
|
Source: "test",
|
|
Service: "backend",
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("create test transaction: %v", err)
|
|
}
|
|
return record
|
|
}
|
|
|
|
// newTestCoordinator 创建用于测试的 Coordinator，日志输出丢弃到 io.Discard。
|
|
//
|
|
// 参数 store 为底层存储。创建失败时以 t.Fatalf 终止测试。
|
|
func newTestCoordinator(t *testing.T, store *Store) *Coordinator {
|
|
t.Helper()
|
|
coordinator, err := NewCoordinator(store, slog.New(slog.NewTextHandler(io.Discard, nil)))
|
|
if err != nil {
|
|
t.Fatalf("create coordinator: %v", err)
|
|
}
|
|
return coordinator
|
|
}
|