package transaction import ( "context" "encoding/json" "errors" "net/url" "path/filepath" "sync" "testing" "time" "github.com/ncruces/go-sqlite3/driver" ) // TestStoreMigratesVersionOneAndCommitsBackendContainerDeployment 验证从 v1 schema // 迁移到最新版本,以及 backend 容器部署提交的完整链路:迁移前无部署记录、非容器类 // backend 事务不计入容器历史、未完成容器事务不计入、提交后单例记录与历史证据正确。 func TestStoreMigratesVersionOneAndCommitsBackendContainerDeployment(t *testing.T) { t.Parallel() ctx := context.Background() databasePath := filepath.Join(t.TempDir(), "transaction.db") dsn := (&url.URL{Scheme: "file", Path: databasePath}).String() database, err := driver.Open(dsn) if err != nil { t.Fatalf("open version one database: %v", err) } if _, err := database.ExecContext(ctx, schemaV1); err != nil { _ = database.Close() t.Fatalf("create version one database: %v", err) } if err := database.Close(); err != nil { t.Fatalf("close version one database: %v", err) } store, err := OpenStore(ctx, databasePath) if err != nil { t.Fatalf("migrate transaction store: %v", err) } t.Cleanup(func() { _ = store.Close() }) if _, err := store.BackendContainerDeployment(ctx); !errors.Is(err, ErrNotFound) { t.Fatalf("unexpected deployment before commit: %v", err) } hasHistory, err := store.HasCommittedBackendContainerTransactionHistory(ctx) if err != nil || hasHistory { t.Fatalf("unexpected empty backend history: found=%t err=%v", hasHistory, err) } nativeRecord, _, err := store.CreateTransaction(ctx, CreateRequest{ ID: "native-backend-transaction", IdempotencyKey: "backend:native-request", Source: "test", Service: "backend", }) if err != nil { t.Fatalf("create native backend transaction: %v", err) } if _, err := store.Transition(ctx, nativeRecord.ID, StateFailed, "finish native backend transaction"); err != nil { t.Fatalf("finish native backend transaction: %v", err) } hasHistory, err = store.HasCommittedBackendContainerTransactionHistory(ctx) if err != nil || hasHistory { t.Fatalf("native backend history was treated as container history: found=%t err=%v", hasHistory, err) } record, created, err := store.CreateTransaction(ctx, CreateRequest{ ID: "container-deployment-transaction", IdempotencyKey: "backend:container:deployment-request", Source: "test", Service: "backend", }) if err != nil || !created { t.Fatalf("create backend container transaction: record=%+v created=%t err=%v", record, created, err) } for _, state := range []State{StateValidating, StatePrepared, StateStarting, StateSwitching, StateVerifying, StateDraining} { if _, err := store.Transition(ctx, record.ID, state, "test transition"); err != nil { t.Fatalf("transition backend container transaction to %s: %v", state, err) } } hasHistory, err = store.HasCommittedBackendContainerTransactionHistory(ctx) if err != nil || hasHistory { t.Fatalf("unfinished backend container transaction was treated as committed history: found=%t err=%v", hasHistory, err) } committed, err := store.CommitBackendContainerDeployment(ctx, record.ID, BackendContainerDeployment{ ActivePort: 8081, ContainerName: "backend-8081", ImageDigest: "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", ContainerID: "container-backend-8081", }, "container deployment committed") if err != nil || committed.State != StateCommitted { t.Fatalf("commit backend container deployment: record=%+v err=%v", committed, err) } hasHistory, err = store.HasCommittedBackendContainerTransactionHistory(ctx) if err != nil || !hasHistory { t.Fatalf("committed backend container history was not recorded: found=%t err=%v", hasHistory, err) } deployment, err := store.BackendContainerDeployment(ctx) if err != nil { t.Fatalf("read backend container deployment: %v", err) } if deployment.ActivePort != 8081 || deployment.ContainerName != "backend-8081" || deployment.ContainerID != "container-backend-8081" || deployment.TransactionID != record.ID { t.Fatalf("unexpected backend container deployment: %+v", deployment) } var version int if err := store.db.QueryRowContext(ctx, "PRAGMA user_version").Scan(&version); err != nil || version != schemaVersion { t.Fatalf("unexpected migrated schema version: version=%d err=%v", version, err) } } // TestListRecentScansSQLiteRequestTextAsRawJSON 验证 ListRecent 能正确把 SQLite 中 // 存储的 request_json 文本还原为原始 JSON 字节,不丢失也不转义。 func TestListRecentScansSQLiteRequestTextAsRawJSON(t *testing.T) { t.Parallel() ctx := context.Background() store := openTestStore(t) record, _, err := store.CreateTransaction(ctx, CreateRequest{ ID: "list-transaction", IdempotencyKey: "list-idempotency", Source: "test", Service: "backend", Request: json.RawMessage(`{"inputType":"native-jar"}`), }) if err != nil { t.Fatalf("create list transaction: %v", err) } items, err := store.ListRecent(ctx, ListFilter{Limit: 20}) if err != nil { t.Fatalf("list transactions: %v", err) } if len(items) != 1 || items[0].ID != record.ID || string(items[0].Request) != `{"inputType":"native-jar"}` { t.Fatalf("unexpected listed transaction: %+v", items) } } // TestBackendContainerDeploymentCommitRequiresDrainingTransaction 验证 backend 容器 // 部署提交要求事务必须处于 DRAINING 状态,否则返回 TransitionError 且不写入部署行。 func TestBackendContainerDeploymentCommitRequiresDrainingTransaction(t *testing.T) { t.Parallel() ctx := context.Background() store := openTestStore(t) record, _, err := store.CreateTransaction(ctx, CreateRequest{ ID: "container-deployment-wrong-state", IdempotencyKey: "container-deployment-wrong-state-request", Source: "test", Service: "backend", }) if err != nil { t.Fatalf("create backend transaction: %v", err) } _, err = store.CommitBackendContainerDeployment(ctx, record.ID, BackendContainerDeployment{ ActivePort: 8080, ContainerName: "backend-8080", ImageDigest: "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", ContainerID: "container-backend-8080", }, "must fail") var transitionErr *TransitionError if !errors.As(err, &transitionErr) { t.Fatalf("expected deployment commit transition error, got %v", err) } if _, err := store.BackendContainerDeployment(ctx); !errors.Is(err, ErrNotFound) { t.Fatalf("failed commit wrote backend deployment: %v", err) } } // TestStoreCreateIsIdempotentAndAllowsOnlyOneActiveTransaction 验证 CreateTransaction // 的幂等性与单活动事务约束:同幂等键重试返回同一事务;活动事务未结束前新事务被 // ActiveTransactionError 拒绝;失败事务的幂等键被归档后可用新 ID 重试,且原事务的 // 幂等键被改写为带 :terminal: 前缀的归档形式。 func TestStoreCreateIsIdempotentAndAllowsOnlyOneActiveTransaction(t *testing.T) { t.Parallel() ctx := context.Background() store := openTestStore(t) request := CreateRequest{ ID: "transaction-1", IdempotencyKey: "request-1", Source: "test", Service: "backend", Request: json.RawMessage(`{"service":"backend"}`), } created, isNew, err := store.CreateTransaction(ctx, request) if err != nil { t.Fatalf("create transaction: %v", err) } if !isNew || created.State != StateCreated { t.Fatalf("unexpected created transaction: %+v new=%v", created, isNew) } retried, isNew, err := store.CreateTransaction(ctx, request) if err != nil { t.Fatalf("retry transaction: %v", err) } if isNew || retried.ID != created.ID { t.Fatalf("idempotent retry created another transaction: %+v", retried) } _, _, err = store.CreateTransaction(ctx, CreateRequest{ ID: "transaction-2", IdempotencyKey: "request-2", Source: "test", Service: "frontend", }) var activeErr *ActiveTransactionError if !errors.As(err, &activeErr) || activeErr.TransactionID != created.ID { t.Fatalf("expected active transaction error, got %v", err) } if _, err := store.Transition(ctx, created.ID, StateFailed, "test terminal state"); err != nil { t.Fatalf("finish first transaction: %v", err) } retriedAfterFailure, isNew, err := store.CreateTransaction(ctx, CreateRequest{ ID: "transaction-1-retry", IdempotencyKey: request.IdempotencyKey, Source: request.Source, Service: request.Service, Request: request.Request, }) if err != nil || !isNew || retriedAfterFailure.ID != "transaction-1-retry" { t.Fatalf("retry failed transaction: record=%+v new=%v err=%v", retriedAfterFailure, isNew, err) } archived, err := store.Transaction(ctx, created.ID) if err != nil { t.Fatalf("read archived failed transaction: %v", err) } if archived.IdempotencyKey != request.IdempotencyKey+":terminal:"+created.ID { t.Fatalf("unexpected archived idempotency key: %q", archived.IdempotencyKey) } if _, err := store.Transition(ctx, retriedAfterFailure.ID, StateFailed, "finish retried transaction"); err != nil { t.Fatalf("finish retried transaction: %v", err) } second, isNew, err := store.CreateTransaction(ctx, CreateRequest{ ID: "transaction-2", IdempotencyKey: "request-2", Source: "test", Service: "frontend", }) if err != nil || !isNew || second.ID != "transaction-2" { t.Fatalf("create transaction after terminal state: record=%+v new=%v err=%v", second, isNew, err) } } // TestStoreTransitionStepAndEventPersistence 验证状态、步骤与事件在关闭并重开数据库 // 后仍能正确恢复:事务状态与版本持久化、待定步骤可查询、步骤完成落库、事件有序递增。 func TestStoreTransitionStepAndEventPersistence(t *testing.T) { t.Parallel() ctx := context.Background() databasePath := filepath.Join(t.TempDir(), "transaction.db") store, err := OpenStore(ctx, databasePath) if err != nil { t.Fatalf("open store: %v", err) } fixedTime := time.Date(2026, time.August, 15, 10, 0, 0, 0, time.UTC) store.now = func() time.Time { return fixedTime } record, _, err := store.CreateTransaction(ctx, CreateRequest{ ID: "transaction-persisted", IdempotencyKey: "request-persisted", Source: "test", Service: "all", }) if err != nil { t.Fatalf("create transaction: %v", err) } if _, err := store.Transition(ctx, record.ID, StateValidating, "validation started"); err != nil { t.Fatalf("transition transaction: %v", err) } if _, err := store.Transition(ctx, record.ID, StatePrepared, "validation completed"); err != nil { t.Fatalf("transition transaction: %v", err) } step, isNew, err := store.RecordStepIntent(ctx, record.ID, StepIntent{ Key: "prepare-files", Name: "prepare immutable files", Intent: json.RawMessage(`{"sha256":"abc"}`), }) if err != nil || !isNew || step.Status != StepIntentRecorded { t.Fatalf("record step intent: step=%+v new=%v err=%v", step, isNew, err) } if err := store.Close(); err != nil { t.Fatalf("close store: %v", err) } reopened, err := OpenStore(ctx, databasePath) if err != nil { t.Fatalf("reopen store: %v", err) } t.Cleanup(func() { _ = reopened.Close() }) persisted, err := reopened.Transaction(ctx, record.ID) if err != nil { t.Fatalf("read persisted transaction: %v", err) } if persisted.State != StatePrepared || persisted.Version != 3 { t.Fatalf("unexpected persisted transaction: %+v", persisted) } pending, err := reopened.PendingSteps(ctx, record.ID) if err != nil { t.Fatalf("read pending steps: %v", err) } if len(pending) != 1 || pending[0].Key != step.Key { t.Fatalf("unexpected pending steps: %+v", pending) } completed, err := reopened.CompleteStep(ctx, record.ID, step.Key, StepSucceeded, json.RawMessage(`{"installed":true}`), "") if err != nil || completed.Status != StepSucceeded { t.Fatalf("complete step: step=%+v err=%v", completed, err) } events, err := reopened.EventsAfter(ctx, record.ID, 0, 100) if err != nil { t.Fatalf("read events: %v", err) } if len(events) != 5 { t.Fatalf("unexpected event count: got %d events=%+v", len(events), events) } for index := 1; index < len(events); index++ { if events[index].Sequence <= events[index-1].Sequence { t.Fatalf("events are not ordered: %+v", events) } } } // TestStoreRejectsInvalidTransitionAndConflictingStepIntent 验证非法状态转换返回 // TransitionError,以及同一步骤键以不同意图重复记录时返回 ErrStepConflict。 func TestStoreRejectsInvalidTransitionAndConflictingStepIntent(t *testing.T) { t.Parallel() ctx := context.Background() store := openTestStore(t) record, _, err := store.CreateTransaction(ctx, CreateRequest{ ID: "transaction-conflict", IdempotencyKey: "request-conflict", Source: "test", Service: "backend", }) if err != nil { t.Fatalf("create transaction: %v", err) } _, err = store.Transition(ctx, record.ID, StatePrepared, "skip validation") var transitionErr *TransitionError if !errors.As(err, &transitionErr) { t.Fatalf("expected transition error, got %v", err) } intent := StepIntent{Key: "same-key", Name: "first", Intent: json.RawMessage(`{"value":1}`)} if _, _, err := store.RecordStepIntent(ctx, record.ID, intent); err != nil { t.Fatalf("record first step intent: %v", err) } _, _, err = store.RecordStepIntent(ctx, record.ID, StepIntent{ Key: intent.Key, Name: "different", Intent: intent.Intent, }) if !errors.Is(err, ErrStepConflict) { t.Fatalf("expected step conflict, got %v", err) } } // TestStoreSerializesConcurrentCreates 验证并发创建事务时单活动事务约束生效:多个 // 并发创建请求中恰好一个成功,其余均以 ErrActiveExists 拒绝。 func TestStoreSerializesConcurrentCreates(t *testing.T) { t.Parallel() ctx := context.Background() store := openTestStore(t) const workers = 12 var wait sync.WaitGroup wait.Add(workers) results := make(chan error, workers) for index := 0; index < workers; index++ { go func(index int) { defer wait.Done() _, _, err := store.CreateTransaction(ctx, CreateRequest{ IdempotencyKey: "concurrent-" + string(rune('A'+index)), Source: "test", Service: "backend", }) results <- err }(index) } wait.Wait() close(results) var created, rejected int for err := range results { switch { case err == nil: created++ case errors.Is(err, ErrActiveExists): rejected++ default: t.Fatalf("unexpected create error: %v", err) } } if created != 1 || rejected != workers-1 { t.Fatalf("unexpected concurrent result: created=%d rejected=%d", created, rejected) } } // openTestStore 在临时目录打开一个测试用 Store,并通过 t.Cleanup 确保测试结束时关闭。 // // 参数 t 用于报告错误与注册清理函数。打开失败时以 t.Fatalf 终止测试。 func openTestStore(t *testing.T) *Store { t.Helper() store, err := OpenStore(context.Background(), filepath.Join(t.TempDir(), "transaction.db")) if err != nil { t.Fatalf("open test store: %v", err) } t.Cleanup(func() { if err := store.Close(); err != nil { t.Errorf("close test store: %v", err) } }) return store }