fix: 对齐规格解析跨端超时预算 (#257)
This commit is contained in:
@@ -26,6 +26,7 @@ const (
|
|||||||
MaxPurchaseSpecResolutionBodyBytes = 64 << 10
|
MaxPurchaseSpecResolutionBodyBytes = 64 << 10
|
||||||
PurchaseSpecRulesVersion = "purchase_rules_v1"
|
PurchaseSpecRulesVersion = "purchase_rules_v1"
|
||||||
purchaseSpecSchemaVersion = 1
|
purchaseSpecSchemaVersion = 1
|
||||||
|
purchaseSpecResolutionAITimeout = 30 * time.Second
|
||||||
)
|
)
|
||||||
|
|
||||||
var (
|
var (
|
||||||
@@ -143,7 +144,7 @@ func resolvePurchaseSpecWithSnapshotLoader(ctx context.Context, db *sql.DB, secr
|
|||||||
// 可以安全接管;最终 UPDATE 仍只允许 pending 完成一次。
|
// 可以安全接管;最终 UPDATE 仍只允许 pending 完成一次。
|
||||||
decision, terminal := matchPurchaseSpecByRule(req.TargetSize, req.Candidates)
|
decision, terminal := matchPurchaseSpecByRule(req.TargetSize, req.Candidates)
|
||||||
if !terminal {
|
if !terminal {
|
||||||
decision = matchPurchaseSpecWithAI(ctx, db, secrets, policy, req, loadSnapshot)
|
decision = matchPurchaseSpecWithRuntimeAI(ctx, db, secrets, policy, req, loadSnapshot)
|
||||||
}
|
}
|
||||||
if purchaseSpecDiagnosticSource(db, req.PddGoodsID) == "shopee_backfill" {
|
if purchaseSpecDiagnosticSource(db, req.PddGoodsID) == "shopee_backfill" {
|
||||||
decision.Reason = truncateRunes(decision.Reason+";诊断来源:shopee_backfill", 500)
|
decision.Reason = truncateRunes(decision.Reason+";诊断来源:shopee_backfill", 500)
|
||||||
@@ -151,6 +152,17 @@ func resolvePurchaseSpecWithSnapshotLoader(ctx context.Context, db *sql.DB, secr
|
|||||||
return completePurchaseSpecResolution(db, idempotencyKey, requestHash, resolution, req.Candidates, decision)
|
return completePurchaseSpecResolution(db, idempotencyKey, requestHash, resolution, req.Candidates, decision)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// matchPurchaseSpecWithRuntimeAI 给采购运行时 AI 单独设置 30 秒上限。
|
||||||
|
// Client 会等待最多 60 秒,因此 Admin 有足够时间保存安全结论并返回;如果请求更早取消,
|
||||||
|
// 子上下文也会立即取消,不会让模型调用脱离原 HTTP 请求继续运行。
|
||||||
|
func matchPurchaseSpecWithRuntimeAI(ctx context.Context, db *sql.DB, secrets AISecretStore,
|
||||||
|
policy AIEndpointPolicy, req PurchaseSpecResolutionRequest,
|
||||||
|
loadSnapshot aiSnapshotLoader) purchaseSpecDecision {
|
||||||
|
aiContext, cancel := context.WithTimeout(ctx, purchaseSpecResolutionAITimeout)
|
||||||
|
defer cancel()
|
||||||
|
return matchPurchaseSpecWithAI(aiContext, db, secrets, policy, req, loadSnapshot)
|
||||||
|
}
|
||||||
|
|
||||||
func parsePurchaseSpecResolutionRequest(taskID, idempotencyKey string, rawBody []byte) (PurchaseSpecResolutionRequest, error) {
|
func parsePurchaseSpecResolutionRequest(taskID, idempotencyKey string, rawBody []byte) (PurchaseSpecResolutionRequest, error) {
|
||||||
var req PurchaseSpecResolutionRequest
|
var req PurchaseSpecResolutionRequest
|
||||||
if len(rawBody) == 0 || len(rawBody) > MaxPurchaseSpecResolutionBodyBytes || !utf8.Valid(rawBody) {
|
if len(rawBody) == 0 || len(rawBody) > MaxPurchaseSpecResolutionBodyBytes || !utf8.Valid(rawBody) {
|
||||||
|
|||||||
@@ -99,6 +99,23 @@ type stubAIModelClient struct {
|
|||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type runtimeDeadlineAIClient struct {
|
||||||
|
deadline time.Time
|
||||||
|
deadlineOK bool
|
||||||
|
contextErr error
|
||||||
|
}
|
||||||
|
|
||||||
|
func (client *runtimeDeadlineAIClient) Match(ctx context.Context, _ model.AIProviderConfig, _ string,
|
||||||
|
_ AIModelMatchRequest) (AIModelMatchResponse, error) {
|
||||||
|
client.deadline, client.deadlineOK = ctx.Deadline()
|
||||||
|
client.contextErr = ctx.Err()
|
||||||
|
if client.contextErr != nil {
|
||||||
|
return AIModelMatchResponse{}, client.contextErr
|
||||||
|
}
|
||||||
|
return AIModelMatchResponse{Conclusion: "uncertain", ConfidenceBPS: 7000,
|
||||||
|
Reason: "测试信息不足"}, nil
|
||||||
|
}
|
||||||
|
|
||||||
func (client *stubAIModelClient) Match(ctx context.Context, _ model.AIProviderConfig, _ string,
|
func (client *stubAIModelClient) Match(ctx context.Context, _ model.AIProviderConfig, _ string,
|
||||||
request AIModelMatchRequest) (AIModelMatchResponse, error) {
|
request AIModelMatchRequest) (AIModelMatchResponse, error) {
|
||||||
client.calls.Add(1)
|
client.calls.Add(1)
|
||||||
@@ -115,6 +132,46 @@ func (client *stubAIModelClient) Match(ctx context.Context, _ model.AIProviderCo
|
|||||||
return client.response, client.err
|
return client.response, client.err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestMatchPurchaseSpecWithRuntimeAI_限制30秒并继承更早取消(t *testing.T) {
|
||||||
|
req, _, _ := validRuntimeSpecRequest(t, "cg257-timeout", "L码", "M码", "XL码")
|
||||||
|
client := &runtimeDeadlineAIClient{}
|
||||||
|
loader := func(context.Context, *sql.DB, AISecretStore, AIEndpointPolicy) (AIMatchSnapshot, error) {
|
||||||
|
return AIMatchSnapshot{Provider: model.AIProviderConfig{
|
||||||
|
ProviderID: "provider-timeout", Model: "model-timeout", TimeoutSeconds: 120,
|
||||||
|
ConfidenceThresholdBPS: 9000,
|
||||||
|
}, ConfigFingerprint: "fingerprint", Secret: "not-a-real-secret", Client: client}, nil
|
||||||
|
}
|
||||||
|
startedAt := time.Now()
|
||||||
|
decision := matchPurchaseSpecWithRuntimeAI(context.Background(), nil, nil,
|
||||||
|
AIEndpointPolicy{}, req, loader)
|
||||||
|
if decision.Outcome != model.PurchaseSpecResolutionUncertain {
|
||||||
|
t.Fatalf("outcome=%s reason=%s", decision.Outcome, decision.Reason)
|
||||||
|
}
|
||||||
|
if !client.deadlineOK {
|
||||||
|
t.Fatal("运行时 AI 上下文缺少截止时间")
|
||||||
|
}
|
||||||
|
budget := client.deadline.Sub(startedAt)
|
||||||
|
if budget < 29*time.Second || budget > purchaseSpecResolutionAITimeout {
|
||||||
|
t.Fatalf("运行时 AI 时间预算=%s,期望不超过 %s", budget, purchaseSpecResolutionAITimeout)
|
||||||
|
}
|
||||||
|
|
||||||
|
cancelledContext, cancel := context.WithCancel(context.Background())
|
||||||
|
cancel()
|
||||||
|
cancelledClient := &runtimeDeadlineAIClient{}
|
||||||
|
cancelledLoader := func(context.Context, *sql.DB, AISecretStore, AIEndpointPolicy) (AIMatchSnapshot, error) {
|
||||||
|
return AIMatchSnapshot{Provider: model.AIProviderConfig{
|
||||||
|
ProviderID: "provider-timeout", Model: "model-timeout", TimeoutSeconds: 120,
|
||||||
|
ConfidenceThresholdBPS: 9000,
|
||||||
|
}, ConfigFingerprint: "fingerprint", Secret: "not-a-real-secret", Client: cancelledClient}, nil
|
||||||
|
}
|
||||||
|
decision = matchPurchaseSpecWithRuntimeAI(cancelledContext, nil, nil,
|
||||||
|
AIEndpointPolicy{}, req, cancelledLoader)
|
||||||
|
if decision.Outcome != model.PurchaseSpecResolutionFailed ||
|
||||||
|
!errors.Is(cancelledClient.contextErr, context.Canceled) {
|
||||||
|
t.Fatalf("更早取消未继承: outcome=%s context_err=%v", decision.Outcome, cancelledClient.contextErr)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestMatchPurchaseSpecWithAI_只接受请求候选并执行安全门禁(t *testing.T) {
|
func TestMatchPurchaseSpecWithAI_只接受请求候选并执行安全门禁(t *testing.T) {
|
||||||
tests := []struct {
|
tests := []struct {
|
||||||
name string
|
name string
|
||||||
|
|||||||
@@ -26,6 +26,7 @@ from .task_models import TaskType
|
|||||||
|
|
||||||
|
|
||||||
DEFAULT_ADMIN_BASE_URL = "https://buy.833729.com"
|
DEFAULT_ADMIN_BASE_URL = "https://buy.833729.com"
|
||||||
|
PURCHASE_SPEC_RESOLUTION_TIMEOUT_SECONDS = 60.0
|
||||||
|
|
||||||
|
|
||||||
class HttpAdminGateway(AdminGateway):
|
class HttpAdminGateway(AdminGateway):
|
||||||
@@ -288,7 +289,10 @@ class HttpAdminGateway(AdminGateway):
|
|||||||
method="POST",
|
method="POST",
|
||||||
)
|
)
|
||||||
try:
|
try:
|
||||||
with self._opener(request, timeout=self._timeout_seconds) as response:
|
with self._opener(
|
||||||
|
request,
|
||||||
|
timeout=PURCHASE_SPEC_RESOLUTION_TIMEOUT_SECONDS,
|
||||||
|
) as response:
|
||||||
status = getattr(response, "status", None) or response.getcode()
|
status = getattr(response, "status", None) or response.getcode()
|
||||||
body = response.read()
|
body = response.read()
|
||||||
except HTTPError as exc:
|
except HTTPError as exc:
|
||||||
|
|||||||
@@ -13,7 +13,11 @@ from src.admin_gateway import (
|
|||||||
ClaimCapabilities,
|
ClaimCapabilities,
|
||||||
ClientInfo,
|
ClientInfo,
|
||||||
)
|
)
|
||||||
from src.http_admin_gateway import DEFAULT_ADMIN_BASE_URL, HttpAdminGateway
|
from src.http_admin_gateway import (
|
||||||
|
DEFAULT_ADMIN_BASE_URL,
|
||||||
|
PURCHASE_SPEC_RESOLUTION_TIMEOUT_SECONDS,
|
||||||
|
HttpAdminGateway,
|
||||||
|
)
|
||||||
from src.task_models import TaskType
|
from src.task_models import TaskType
|
||||||
|
|
||||||
|
|
||||||
@@ -243,7 +247,10 @@ class HttpAdminGatewayTest(unittest.TestCase):
|
|||||||
self.assertEqual(
|
self.assertEqual(
|
||||||
headers["idempotency-key"], "spec-resolution-v1:key"
|
headers["idempotency-key"], "spec-resolution-v1:key"
|
||||||
)
|
)
|
||||||
self.assertEqual(opener.timeout, 5)
|
self.assertEqual(
|
||||||
|
opener.timeout, PURCHASE_SPEC_RESOLUTION_TIMEOUT_SECONDS
|
||||||
|
)
|
||||||
|
self.assertEqual(PURCHASE_SPEC_RESOLUTION_TIMEOUT_SECONDS, 60.0)
|
||||||
|
|
||||||
def test_resolve_purchase_spec_accepts_all_nonmatched_business_outcomes(self):
|
def test_resolve_purchase_spec_accepts_all_nonmatched_business_outcomes(self):
|
||||||
for outcome in ("uncertain", "rejected", "failed"):
|
for outcome in ("uncertain", "rejected", "failed"):
|
||||||
|
|||||||
@@ -52,9 +52,11 @@ class _RecordingOpener:
|
|||||||
def __init__(self, response):
|
def __init__(self, response):
|
||||||
self.response = response
|
self.response = response
|
||||||
self.request = None
|
self.request = None
|
||||||
|
self.timeout = None
|
||||||
|
|
||||||
def __call__(self, request, timeout):
|
def __call__(self, request, timeout):
|
||||||
self.request = request
|
self.request = request
|
||||||
|
self.timeout = timeout
|
||||||
if isinstance(self.response, Exception):
|
if isinstance(self.response, Exception):
|
||||||
raise self.response
|
raise self.response
|
||||||
return self.response
|
return self.response
|
||||||
@@ -279,6 +281,7 @@ class PurchaseSpecResolutionContractVectorTest(unittest.TestCase):
|
|||||||
for name, value in opener.request.header_items()
|
for name, value in opener.request.header_items()
|
||||||
}
|
}
|
||||||
self.assertEqual(headers["idempotency-key"], key)
|
self.assertEqual(headers["idempotency-key"], key)
|
||||||
|
self.assertEqual(opener.timeout, 60.0)
|
||||||
|
|
||||||
def test_error_vectors_are_identical_for_http_and_mock(self):
|
def test_error_vectors_are_identical_for_http_and_mock(self):
|
||||||
defaults = self.fixture["defaults"]
|
defaults = self.fixture["defaults"]
|
||||||
|
|||||||
@@ -267,6 +267,11 @@ schema v1 只包含:`task_version`、`attempt_id`、`pdd_goods_id`、领取任
|
|||||||
候选快照哈希和观测时间。整个请求体最多 64 KiB。候选编号必须按页面顺序严格使用
|
候选快照哈希和观测时间。整个请求体最多 64 KiB。候选编号必须按页面顺序严格使用
|
||||||
`c1`~`c100`,每条只保存页面原文及原始 `color/size`;禁止原始控件树和截图。
|
`c1`~`c100`,每条只保存页面原文及原始 `color/size`;禁止原始控件树和截图。
|
||||||
|
|
||||||
|
运行时 AI 调用必须使用独立的 **30 秒上限**;服务商 `timeout_seconds` 更短时以更短值
|
||||||
|
为准,并继承 Client 断开造成的更早取消。Client 为本命令保留最多 **60 秒**等待时间,
|
||||||
|
因此 Admin 必须在自己的预算内完成审计和响应。其他 Client 接口继续使用原短超时,
|
||||||
|
本命令仍不得改成 GET、后台轮询或无限等待。
|
||||||
|
|
||||||
处理顺序:
|
处理顺序:
|
||||||
|
|
||||||
1. 限制请求体大小并校验 schema、字段长度、候选数量、连续短编号、原文和 options;
|
1. 限制请求体大小并校验 schema、字段长度、候选数量、连续短编号、原文和 options;
|
||||||
|
|||||||
@@ -418,6 +418,11 @@ Idempotency-Key: task-id:attempt-id:failure-v1
|
|||||||
规格决策。它是**一次性 POST 业务命令**,不是任务状态查询、心跳或轮询;Client 不得
|
规格决策。它是**一次性 POST 业务命令**,不是任务状态查询、心跳或轮询;Client 不得
|
||||||
用它询问任务是否取消、是否重派或 AI 是否完成。
|
用它询问任务是否取消、是否重派或 AI 是否完成。
|
||||||
|
|
||||||
|
本命令有独立的跨端时间预算:Client HTTP 调用最多等待 **60 秒**;Admin 运行时 AI
|
||||||
|
调用最多等待 **30 秒**,服务商配置更短时以更短时间为准。普通登记、领取和结果提交
|
||||||
|
仍使用 Client 原有的短请求超时,不因 AI 调用而延长。Admin 应在 30 秒预算内持久化并
|
||||||
|
返回安全结论;Client 等待超时后按失败结束本次执行,不自动循环请求或重新下单。
|
||||||
|
|
||||||
```http
|
```http
|
||||||
POST /api/v1/client/tasks/{task_id}/spec-resolution
|
POST /api/v1/client/tasks/{task_id}/spec-resolution
|
||||||
Idempotency-Key: spec-resolution-v1:<identity-sha256>
|
Idempotency-Key: spec-resolution-v1:<identity-sha256>
|
||||||
|
|||||||
Reference in New Issue
Block a user