fix(acceptance): 修复快照迁移重放与派发死锁

本地同构验收在全量迁移后导入生产快照，导致本次发布的数据迁移被旧能力覆盖；增加仅允许严格本地集群标记启用的导入后迁移重放，并新增幂等 Seedance 约束校准。\n\n批量异步派发改为在事务开始按全局顺序预锁全部任务与容量作用域，同时对 PostgreSQL 死锁和序列化失败做退避重试，避免双 API 重叠批次形成环形等待。\n\n验证：Go 全量测试、gofmt、bash -n、ShellCheck、迁移安全检查。

apps/api/cmd/migrate/main.go

@@ -6,6 +6,7 @@ import (
 	"log/slog"
 	"os"
 	"path/filepath"
+	"regexp"
 	"sort"
 	"strings"
 
@@ -14,10 +15,15 @@ import (
 )
 
 const (
-	noTransactionMigrationMarker = "-- easyai:migration:no-transaction"
-	migrationStatementSeparator  = "-- easyai:migration:statement"
+	noTransactionMigrationMarker      = "-- easyai:migration:no-transaction"
+	migrationStatementSeparator       = "-- easyai:migration:statement"
+	acceptanceImportReplayMarker      = "-- easyai:migration:reapply-after-acceptance-import"
+	acceptanceImportReplayEnvironment = "AI_GATEWAY_MIGRATION_REAPPLY_ACCEPTANCE_IMPORT"
+	acceptanceLocalClusterSettingKey  = "acceptance_local_cluster_id"
 )
 
+var acceptanceLocalClusterMarkerPattern = regexp.MustCompile(`^easyai-local-[0-9a-f]{24}$`)
+
 func main() {
 	cfg := config.Load()
 	logger := slog.New(slog.NewTextHandler(os.Stdout, nil))
@@ -42,6 +48,21 @@ CREATE TABLE IF NOT EXISTS schema_migrations (
 		logger.Error("ensure schema_migrations failed", "error", err)
 		os.Exit(1)
 	}
+	reapplyAcceptanceImport := strings.EqualFold(strings.TrimSpace(os.Getenv(acceptanceImportReplayEnvironment)), "true")
+	if reapplyAcceptanceImport {
+		var localClusterID string
+		if err := conn.QueryRow(ctx, `
+SELECT COALESCE(value->>'clusterId', '')
+FROM system_settings
+WHERE setting_key = $1`, acceptanceLocalClusterSettingKey).Scan(&localClusterID); err != nil {
+			logger.Error("verify local acceptance database failed", "error", err)
+			os.Exit(1)
+		}
+		if !acceptanceLocalClusterMarkerPattern.MatchString(localClusterID) {
+			logger.Error("refusing acceptance migration replay outside a marked local cluster")
+			os.Exit(1)
+		}
+	}
 
 	files, err := filepath.Glob("migrations/*.sql")
 	if err != nil {
@@ -57,16 +78,16 @@ CREATE TABLE IF NOT EXISTS schema_migrations (
 			logger.Error("check migration failed", "version", version, "error", err)
 			os.Exit(1)
 		}
-		if exists {
-			logger.Info("migration skipped", "version", version)
-			continue
-		}
-
 		sqlBytes, err := os.ReadFile(file)
 		if err != nil {
 			logger.Error("read migration file failed", "file", file, "error", err)
 			os.Exit(1)
 		}
+		replaying := exists && reapplyAcceptanceImport && hasMigrationMarker(string(sqlBytes), acceptanceImportReplayMarker)
+		if exists && !replaying {
+			logger.Info("migration skipped", "version", version)
+			continue
+		}
 
 		noTransaction, statements := migrationStatements(string(sqlBytes))
 		if noTransaction {
@@ -76,11 +97,17 @@ CREATE TABLE IF NOT EXISTS schema_migrations (
 					os.Exit(1)
 				}
 			}
-			if _, err := conn.Exec(ctx, "INSERT INTO schema_migrations(version) VALUES($1)", version); err != nil {
-				logger.Error("record non-transaction migration failed", "version", version, "error", err)
-				os.Exit(1)
+			if !exists {
+				if _, err := conn.Exec(ctx, "INSERT INTO schema_migrations(version) VALUES($1)", version); err != nil {
+					logger.Error("record non-transaction migration failed", "version", version, "error", err)
+					os.Exit(1)
+				}
+			}
+			if replaying {
+				logger.Info("acceptance migration replayed", "version", version, "transactional", false)
+			} else {
+				logger.Info("migration applied", "version", version, "transactional", false)
 			}
-			logger.Info("migration applied", "version", version, "transactional", false)
 			continue
 		}
 
@@ -102,21 +129,36 @@ CREATE TABLE IF NOT EXISTS schema_migrations (
 			logger.Error("execute migration failed", "version", version, "error", err)
 			os.Exit(1)
 		}
-		if _, err := tx.Exec(ctx, "INSERT INTO schema_migrations(version) VALUES($1)", version); err != nil {
-			_ = tx.Rollback(ctx)
-			logger.Error("record migration failed", "version", version, "error", err)
-			os.Exit(1)
+		if !exists {
+			if _, err := tx.Exec(ctx, "INSERT INTO schema_migrations(version) VALUES($1)", version); err != nil {
+				_ = tx.Rollback(ctx)
+				logger.Error("record migration failed", "version", version, "error", err)
+				os.Exit(1)
+			}
 		}
 		if err := tx.Commit(ctx); err != nil {
 			logger.Error("commit migration failed", "version", version, "error", err)
 			os.Exit(1)
 		}
-		logger.Info("migration applied", "version", version)
+		if replaying {
+			logger.Info("acceptance migration replayed", "version", version)
+		} else {
+			logger.Info("migration applied", "version", version)
+		}
 	}
 
 	fmt.Println("migrations complete")
 }
 
+func hasMigrationMarker(sql string, marker string) bool {
+	for _, line := range strings.Split(sql, "\n") {
+		if strings.TrimSpace(line) == marker {
+			return true
+		}
+	}
+	return false
+}
+
 func migrationStatements(sql string) (bool, []string) {
 	trimmed := strings.TrimSpace(sql)
 	if !strings.HasPrefix(trimmed, noTransactionMigrationMarker) {

apps/api/cmd/migrate/main_test.go

@@ -32,6 +32,21 @@ CREATE INDEX CONCURRENTLY IF NOT EXISTS second_index ON second_table(id);
 	}
 }
 
+func TestAcceptanceImportReplayMarkerRequiresAnExactCommentLine(t *testing.T) {
+	if !hasMigrationMarker(
+		"-- preface\n"+acceptanceImportReplayMarker+"\nSELECT 1;\n",
+		acceptanceImportReplayMarker,
+	) {
+		t.Fatal("expected exact acceptance import replay marker")
+	}
+	if hasMigrationMarker(
+		"-- mentions "+acceptanceImportReplayMarker+" in prose\nSELECT 1;\n",
+		acceptanceImportReplayMarker,
+	) {
+		t.Fatal("prose mention must not enable acceptance import replay")
+	}
+}
+
 func TestSeedanceInputImageConstraintMigrationKeepsCatalogSnapshotsInSync(t *testing.T) {
 	payload, err := os.ReadFile("../../migrations/0078_seedance_input_image_constraints.sql")
 	if err != nil {
@@ -74,6 +89,27 @@ func TestVolcesSeedanceInputImageConstraintMigrationUsesDocumentedBounds(t *test
 	}
 }
 
+func TestVolcesSeedanceAcceptanceReconciliationIsReplayable(t *testing.T) {
+	payload, err := os.ReadFile("../../migrations/0099_reconcile_volces_seedance20_acceptance_constraints.sql")
+	if err != nil {
+		t.Fatal(err)
+	}
+	content := string(payload)
+	for _, required := range []string{
+		acceptanceImportReplayMarker,
+		"volces:doubao-seedance-2-0-260128",
+		"volces:doubao-seedance-2-0-fast-260128",
+		"volces:doubao-seedance-2-0-mini-260615",
+		`"long_edge":300`,
+		`"long_edge":6000`,
+		`'[0.4,2.5]'::jsonb`,
+	} {
+		if !strings.Contains(content, required) {
+			t.Fatalf("Volces acceptance reconciliation is missing %q", required)
+		}
+	}
+}
+
 func TestSecurityEventSchemaMigrationsDefineCurrentLifecycle(t *testing.T) {
 	streamPayload, err := os.ReadFile("../../migrations/0063_oidc_security_events.sql")
 	if err != nil {

apps/api/internal/store/admission_lock.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
 )
 
 const (
@@ -153,7 +154,7 @@ func retryAdmissionOperation[T any](
 		}
 		result, operationErr := operation()
 		release()
-		if !errors.Is(operationErr, errAdmissionLockBusy) {
+		if !isRetryableAdmissionTransactionError(operationErr) {
 			return result, operationErr
 		}
 		if err := waitAdmissionLockRetry(ctx, attempt); err != nil {
@@ -162,6 +163,17 @@ func retryAdmissionOperation[T any](
 	}
 }
 
+func isRetryableAdmissionTransactionError(err error) bool {
+	if errors.Is(err, errAdmissionLockBusy) {
+		return true
+	}
+	var postgresError *pgconn.PgError
+	if !errors.As(err, &postgresError) {
+		return false
+	}
+	return postgresError.Code == "40P01" || postgresError.Code == "40001"
+}
+
 func waitAdmissionLockRetry(ctx context.Context, attempt int) error {
 	delay := admissionLockRetryMin
 	for index := 0; index < attempt && delay < admissionLockRetryMax; index++ {

apps/api/internal/store/admission_lock_test.go

@@ -11,8 +11,23 @@ import (
 
 	"github.com/google/uuid"
 	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
 )
 
+func TestAdmissionTransactionRetryClassification(t *testing.T) {
+	for _, code := range []string{"40P01", "40001"} {
+		if !isRetryableAdmissionTransactionError(&pgconn.PgError{Code: code}) {
+			t.Fatalf("PostgreSQL error %s should be retried", code)
+		}
+	}
+	if !isRetryableAdmissionTransactionError(errAdmissionLockBusy) {
+		t.Fatal("admission lock timeout should be retried")
+	}
+	if isRetryableAdmissionTransactionError(&pgconn.PgError{Code: "23505"}) {
+		t.Fatal("unique violations must not be retried")
+	}
+}
+
 func TestAdmissionLocalLockSetSerializesSharedKeys(t *testing.T) {
 	lockSet := newAdmissionLocalLockSet()
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)

apps/api/internal/store/admission_queue.go

@@ -474,12 +474,22 @@ func (s *Store) TryTaskAdmissionBatchWithAdmittedHook(
 		}
 		lockKeys = append(lockKeys, admissionOperationLockKeys(input)...)
 	}
+	lockKeys = normalizedAdmissionLockKeys(lockKeys)
 	return retryAdmissionOperation(ctx, lockKeys, func() ([]TaskAdmissionBatchOutcome, error) {
 		tx, err := s.pool.Begin(ctx)
 		if err != nil {
 			return nil, err
 		}
 		defer rollbackTransaction(tx)
+		// A batch touches one shared capacity scope and multiple task keys. Lock
+		// the complete union in one global order before processing any task. If
+		// two API processes dispatch overlapping FIFO windows, neither can hold a
+		// scope while waiting on a task key already owned by the other batch.
+		for _, lockKey := range lockKeys {
+			if err := tryAdmissionTransactionLock(ctx, tx, lockKey); err != nil {
+				return nil, err
+			}
+		}
 
 		outcomes := make([]TaskAdmissionBatchOutcome, 0, len(inputs))
 		notify := false

apps/api/migrations/0099_reconcile_volces_seedance20_acceptance_constraints.sql

@@ -0,0 +1,96 @@
+-- easyai:migration:reapply-after-acceptance-import
+-- A production snapshot is imported after the local schema is migrated. Replay
+-- this idempotent catalog reconciliation so the local target state includes the
+-- data changes delivered by the release under test.
+WITH input_constraints AS (
+  SELECT jsonb_build_object(
+    'input_image_resolution_range',
+    '{"min":{"long_edge":300,"short_edge":300},"max":{"long_edge":6000,"short_edge":6000}}'::jsonb,
+    'input_image_aspect_ratio_range',
+    '[0.4,2.5]'::jsonb
+  ) AS value
+),
+target_base_models AS (
+  SELECT
+    base_model.id,
+    COALESCE(base_model.capabilities, '{}'::jsonb) || jsonb_build_object(
+      'image_to_video',
+      COALESCE(base_model.capabilities->'image_to_video', '{}'::jsonb) || input_constraints.value,
+      'omni_video',
+      COALESCE(base_model.capabilities->'omni_video', '{}'::jsonb) || input_constraints.value
+    ) AS image_capabilities
+  FROM base_model_catalog base_model
+  CROSS JOIN input_constraints
+  WHERE base_model.provider_key = 'volces'
+    AND (
+      base_model.canonical_model_key IN (
+        'volces:doubao-seedance-2-0-260128',
+        'volces:doubao-seedance-2-0-fast-260128',
+        'volces:doubao-seedance-2-0-mini-260615'
+      )
+      OR base_model.provider_model_name IN (
+        'doubao-seedance-2-0-260128',
+        'doubao-seedance-2-0-fast-260128',
+        'doubao-seedance-2-0-mini-260615'
+      )
+    )
+),
+updated_base_models AS (
+  UPDATE base_model_catalog base_model
+  SET capabilities = target.image_capabilities,
+      metadata = jsonb_set(
+        COALESCE(base_model.metadata, '{}'::jsonb) || jsonb_build_object(
+          'rawModel',
+          COALESCE(base_model.metadata->'rawModel', '{}'::jsonb)
+        ),
+        '{rawModel,capabilities}',
+        target.image_capabilities - 'originalTypes',
+        true
+      ),
+      default_snapshot = CASE
+        WHEN COALESCE(base_model.default_snapshot, '{}'::jsonb) = '{}'::jsonb
+          THEN base_model.default_snapshot
+        ELSE jsonb_set(
+          jsonb_set(
+            base_model.default_snapshot || jsonb_build_object(
+              'metadata',
+              COALESCE(base_model.default_snapshot->'metadata', '{}'::jsonb) || jsonb_build_object(
+                'rawModel',
+                COALESCE(base_model.default_snapshot#>'{metadata,rawModel}', '{}'::jsonb)
+              )
+            ),
+            '{capabilities}',
+            target.image_capabilities,
+            true
+          ),
+          '{metadata,rawModel,capabilities}',
+          target.image_capabilities - 'originalTypes',
+          true
+        )
+      END,
+      updated_at = now()
+  FROM target_base_models target
+  WHERE base_model.id = target.id
+  RETURNING base_model.id
+)
+UPDATE platform_models platform_model
+SET capabilities = COALESCE(platform_model.capabilities, '{}'::jsonb) || jsonb_build_object(
+      'image_to_video',
+      COALESCE(platform_model.capabilities->'image_to_video', '{}'::jsonb) || input_constraints.value,
+      'omni_video',
+      COALESCE(platform_model.capabilities->'omni_video', '{}'::jsonb) || input_constraints.value
+    ),
+    updated_at = now()
+FROM integration_platforms platform
+CROSS JOIN input_constraints
+WHERE platform_model.platform_id = platform.id
+  AND platform.deleted_at IS NULL
+  AND platform.provider = 'volces'
+  AND (
+    platform_model.base_model_id IN (SELECT id FROM updated_base_models)
+    OR COALESCE(NULLIF(platform_model.provider_model_name, ''), platform_model.model_name) IN (
+      'doubao-seedance-2-0-260128',
+      'doubao-seedance-2-0-fast-260128',
+      'doubao-seedance-2-0-mini-260615'
+    )
+  );