yms-daemon/internal/transaction/coordinator_test.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
}