From 262090db7b9a4b2f8dbc8396016e7282883e1d86 Mon Sep 17 00:00:00 2001 From: wangbo Date: Fri, 31 Jul 2026 04:25:09 +0800 Subject: [PATCH] =?UTF-8?q?perf(acceptance):=20=E4=BD=BF=E7=94=A8=E5=A4=9A?= =?UTF-8?q?=E9=92=B1=E5=8C=85=E8=BA=AB=E4=BB=BD=E5=88=86=E7=89=87=E6=A8=A1?= =?UTF-8?q?=E6=8B=9F=E5=B9=B6=E5=8F=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 单一验收钱包会把预留和结算串行化，掩盖双节点 Worker 的真实吞吐。验收现在幂等准备 32 个隔离身份和钱包，按请求轮询 API Key，并在 Run 配置中登记允许的 Key/User 身份对；密钥仅经 stdin 传给集群内压测进程。\n\n仍保留真实账务、候选权限、回调、重复扣费和强杀恢复校验。\n\n验证：Go 全量测试、临时 PostgreSQL 集成测试、go vet、OpenAPI 生成、迁移安全检查、bash -n、ShellCheck。 --- apps/api/cmd/acceptance-load/main.go | 18 +- apps/api/cmd/acceptance-load/main_test.go | 19 +- apps/api/internal/store/acceptance.go | 24 ++- .../store/acceptance_integration_test.go | 11 + docs/operations/production-acceptance.md | 29 ++- scripts/cluster/run-production-acceptance.sh | 204 ++++++++++++++++-- 6 files changed, 263 insertions(+), 42 deletions(-) diff --git a/apps/api/cmd/acceptance-load/main.go b/apps/api/cmd/acceptance-load/main.go index 04dfd87..df2a4ba 100644 --- a/apps/api/cmd/acceptance-load/main.go +++ b/apps/api/cmd/acceptance-load/main.go @@ -39,7 +39,7 @@ type options struct { gateways []string gatewayTLSName string emulatorURL string - apiKey string + apiKeys []string runID string runToken string geminiModel string @@ -135,7 +135,7 @@ func parseOptions() (options, error) { gateways: splitCSV(os.Getenv("AI_GATEWAY_ACCEPTANCE_GATEWAYS")), gatewayTLSName: strings.TrimSpace(os.Getenv("AI_GATEWAY_ACCEPTANCE_GATEWAY_TLS_SERVER_NAME")), emulatorURL: strings.TrimRight(strings.TrimSpace(os.Getenv("AI_GATEWAY_ACCEPTANCE_EMULATOR_URL")), "/"), - apiKey: strings.TrimSpace(os.Getenv("AI_GATEWAY_ACCEPTANCE_API_KEY")), + apiKeys: splitCSV(env("AI_GATEWAY_ACCEPTANCE_API_KEYS", os.Getenv("AI_GATEWAY_ACCEPTANCE_API_KEY"))), runID: strings.TrimSpace(os.Getenv("AI_GATEWAY_ACCEPTANCE_RUN_ID")), runToken: strings.TrimSpace(os.Getenv("AI_GATEWAY_ACCEPTANCE_RUN_TOKEN")), geminiModel: strings.TrimSpace(os.Getenv("AI_GATEWAY_ACCEPTANCE_GEMINI_MODEL")), @@ -158,7 +158,7 @@ func parseOptions() (options, error) { (strings.ContainsAny(opts.gatewayTLSName, "/:@") || !strings.Contains(opts.gatewayTLSName, ".")) { return options{}, errors.New("AI_GATEWAY_ACCEPTANCE_GATEWAY_TLS_SERVER_NAME must be a DNS hostname") } - if opts.apiKey == "" || opts.runID == "" || opts.runToken == "" { + if len(opts.apiKeys) == 0 || opts.runID == "" || opts.runToken == "" { return options{}, errors.New("acceptance API key, Run ID, and Run Token environment variables are required") } if opts.geminiModel == "" || opts.videoModel == "" { @@ -289,7 +289,7 @@ func runGemini( endpoint := opts.gateways[index%len(opts.gateways)] + "/v1beta/models/" + url.PathEscape(opts.geminiModel) + ":generateContent" req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(requestBody)) if err == nil { - opts.setHeaders(req, realUpstream) + opts.setHeaders(req, index, realUpstream) req.Header.Set("Content-Type", "application/json") var response *http.Response response, err = client.Do(req) @@ -414,7 +414,7 @@ func runVideo( endpoint := opts.gateways[index%len(opts.gateways)] + "/api/v1/videos/generations" req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(body)) if err == nil { - opts.setHeaders(req, realUpstream) + opts.setHeaders(req, index, realUpstream) req.Header.Set("Content-Type", "application/json") req.Header.Set("X-Async", "true") var response *http.Response @@ -519,7 +519,7 @@ func pollVideoTask(ctx context.Context, client *http.Client, opts options, taskI if err != nil { return err } - opts.setHeaders(req, realUpstream) + opts.setHeaders(req, index, realUpstream) response, err := client.Do(req) if err == nil { if response.StatusCode != http.StatusOK { @@ -599,11 +599,11 @@ func validateVideoAsset(ctx context.Context, client *http.Client, mediaURL strin return nil } -func (o options) setHeaders(request *http.Request, realUpstream bool) { +func (o options) setHeaders(request *http.Request, requestIndex int, realUpstream bool) { if o.gatewayTLSName != "" { request.Host = o.gatewayTLSName } - request.Header.Set("Authorization", "Bearer "+o.apiKey) + request.Header.Set("Authorization", "Bearer "+o.apiKeys[requestIndex%len(o.apiKeys)]) request.Header.Set(runHeader, o.runID) request.Header.Set(tokenHeader, o.runToken) if realUpstream { @@ -799,7 +799,7 @@ func paddedPNG(size int) []byte { } func redactError(message string, opts options) string { - for _, secret := range []string{opts.apiKey, opts.runToken} { + for _, secret := range append(append([]string(nil), opts.apiKeys...), opts.runToken) { if secret != "" { message = strings.ReplaceAll(message, secret, "[REDACTED]") } diff --git a/apps/api/cmd/acceptance-load/main_test.go b/apps/api/cmd/acceptance-load/main_test.go index 90edcb9..10fed3d 100644 --- a/apps/api/cmd/acceptance-load/main_test.go +++ b/apps/api/cmd/acceptance-load/main_test.go @@ -53,9 +53,19 @@ func TestGeminiLoadIsSplitAcrossTwoGatewayAPIs(t *testing.T) { encoded := base64.StdEncoding.EncodeToString(output) var firstCalls atomic.Int64 var secondCalls atomic.Int64 + var firstKeyCalls atomic.Int64 + var secondKeyCalls atomic.Int64 newGateway := func(calls *atomic.Int64) *httptest.Server { return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { calls.Add(1) + switch r.Header.Get("Authorization") { + case "Bearer key-1": + firstKeyCalls.Add(1) + case "Bearer key-2": + secondKeyCalls.Add(1) + default: + t.Errorf("unexpected authorization header") + } if r.Header.Get(runHeader) != "run-1" || r.Header.Get(tokenHeader) != "token-1" { t.Errorf("missing acceptance headers") } @@ -78,7 +88,7 @@ func TestGeminiLoadIsSplitAcrossTwoGatewayAPIs(t *testing.T) { defer second.Close() opts := options{ - gateways: []string{first.URL, second.URL}, apiKey: "key-1", runID: "run-1", + gateways: []string{first.URL, second.URL}, apiKeys: []string{"key-1", "key-2"}, runID: "run-1", runToken: "token-1", geminiModel: "gemini-image-test", } result := runGemini(t.Context(), http.DefaultClient, opts, "dual-api", 8, 256<<10, 256<<10, false) @@ -88,6 +98,9 @@ func TestGeminiLoadIsSplitAcrossTwoGatewayAPIs(t *testing.T) { if firstCalls.Load() != 4 || secondCalls.Load() != 4 { t.Fatalf("gateway calls=%d/%d", firstCalls.Load(), secondCalls.Load()) } + if firstKeyCalls.Load() != 4 || secondKeyCalls.Load() != 4 { + t.Fatalf("API key calls=%d/%d", firstKeyCalls.Load(), secondKeyCalls.Load()) + } } func TestValidateVideoAssetDownloadsFinalMedia(t *testing.T) { @@ -117,10 +130,10 @@ func TestAcceptanceReportErrorRedactsSecretsAndURLs(t *testing.T) { func TestAcceptanceGatewayTLSNameSetsHostHeader(t *testing.T) { request := httptest.NewRequest(http.MethodGet, "https://127.0.0.1/api/v1/healthz", nil) opts := options{ - apiKey: "key-1", runID: "run-1", runToken: "token-1", + apiKeys: []string{"key-1"}, runID: "run-1", runToken: "token-1", gatewayTLSName: "ai.example.com", } - opts.setHeaders(request, false) + opts.setHeaders(request, 0, false) if request.Host != "ai.example.com" { t.Fatalf("Host=%q", request.Host) } diff --git a/apps/api/internal/store/acceptance.go b/apps/api/internal/store/acceptance.go index ab1bfab..5e608ed 100644 --- a/apps/api/internal/store/acceptance.go +++ b/apps/api/internal/store/acceptance.go @@ -583,17 +583,35 @@ func (s *Store) AuthorizeAcceptanceTask(ctx context.Context, runID string, token return "", ErrAcceptanceNotAuthorized } var activeRunID, apiKeyID, userID, tokenHash, status string + var participantAuthorized bool err = s.pool.QueryRow(ctx, ` -SELECT id::text, api_key_id, user_id, token_sha256, status +SELECT id::text, api_key_id, user_id, token_sha256, status, + ( + (api_key_id = $2 AND user_id = $3) + OR EXISTS ( + SELECT 1 + FROM jsonb_array_elements( + CASE + WHEN jsonb_typeof(config->'participants') = 'array' + THEN config->'participants' + ELSE '[]'::jsonb + END + ) participant + WHERE participant->>'apiKeyId' = $2 + AND participant->>'userId' = $3 + ) + ) FROM gateway_acceptance_runs -WHERE id = $1::uuid`, runID).Scan(&activeRunID, &apiKeyID, &userID, &tokenHash, &status) +WHERE id = $1::uuid`, runID, strings.TrimSpace(user.APIKeyID), strings.TrimSpace(user.ID)).Scan( + &activeRunID, &apiKeyID, &userID, &tokenHash, &status, &participantAuthorized, + ) if err != nil { if errors.Is(err, pgx.ErrNoRows) { return "", ErrAcceptanceNotAuthorized } return "", err } - if status != "running" || apiKeyID != strings.TrimSpace(user.APIKeyID) || userID != strings.TrimSpace(user.ID) || + if status != "running" || !participantAuthorized || subtle.ConstantTimeCompare([]byte(tokenHash), []byte(acceptanceTokenSHA256(token))) != 1 { return "", ErrAcceptanceNotAuthorized } diff --git a/apps/api/internal/store/acceptance_integration_test.go b/apps/api/internal/store/acceptance_integration_test.go index d7486a9..afd8f40 100644 --- a/apps/api/internal/store/acceptance_integration_test.go +++ b/apps/api/internal/store/acceptance_integration_test.go @@ -45,6 +45,9 @@ WHERE setting_key = $1`, SystemSettingGatewayTrafficMode) EmulatorBaseURL: "http://acceptance-emulator:8090", CallbackURL: "http://acceptance-emulator:8090/callbacks", CapacityProfile: "P24", + Config: map[string]any{"participants": []any{ + map[string]any{"apiKeyId": "acceptance-shard-key", "userId": "acceptance-shard-user"}, + }}, }) if err != nil { t.Fatalf("create acceptance run: %v", err) @@ -70,6 +73,14 @@ WHERE setting_key = $1`, SystemSettingGatewayTrafficMode) if authorizedRunID, err := db.AuthorizeAcceptanceTask(ctx, run.ID, token, user); err != nil || authorizedRunID != run.ID { t.Fatalf("authorize acceptance task run=%q err=%v", authorizedRunID, err) } + shardUser := &auth.User{ID: "acceptance-shard-user", APIKeyID: "acceptance-shard-key"} + if authorizedRunID, err := db.AuthorizeAcceptanceTask(ctx, run.ID, token, shardUser); err != nil || authorizedRunID != run.ID { + t.Fatalf("authorize acceptance shard run=%q err=%v", authorizedRunID, err) + } + unpairedShard := &auth.User{ID: "acceptance-shard-user", APIKeyID: "acceptance-api-key"} + if _, err := db.AuthorizeAcceptanceTask(ctx, run.ID, token, unpairedShard); !errors.Is(err, ErrAcceptanceNotAuthorized) { + t.Fatalf("unpaired acceptance shard error=%v, want unauthorized", err) + } if baseURL, credential, err := db.AcceptanceCandidateOverride(ctx, run.ID); err != nil || baseURL != "http://acceptance-emulator:8090" || credential == "" || credential == token { t.Fatalf("candidate override base=%q credential=%q err=%v", baseURL, credential, err) diff --git a/docs/operations/production-acceptance.md b/docs/operations/production-acceptance.md index fe66a68..179abf8 100644 --- a/docs/operations/production-acceptance.md +++ b/docs/operations/production-acceptance.md @@ -16,7 +16,8 @@ Gateway 流量模式保存在数据库中，支持 `live` 和 `validation`： - `live`：正常接收正式任务，带验收 Header 的请求被拒绝。 - `validation`：普通新任务返回 `503 validation_in_progress`；已有正式任务继续排空。 -- 验收请求必须同时匹配当前 Run ID、随机 Run Token、专属 API Key ID 和专属用户 ID。 +- 验收请求必须同时匹配当前 Run ID、随机 Run Token，以及 Run 中登记的专属 API Key/User + 身份对；任意错配都被拒绝。 - 普通生产请求体不能打开 `simulation`。 - `X-EasyAI-Acceptance-Upstream: real` 只对已授权验收身份有效，用于最后两条真实小流量请求。 @@ -57,6 +58,12 @@ WebP 和 4K 图片。128 组输入组合避免只命中相同缓存；每四个 `/contents/generations/tasks` 提交、查询、取消协议，还提供格式混合的图片池和独立回调 收集器。它校验参考图数量、角色、格式、尺寸与内容哈希，并统计重复上游提交和重复回调。 +默认通过 `AI_GATEWAY_ACCEPTANCE_IDENTITY_SHARDS=32` 幂等准备 32 个仅属于验收用户组的 +身份、API Key 和独立钱包，并按请求序号轮询使用。这样可模拟真实多租户并发，避免把 1000 +个任务全部压在同一钱包行锁上；每个任务仍执行真实预留、结算和重复扣费检查。该值可在 +`1..128` 内通过环境变量调整，Run 只保存 Key/User ID 对，Key Secret 仅通过 stdin 传入 +集群内压测进程，不进入 Pod spec、数据库 Run 配置、报告或日志。 + ## Worker 配置 执行槽、连接池和媒体并发进入 Pod 环境变量，副本数由发布工具更新 Deployment： @@ -101,9 +108,10 @@ easyai-ai-gateway-cluster-release capacity ## 执行 -先把专属验收用户、钱包、API Key、管理员 Token、两个 Gateway 地址和三张真实小流量 -参考图写入本地私有环境。模型变量可省略，脚本会从当前启用候选中选择 Gemini 图片编辑 -模型，以及 `omni_video.max_images >= 9` 的视频模型并优先 Seedance 2.0/fast。 +先把主验收用户、钱包、API Key、管理员 Token、两个 Gateway 地址和三张真实小流量参考图 +写入本地私有环境。脚本会以主身份为模板幂等准备其余隔离身份；模型变量可省略，脚本会从 +当前启用候选中选择 Gemini 图片编辑模型，以及 `omni_video.max_images >= 9` 的视频模型 +并优先 Seedance 2.0/fast。 为保证请求严格各有一半命中宁波和香港，可以把两个 Gateway URL 配成两个节点的 `https://<节点IP>`，并设置 @@ -118,13 +126,14 @@ scripts/cluster/run-production-acceptance.sh \ 脚本依次执行： 1. 校验工作区、manifest、线上 release SHA、digest、双站点镜像和副本数。 -2. 部署 digest 固定且仅集群内可访问的协议模拟器。 -3. 创建 Run、切换 `validation`、等待已有正式任务排空。 -4. 按 P24、P28、P32 执行三轮负载和 Worker 强杀。 -5. 执行两条真实小流量请求。 -6. 校验任务、账务、回调、队列、连接池、RSS、节点内存、数据库同步、六向 WireGuard 和 +2. 准备专属验收用户组、身份分片、独立钱包与候选访问规则。 +3. 部署 digest 固定且仅集群内可访问的协议模拟器。 +4. 创建 Run、切换 `validation`、等待已有正式任务排空。 +5. 按 P24、P28、P32 执行三轮负载和 Worker 强杀。 +6. 执行两条真实小流量请求。 +7. 校验任务、账务、回调、队列、连接池、RSS、节点内存、数据库同步、六向 WireGuard 和 公网健康。 -7. 重新校验 SHA/digest/revision 后切回 `live`。 +8. 重新校验 SHA/digest/revision 后切回 `live`。 报告保存在 `dist/acceptance//`，包含每阶段吞吐、p50/p95/p99、队列和资源 CSV、 协议模拟器统计及最终摘要。报告不包含图片原文、Base64、Token、密码或连接串。 diff --git a/scripts/cluster/run-production-acceptance.sh b/scripts/cluster/run-production-acceptance.sh index 82c1a98..74ff968 100755 --- a/scripts/cluster/run-production-acceptance.sh +++ b/scripts/cluster/run-production-acceptance.sh @@ -28,6 +28,7 @@ Optional model overrides (otherwise selected from current production candidates) AI_GATEWAY_ACCEPTANCE_GEMINI_MODEL AI_GATEWAY_ACCEPTANCE_VIDEO_MODEL AI_GATEWAY_ACCEPTANCE_GATEWAY_TLS_SERVER_NAME + AI_GATEWAY_ACCEPTANCE_IDENTITY_SHARDS EOF } @@ -52,6 +53,7 @@ require_commands curl git jq node openssl sed : "${AI_GATEWAY_ACCEPTANCE_GEMINI_MODEL:=}" : "${AI_GATEWAY_ACCEPTANCE_VIDEO_MODEL:=}" : "${AI_GATEWAY_ACCEPTANCE_API_DATABASE_MAX_CONNS:=64}" +: "${AI_GATEWAY_ACCEPTANCE_IDENTITY_SHARDS:=32}" : "${AI_GATEWAY_ACCEPTANCE_GATEWAY_TLS_SERVER_NAME:=}" : "${AI_GATEWAY_ACCEPTANCE_GATEWAYS:?}" : "${AI_GATEWAY_ACCEPTANCE_REAL_IMAGE_URLS:?}" @@ -60,6 +62,11 @@ require_commands curl git jq node openssl sed echo 'AI_GATEWAY_ACCEPTANCE_API_DATABASE_MAX_CONNS must be between 1 and 256' >&2 exit 1 } +[[ $AI_GATEWAY_ACCEPTANCE_IDENTITY_SHARDS =~ ^[1-9][0-9]*$ && + $AI_GATEWAY_ACCEPTANCE_IDENTITY_SHARDS -le 128 ]] || { + echo 'AI_GATEWAY_ACCEPTANCE_IDENTITY_SHARDS must be between 1 and 128' >&2 + exit 1 +} node "$cluster_root/scripts/release-manifest.mjs" validate "$release_manifest" >/dev/null release_sha=$(node "$cluster_root/scripts/release-manifest.mjs" get "$release_manifest" sourceSha) @@ -102,6 +109,9 @@ active_profile=P24 failure_reason= failure_recorded=false acceptance_admin_base= +acceptance_group_id= +acceptance_participants_json='[]' +AI_GATEWAY_ACCEPTANCE_API_KEYS=$AI_GATEWAY_ACCEPTANCE_API_KEY cleanup() { local status=$? @@ -306,12 +316,15 @@ SELECT (SELECT count(*) FROM gateway_users WHERE default_user_group_id='$group_id'::uuid - AND id <> '$AI_GATEWAY_ACCEPTANCE_USER_ID'::uuid) + AND id <> '$AI_GATEWAY_ACCEPTANCE_USER_ID'::uuid + AND COALESCE(metadata->>'purpose','') <> 'production_acceptance_shard') ||':'|| (SELECT count(*) - FROM gateway_api_keys - WHERE user_group_id='$group_id'::uuid - AND gateway_user_id <> '$AI_GATEWAY_ACCEPTANCE_USER_ID'::uuid);") + FROM gateway_api_keys key + JOIN gateway_users user_record ON user_record.id=key.gateway_user_id + WHERE key.user_group_id='$group_id'::uuid + AND key.gateway_user_id <> '$AI_GATEWAY_ACCEPTANCE_USER_ID'::uuid + AND COALESCE(user_record.metadata->>'purpose','') <> 'production_acceptance_shard');") [[ $isolation_state == 0:0 ]] || { echo 'acceptance user group is referenced by non-acceptance identities' >&2 return 1 @@ -323,6 +336,7 @@ SELECT echo 'acceptance user group API returned an invalid group ID' >&2 return 1 } + acceptance_group_id=$group_id admin_request GET /api/admin/users '' "$users_response" user_body=$(jq -c \ @@ -392,12 +406,15 @@ SELECT (SELECT count(*) FROM gateway_users WHERE default_user_group_id='$group_id'::uuid - AND id <> '$AI_GATEWAY_ACCEPTANCE_USER_ID'::uuid) + AND id <> '$AI_GATEWAY_ACCEPTANCE_USER_ID'::uuid + AND COALESCE(metadata->>'purpose','') <> 'production_acceptance_shard') ||':'|| (SELECT count(*) - FROM gateway_api_keys - WHERE user_group_id='$group_id'::uuid - AND gateway_user_id <> '$AI_GATEWAY_ACCEPTANCE_USER_ID'::uuid);") + FROM gateway_api_keys key + JOIN gateway_users user_record ON user_record.id=key.gateway_user_id + WHERE key.user_group_id='$group_id'::uuid + AND key.gateway_user_id <> '$AI_GATEWAY_ACCEPTANCE_USER_ID'::uuid + AND COALESCE(user_record.metadata->>'purpose','') <> 'production_acceptance_shard');") [[ $isolation_state == 1:1:1:1:0:0 ]] || { echo 'dedicated acceptance identity isolation verification failed' >&2 return 1 @@ -405,6 +422,154 @@ SELECT echo 'acceptance_user_group=PASS rate_limit=production_candidates active_keys=1 isolated=true' } +ensure_acceptance_identity_shards() { + local identity_material key_count participant_count + [[ $acceptance_group_id =~ ^[0-9a-f-]{36}$ ]] || { + echo 'acceptance user group must be prepared before identity shards' >&2 + return 1 + } + identity_material=$(database_query " +WITH desired AS ( + SELECT ordinal, + 'production-acceptance-shard-' || lpad(ordinal::text, 3, '0') AS user_key + FROM generate_series(1, $((AI_GATEWAY_ACCEPTANCE_IDENTITY_SHARDS - 1))) ordinal +), shard_users AS ( + INSERT INTO gateway_users ( + user_key, source, username, display_name, default_user_group_id, + roles, auth_profile, metadata, status + ) + SELECT desired.user_key, + 'gateway', + desired.user_key, + 'Production Acceptance Shard ' || lpad(desired.ordinal::text, 3, '0'), + '$acceptance_group_id'::uuid, + '[]'::jsonb, + '{}'::jsonb, + jsonb_build_object( + 'purpose', 'production_acceptance_shard', + 'ordinal', desired.ordinal, + 'isolated', true + ), + 'active' + FROM desired + ON CONFLICT (user_key) DO UPDATE + SET default_user_group_id=EXCLUDED.default_user_group_id, + metadata=EXCLUDED.metadata, + status='active', + deleted_at=NULL, + updated_at=now() + RETURNING id, user_key +), primary_key AS ( + SELECT id, gateway_user_id, key_secret, scopes, rate_limit_policy, quota_policy + FROM gateway_api_keys + WHERE id='$AI_GATEWAY_ACCEPTANCE_API_KEY_ID'::uuid + AND gateway_user_id='$AI_GATEWAY_ACCEPTANCE_USER_ID'::uuid + AND status='active' + AND deleted_at IS NULL + AND COALESCE(key_secret, '') <> '' +), existing_keys AS ( + SELECT selected.id, selected.gateway_user_id, selected.key_secret + FROM shard_users shard + JOIN LATERAL ( + SELECT key.id, key.gateway_user_id, key.key_secret + FROM gateway_api_keys key + WHERE key.gateway_user_id=shard.id + AND key.name='Production Acceptance Shard' + AND key.status='active' + AND key.deleted_at IS NULL + AND COALESCE(key.key_secret, '') <> '' + ORDER BY key.created_at + LIMIT 1 + ) selected ON true +), new_key_material AS ( + SELECT shard.id AS gateway_user_id, + 'sk-gw-' || encode(gen_random_bytes(32), 'hex') AS secret + FROM shard_users shard + WHERE NOT EXISTS ( + SELECT 1 FROM existing_keys existing + WHERE existing.gateway_user_id=shard.id + ) +), inserted_keys AS ( + INSERT INTO gateway_api_keys ( + gateway_user_id, user_id, key_prefix, key_secret, key_hash, name, scopes, + user_group_id, rate_limit_policy, quota_policy, status + ) + SELECT material.gateway_user_id, + material.gateway_user_id::text, + left(material.secret, 18), + material.secret, + crypt(material.secret, gen_salt('bf', 10)), + 'Production Acceptance Shard', + template.scopes, + '$acceptance_group_id'::uuid, + template.rate_limit_policy, + template.quota_policy, + 'active' + FROM new_key_material material + CROSS JOIN primary_key template + RETURNING id, gateway_user_id, key_secret +), shard_keys AS ( + SELECT * FROM existing_keys + UNION ALL + SELECT * FROM inserted_keys +), wallets AS ( + INSERT INTO gateway_wallet_accounts ( + gateway_user_id, user_id, currency, balance, total_recharged, metadata, status + ) + SELECT shard.id, + shard.id::text, + 'resource', + 1000000000::numeric, + 1000000000::numeric, + '{\"purpose\":\"production_acceptance_shard\",\"isolated\":true}'::jsonb, + 'active' + FROM shard_users shard + ON CONFLICT (gateway_user_id, currency) DO UPDATE + SET balance=GREATEST(gateway_wallet_accounts.balance, EXCLUDED.balance), + total_recharged=GREATEST(gateway_wallet_accounts.total_recharged, EXCLUDED.total_recharged), + metadata=EXCLUDED.metadata, + status='active', + updated_at=now() + RETURNING gateway_user_id +), participants AS ( + SELECT 0 AS ordinal, key.id, key.gateway_user_id, key.key_secret + FROM primary_key key + UNION ALL + SELECT row_number() OVER (ORDER BY shard.user_key)::int, + key.id, + key.gateway_user_id, + key.key_secret + FROM shard_keys key + JOIN shard_users shard ON shard.id=key.gateway_user_id +) +SELECT jsonb_agg( + jsonb_build_object( + 'apiKeyId', participant.id, + 'userId', participant.gateway_user_id + ) + ORDER BY participant.ordinal + )::text + || E'\\t' || + string_agg(participant.key_secret, ',' ORDER BY participant.ordinal) +FROM participants participant +CROSS JOIN (SELECT count(*) FROM wallets) wallet_write_barrier;") + [[ $identity_material == *$'\t'* ]] || { + echo 'acceptance identity shard provisioning returned an invalid payload' >&2 + return 1 + } + acceptance_participants_json=${identity_material%%$'\t'*} + AI_GATEWAY_ACCEPTANCE_API_KEYS=${identity_material#*$'\t'} + participant_count=$(jq 'length' <<<"$acceptance_participants_json") + key_count=$(awk -F',' '{print NF}' <<<"$AI_GATEWAY_ACCEPTANCE_API_KEYS") + [[ $participant_count == "$AI_GATEWAY_ACCEPTANCE_IDENTITY_SHARDS" && + $key_count == "$AI_GATEWAY_ACCEPTANCE_IDENTITY_SHARDS" ]] || { + echo 'acceptance identity shard count verification failed' >&2 + return 1 + } + export AI_GATEWAY_ACCEPTANCE_API_KEYS + echo "acceptance_identity_shards=PASS count=$participant_count isolated_wallets=true" +} + select_acceptance_models() { if [[ -z $AI_GATEWAY_ACCEPTANCE_GEMINI_MODEL ]]; then AI_GATEWAY_ACCEPTANCE_GEMINI_MODEL=$(database_query " @@ -510,10 +675,10 @@ FROM resources;") return 1 } body=$(jq -cn \ - --arg subjectID "$AI_GATEWAY_ACCEPTANCE_API_KEY_ID" \ + --arg subjectID "$acceptance_group_id" \ --argjson resources "$resources" \ '{ - subjectType:"api_key", + subjectType:"user_group", subjectId:$subjectID, effect:"allow", upsertResources:$resources, @@ -549,15 +714,15 @@ FROM resources resource WHERE NOT EXISTS ( SELECT 1 FROM gateway_access_rules rule - WHERE rule.subject_type='api_key' - AND rule.subject_id='$AI_GATEWAY_ACCEPTANCE_API_KEY_ID'::uuid + WHERE rule.subject_type='user_group' + AND rule.subject_id='$acceptance_group_id'::uuid AND rule.effect='allow' AND rule.status='active' AND rule.resource_type=resource.resource_type AND rule.resource_id=resource.resource_id );") [[ $missing == 0 ]] || { - echo "acceptance API Key is missing $missing selected candidate access rules" >&2 + echo "acceptance user group is missing $missing selected candidate access rules" >&2 return 1 } echo "acceptance_model_access=PASS resources=$expected" @@ -587,6 +752,7 @@ create_and_activate_run() { --arg token "$run_token" \ --arg emulatorURL 'http://easyai-acceptance-emulator.easyai.svc.cluster.local:8090' \ --arg callbackURL 'http://easyai-acceptance-emulator.easyai.svc.cluster.local:8090/callbacks' \ + --argjson participants "$acceptance_participants_json" \ '{ releaseSha: $releaseSha, apiImageDigest: $apiDigest, @@ -597,7 +763,10 @@ create_and_activate_run() { emulatorBaseUrl: $emulatorURL, callbackUrl: $callbackURL, capacityProfile: "P24", - config: {workloads: ["gemini_image_edit", "multi_reference_video"]} + config: { + workloads: ["gemini_image_edit", "multi_reference_video"], + participants: $participants + } }') admin_request POST /api/admin/system/acceptance/runs "$body" "$create_response" run_id=$(jq -r '.id // empty' "$create_response") @@ -746,7 +915,7 @@ set -eu read -r gateways_b64 read -r tls_name_b64 read -r emulator_b64 -read -r api_key_b64 +read -r api_keys_b64 read -r run_id_b64 read -r run_token_b64 read -r gemini_model_b64 @@ -758,7 +927,7 @@ decode() { export AI_GATEWAY_ACCEPTANCE_GATEWAYS=$(decode "$gateways_b64") export AI_GATEWAY_ACCEPTANCE_GATEWAY_TLS_SERVER_NAME=$(decode "$tls_name_b64") export AI_GATEWAY_ACCEPTANCE_EMULATOR_URL=$(decode "$emulator_b64") -export AI_GATEWAY_ACCEPTANCE_API_KEY=$(decode "$api_key_b64") +export AI_GATEWAY_ACCEPTANCE_API_KEYS=$(decode "$api_keys_b64") export AI_GATEWAY_ACCEPTANCE_RUN_ID=$(decode "$run_id_b64") export AI_GATEWAY_ACCEPTANCE_RUN_TOKEN=$(decode "$run_token_b64") export AI_GATEWAY_ACCEPTANCE_GEMINI_MODEL=$(decode "$gemini_model_b64") @@ -779,7 +948,7 @@ exit "$status" printf '%s' 'http://easyai-acceptance-emulator.easyai.svc.cluster.local:8090' | openssl base64 -A printf '\n' - printf '%s' "$AI_GATEWAY_ACCEPTANCE_API_KEY" | openssl base64 -A + printf '%s' "$AI_GATEWAY_ACCEPTANCE_API_KEYS" | openssl base64 -A printf '\n' printf '%s' "$run_id" | openssl base64 -A printf '\n' @@ -1359,6 +1528,7 @@ finish_and_promote() { verify_release_cas ensure_acceptance_user_group +ensure_acceptance_identity_shards select_acceptance_models ensure_acceptance_model_access deploy_protocol_emulator