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 }