feat: 新增 cancelled 状态区分客户端主动断开连接

- 新增 cancelled 请求状态,与系统失败 (failed) 区分
- 客户端主动断开连接 (HTTP 499) 不计入失败率统计
- 前后端同步更新状态枚举、UI 样式和统计逻辑
This commit is contained in:
fawney19
2026-01-20 15:40:16 +08:00
parent ffca11bc18
commit 4eeba69381
11 changed files with 211 additions and 15 deletions

View File

@@ -15,7 +15,7 @@ export interface CandidateRecord {
key_preview?: string // 密钥脱敏预览(如 sk-***abc key_preview?: string // 密钥脱敏预览(如 sk-***abc
key_capabilities?: Record<string, boolean> | null // Key 支持的能力 key_capabilities?: Record<string, boolean> | null // Key 支持的能力
required_capabilities?: Record<string, boolean> | null // 请求实际需要的能力标签 required_capabilities?: Record<string, boolean> | null // 请求实际需要的能力标签
status: 'pending' | 'streaming' | 'success' | 'failed' | 'skipped' status: 'pending' | 'streaming' | 'success' | 'failed' | 'skipped' | 'cancelled'
skip_reason?: string skip_reason?: string
is_cached: boolean is_cached: boolean
// 执行结果字段 // 执行结果字段
@@ -33,7 +33,7 @@ export interface CandidateRecord {
export interface RequestTrace { export interface RequestTrace {
request_id: string request_id: string
total_candidates: number total_candidates: number
final_status: 'success' | 'failed' | 'streaming' | 'pending' final_status: 'success' | 'failed' | 'streaming' | 'pending' | 'cancelled'
total_latency_ms: number total_latency_ms: number
candidates: CandidateRecord[] candidates: CandidateRecord[]
} }
@@ -42,6 +42,7 @@ export interface ProviderStats {
total_attempts: number total_attempts: number
success_count: number success_count: number
failed_count: number failed_count: number
cancelled_count: number
skipped_count: number skipped_count: number
pending_count: number pending_count: number
available_count: number available_count: number

View File

@@ -734,6 +734,7 @@ const getStatusLabel = (status: string) => {
streaming: '传输中', streaming: '传输中',
success: '成功', success: '成功',
failed: '失败', failed: '失败',
cancelled: '已取消',
skipped: '跳过' skipped: '跳过'
} }
return labels[status] || status return labels[status] || status
@@ -747,6 +748,7 @@ const getStatusColorClass = (status: string) => {
streaming: 'status-pending', streaming: 'status-pending',
success: 'status-success', success: 'status-success',
failed: 'status-failed', failed: 'status-failed',
cancelled: 'status-cancelled',
skipped: 'status-skipped' skipped: 'status-skipped'
} }
return classes[status] || 'status-available' return classes[status] || 'status-available'
@@ -901,6 +903,7 @@ const getStatusColorClass = (status: string) => {
/* 子节点状态颜色 */ /* 子节点状态颜色 */
.sub-dot.status-success { background: #22c55e; color: #22c55e; } .sub-dot.status-success { background: #22c55e; color: #22c55e; }
.sub-dot.status-failed { background: #ef4444; color: #ef4444; } .sub-dot.status-failed { background: #ef4444; color: #ef4444; }
.sub-dot.status-cancelled { background: #f59e0b; color: #f59e0b; }
.sub-dot.status-pending { background: #3b82f6; color: #3b82f6; } .sub-dot.status-pending { background: #3b82f6; color: #3b82f6; }
.sub-dot.status-skipped { background: #1f2937; color: #1f2937; } .sub-dot.status-skipped { background: #1f2937; color: #1f2937; }
.sub-dot.status-available { background: #d1d5db; color: #d1d5db; } .sub-dot.status-available { background: #d1d5db; color: #d1d5db; }
@@ -963,6 +966,7 @@ const getStatusColorClass = (status: string) => {
/* 状态颜色 - 同心圆使用 color */ /* 状态颜色 - 同心圆使用 color */
.node-dot.status-success { color: #22c55e; } .node-dot.status-success { color: #22c55e; }
.node-dot.status-failed { color: #ef4444; } .node-dot.status-failed { color: #ef4444; }
.node-dot.status-cancelled { color: #f59e0b; }
.node-dot.status-pending { color: #3b82f6; } .node-dot.status-pending { color: #3b82f6; }
.node-dot.status-skipped { color: #1f2937; } .node-dot.status-skipped { color: #1f2937; }
.node-dot.status-available { color: #d1d5db; } .node-dot.status-available { color: #d1d5db; }
@@ -1010,6 +1014,7 @@ const getStatusColorClass = (status: string) => {
.title-dot.status-success { background: #22c55e; } .title-dot.status-success { background: #22c55e; }
.title-dot.status-failed { background: #ef4444; } .title-dot.status-failed { background: #ef4444; }
.title-dot.status-cancelled { background: #f59e0b; }
.title-dot.status-pending { background: #3b82f6; } .title-dot.status-pending { background: #3b82f6; }
.title-dot.status-skipped { background: #1f2937; } .title-dot.status-skipped { background: #1f2937; }
.title-dot.status-available { background: #d1d5db; } .title-dot.status-available { background: #d1d5db; }
@@ -1093,6 +1098,11 @@ const getStatusColorClass = (status: string) => {
color: #dc2626; color: #dc2626;
} }
.status-tag.status-cancelled {
background: #f59e0b20;
color: #d97706;
}
.status-tag.status-pending { .status-tag.status-pending {
background: #3b82f620; background: #3b82f620;
color: #2563eb; color: #2563eb;

View File

@@ -54,7 +54,7 @@ export interface ApiFormatStatsItem {
// 请求记录 // 请求记录
// 请求状态类型 // 请求状态类型
export type RequestStatus = 'pending' | 'streaming' | 'completed' | 'failed' export type RequestStatus = 'pending' | 'streaming' | 'completed' | 'failed' | 'cancelled'
export interface UsageRecord { export interface UsageRecord {
id: string id: string
@@ -104,7 +104,7 @@ export interface DateRangeParams {
export type PeriodValue = 'today' | 'yesterday' | 'last7days' | 'last30days' | 'last90days' export type PeriodValue = 'today' | 'yesterday' | 'last7days' | 'last30days' | 'last90days'
// 筛选状态(包含新的请求状态值) // 筛选状态(包含新的请求状态值)
export type FilterStatusValue = '__all__' | 'stream' | 'standard' | 'error' | 'active' | 'pending' | 'streaming' | 'completed' | 'failed' export type FilterStatusValue = '__all__' | 'stream' | 'standard' | 'error' | 'active' | 'pending' | 'streaming' | 'completed' | 'failed' | 'cancelled'
// 默认统计状态 // 默认统计状态
export function createDefaultStats(): UsageStatsState { export function createDefaultStats(): UsageStatsState {

View File

@@ -1219,8 +1219,8 @@ function generateMockEndpointsForProvider(providerId: string) {
provider_name: provider.name, provider_name: provider.name,
api_format: format, api_format: format,
base_url: format.includes('CLAUDE') ? 'https://api.anthropic.com' : base_url: format.includes('CLAUDE') ? 'https://api.anthropic.com' :
format.includes('OPENAI') ? 'https://api.openai.com' : format.includes('OPENAI') ? 'https://api.openai.com' :
'https://generativelanguage.googleapis.com', 'https://generativelanguage.googleapis.com',
max_retries: 2, max_retries: 2,
is_active: healthDetail?.is_active ?? true, is_active: healthDetail?.is_active ?? true,
total_keys: Math.ceil(Math.random() * 3) + 1, total_keys: Math.ceil(Math.random() * 3) + 1,

View File

@@ -60,7 +60,7 @@ class RequestTraceResponse(BaseModel):
request_id: str request_id: str
total_candidates: int total_candidates: int
final_status: str # 'success', 'failed', 'streaming', 'pending' final_status: str # 'success', 'failed', 'cancelled', 'streaming', 'pending'
total_latency_ms: int total_latency_ms: int
candidates: List[CandidateResponse] candidates: List[CandidateResponse]
@@ -164,12 +164,12 @@ class AdminGetRequestTraceAdapter(AdminApiAdapter):
if not candidates: if not candidates:
raise HTTPException(status_code=404, detail="Request not found") raise HTTPException(status_code=404, detail="Request not found")
# 计算总延迟只统计已完成的候选success failed # 计算总延迟只统计已完成的候选success, failed, cancelled
# 使用显式的 is not None 检查,避免过滤掉 0ms 的快速响应 # 使用显式的 is not None 检查,避免过滤掉 0ms 的快速响应
total_latency = sum( total_latency = sum(
c.latency_ms c.latency_ms
for c in candidates for c in candidates
if c.status in ("success", "failed") and c.latency_ms is not None if c.status in ("success", "failed", "cancelled") and c.latency_ms is not None
) )
# 判断最终状态: # 判断最终状态:
@@ -179,6 +179,7 @@ class AdminGetRequestTraceAdapter(AdminApiAdapter):
# - 用于兼容非流式请求或未正确设置 status 的旧数据 # - 用于兼容非流式请求或未正确设置 status 的旧数据
# 3. status="streaming" 表示流式请求正在进行中 # 3. status="streaming" 表示流式请求正在进行中
# 4. status="pending" 表示请求尚未开始执行 # 4. status="pending" 表示请求尚未开始执行
# 5. status="cancelled" 表示客户端主动断开连接(不算失败)
has_success = any( has_success = any(
c.status == "success" c.status == "success"
or (c.status_code is not None and 200 <= c.status_code < 300) or (c.status_code is not None and 200 <= c.status_code < 300)
@@ -186,6 +187,8 @@ class AdminGetRequestTraceAdapter(AdminApiAdapter):
) )
has_streaming = any(c.status == "streaming" for c in candidates) has_streaming = any(c.status == "streaming" for c in candidates)
has_pending = any(c.status == "pending" for c in candidates) has_pending = any(c.status == "pending" for c in candidates)
has_cancelled = any(c.status == "cancelled" for c in candidates)
has_failed = any(c.status == "failed" for c in candidates)
if has_success: if has_success:
final_status = "success" final_status = "success"
@@ -195,6 +198,9 @@ class AdminGetRequestTraceAdapter(AdminApiAdapter):
elif has_pending: elif has_pending:
# 有候选正在等待执行 # 有候选正在等待执行
final_status = "pending" final_status = "pending"
elif has_cancelled and not has_failed:
# 只有取消没有失败,算作取消
final_status = "cancelled"
else: else:
final_status = "failed" final_status = "failed"

View File

@@ -246,6 +246,62 @@ class MessageTelemetry:
target_model=target_model, target_model=target_model,
) )
async def record_cancelled(
self,
*,
provider: str,
model: str,
response_time_ms: int,
first_byte_time_ms: Optional[int],
status_code: int,
request_body: Dict[str, Any],
request_headers: Dict[str, Any],
is_stream: bool,
api_format: Optional[str] = None,
provider_request_headers: Optional[Dict[str, Any]] = None,
input_tokens: int = 0,
output_tokens: int = 0,
cache_creation_tokens: int = 0,
cache_read_tokens: int = 0,
response_body: Optional[Dict[str, Any]] = None,
response_headers: Optional[Dict[str, Any]] = None,
client_response_headers: Optional[Dict[str, Any]] = None,
target_model: Optional[str] = None,
) -> None:
"""
记录客户端取消的请求
客户端主动断开连接不算系统失败,使用 cancelled 状态。
"""
provider_name = provider or "unknown"
await UsageService.record_usage(
db=self.db,
user=self.user,
api_key=self.api_key,
provider=provider_name,
model=model,
input_tokens=input_tokens,
output_tokens=output_tokens,
cache_creation_input_tokens=cache_creation_tokens,
cache_read_input_tokens=cache_read_tokens,
request_type="chat",
api_format=api_format,
is_stream=is_stream,
response_time_ms=response_time_ms,
first_byte_time_ms=first_byte_time_ms,
status_code=status_code,
status="cancelled",
request_headers=request_headers,
request_body=request_body,
provider_request_headers=provider_request_headers or {},
response_headers=response_headers or {},
client_response_headers=client_response_headers,
response_body=response_body or {},
request_id=self.request_id,
target_model=target_model,
)
@runtime_checkable @runtime_checkable
class MessageHandlerProtocol(Protocol): class MessageHandlerProtocol(Protocol):

View File

@@ -214,6 +214,10 @@ class StreamContext:
"""检查请求是否成功""" """检查请求是否成功"""
return self.status_code < 400 return self.status_code < 400
def is_client_disconnected(self) -> bool:
"""检查是否因客户端断开连接而结束"""
return self.status_code == 499
def build_response_body(self, response_time_ms: int) -> Dict[str, Any]: def build_response_body(self, response_time_ms: int) -> Dict[str, Any]:
""" """
构建响应体元数据 构建响应体元数据

View File

@@ -97,9 +97,15 @@ class StreamTelemetryRecorder:
logger.warning( logger.warning(
f"[{self.request_id}] User or ApiKey not found, updating status directly" f"[{self.request_id}] User or ApiKey not found, updating status directly"
) )
if ctx.is_success():
status = "completed"
elif ctx.is_client_disconnected():
status = "cancelled"
else:
status = "failed"
await self._update_usage_status_directly( await self._update_usage_status_directly(
bg_db, bg_db,
status="completed" if ctx.is_success() else "failed", status=status,
response_time_ms=response_time_ms, response_time_ms=response_time_ms,
status_code=ctx.status_code, status_code=ctx.status_code,
) )
@@ -121,6 +127,15 @@ class StreamTelemetryRecorder:
response_body, response_body,
response_time_ms, response_time_ms,
) )
elif ctx.is_client_disconnected():
await self._record_cancelled(
bg_telemetry,
ctx,
original_headers,
actual_request_body,
response_body,
response_time_ms,
)
else: else:
await self._record_failure( await self._record_failure(
bg_telemetry, bg_telemetry,
@@ -229,6 +244,42 @@ class StreamTelemetryRecorder:
# 对于失败日志,添加缓存信息 # 对于失败日志,添加缓存信息
logger.info(f"{log_summary} cache:{ctx.cached_tokens}") logger.info(f"{log_summary} cache:{ctx.cached_tokens}")
async def _record_cancelled(
self,
telemetry: MessageTelemetry,
ctx: StreamContext,
original_headers: Dict[str, str],
actual_request_body: Dict[str, Any],
response_body: Dict[str, Any],
response_time_ms: int,
) -> None:
"""记录客户端取消的请求"""
client_response_headers = ctx.client_response_headers or {"content-type": "application/json"}
await telemetry.record_cancelled(
provider=ctx.provider_name or "unknown",
model=ctx.model,
response_time_ms=response_time_ms,
first_byte_time_ms=ctx.first_byte_time_ms,
status_code=ctx.status_code,
request_headers=original_headers,
request_body=actual_request_body,
is_stream=True,
api_format=ctx.api_format,
provider_request_headers=ctx.provider_request_headers,
input_tokens=ctx.input_tokens,
output_tokens=ctx.output_tokens,
cache_creation_tokens=ctx.cache_creation_tokens,
cache_read_tokens=ctx.cached_tokens,
response_body=response_body,
response_headers=ctx.response_headers,
client_response_headers=client_response_headers,
target_model=ctx.mapped_model,
)
logger.debug(f"{self.format_id} 流式响应被客户端取消")
logger.info(ctx.get_log_summary(self.request_id, response_time_ms))
async def _update_candidate_status( async def _update_candidate_status(
self, self,
db: Session, db: Session,
@@ -264,14 +315,21 @@ class StreamTelemetryRecorder:
latency_ms=response_time_ms, latency_ms=response_time_ms,
extra_data=extra_data, extra_data=extra_data,
) )
elif ctx.is_client_disconnected():
RequestCandidateService.mark_candidate_cancelled(
db=db,
candidate_id=ctx.attempt_id,
status_code=ctx.status_code,
latency_ms=response_time_ms,
extra_data=extra_data,
)
else: else:
error_type = "client_disconnected" if ctx.status_code == 499 else "stream_error"
# 请求链路追踪使用 upstream_response原始响应回退到 error_message友好消息 # 请求链路追踪使用 upstream_response原始响应回退到 error_message友好消息
trace_error_message = ctx.upstream_response or ctx.error_message or f"HTTP {ctx.status_code}" trace_error_message = ctx.upstream_response or ctx.error_message or f"HTTP {ctx.status_code}"
RequestCandidateService.mark_candidate_failed( RequestCandidateService.mark_candidate_failed(
db=db, db=db,
candidate_id=ctx.attempt_id, candidate_id=ctx.attempt_id,
error_type=error_type, error_type="stream_error",
error_message=trace_error_message, error_message=trace_error_message,
status_code=ctx.status_code, status_code=ctx.status_code,
latency_ms=response_time_ms, latency_ms=response_time_ms,

View File

@@ -337,6 +337,7 @@ class Usage(Base):
# streaming: 流式响应进行中 # streaming: 流式响应进行中
# completed: 请求成功完成 # completed: 请求成功完成
# failed: 请求失败 # failed: 请求失败
# cancelled: 客户端主动断开连接
status = Column(String(20), default="completed", nullable=False, index=True) status = Column(String(20), default="completed", nullable=False, index=True)
# 完整请求和响应记录 # 完整请求和响应记录
@@ -1567,7 +1568,7 @@ class RequestCandidate(Base):
) )
# 状态信息 # 状态信息
status = Column(String(20), nullable=False) # 'pending', 'success', 'failed', 'skipped' status = Column(String(20), nullable=False) # 'pending', 'streaming', 'success', 'failed', 'cancelled', 'skipped'
skip_reason = Column(Text, nullable=True) # 跳过/失败原因 skip_reason = Column(Text, nullable=True) # 跳过/失败原因
is_cached = Column(Boolean, default=False) # 是否为缓存亲和性候选 is_cached = Column(Boolean, default=False) # 是否为缓存亲和性候选

View File

@@ -210,6 +210,39 @@ class RequestCandidateService:
# 原因:前端需要实时看到请求成功/失败状态 # 原因:前端需要实时看到请求成功/失败状态
db.commit() db.commit()
@staticmethod
def mark_candidate_cancelled(
db: Session,
candidate_id: str,
status_code: int = 499,
latency_ms: Optional[int] = None,
concurrent_requests: Optional[int] = None,
extra_data: Optional[dict] = None,
) -> None:
"""
标记候选被客户端取消
客户端主动断开连接不算系统失败,使用 cancelled 状态。
Args:
db: 数据库会话
candidate_id: 候选ID
status_code: HTTP 状态码(通常是 499
latency_ms: 延迟(毫秒)
concurrent_requests: 并发请求数
extra_data: 额外数据
"""
candidate = db.query(RequestCandidate).filter(RequestCandidate.id == candidate_id).first()
if candidate:
candidate.status = "cancelled"
candidate.status_code = status_code
candidate.latency_ms = latency_ms
candidate.concurrent_requests = concurrent_requests
candidate.finished_at = datetime.now(timezone.utc)
if extra_data:
candidate.extra_data = {**(candidate.extra_data or {}), **extra_data}
db.commit()
@staticmethod @staticmethod
def mark_candidate_skipped( def mark_candidate_skipped(
db: Session, candidate_id: str, skip_reason: Optional[str] = None db: Session, candidate_id: str, skip_reason: Optional[str] = None
@@ -273,11 +306,12 @@ class RequestCandidateService:
total_candidates = len(candidates) total_candidates = len(candidates)
success_count = sum(1 for c in candidates if c.status == "success") success_count = sum(1 for c in candidates if c.status == "success")
failed_count = sum(1 for c in candidates if c.status == "failed") failed_count = sum(1 for c in candidates if c.status == "failed")
cancelled_count = sum(1 for c in candidates if c.status == "cancelled")
skipped_count = sum(1 for c in candidates if c.status == "skipped") skipped_count = sum(1 for c in candidates if c.status == "skipped")
pending_count = sum(1 for c in candidates if c.status == "pending") pending_count = sum(1 for c in candidates if c.status == "pending")
available_count = sum(1 for c in candidates if c.status == "available") available_count = sum(1 for c in candidates if c.status == "available")
# 计算失败率(只统计已完成的候选,即成功或失败的) # 计算失败率(只统计已完成的候选,即成功或失败的cancelled 不算失败
completed_count = success_count + failed_count completed_count = success_count + failed_count
failure_rate = (failed_count / completed_count * 100) if completed_count > 0 else 0 failure_rate = (failed_count / completed_count * 100) if completed_count > 0 else 0
@@ -285,9 +319,10 @@ class RequestCandidateService:
"total_attempts": total_candidates, # 前端使用 total_attempts 字段 "total_attempts": total_candidates, # 前端使用 total_attempts 字段
"success_count": success_count, "success_count": success_count,
"failed_count": failed_count, "failed_count": failed_count,
"cancelled_count": cancelled_count, # 客户端取消数
"skipped_count": skipped_count, "skipped_count": skipped_count,
"pending_count": pending_count, "pending_count": pending_count,
"available_count": available_count, # 新增:尚未被调度的候选数 "available_count": available_count, # 尚未被调度的候选数
"failure_rate": round(failure_rate, 2), "failure_rate": round(failure_rate, 2),
} }

View File

@@ -24,6 +24,7 @@ class RequestStatus(Enum):
SUCCESS = "success" SUCCESS = "success"
FAILED = "failed" FAILED = "failed"
PARTIAL = "partial" # 流式请求部分成功 PARTIAL = "partial" # 流式请求部分成功
CANCELLED = "cancelled" # 客户端主动断开连接
@dataclass @dataclass
@@ -178,6 +179,10 @@ class RequestResult:
def is_failed(self) -> bool: def is_failed(self) -> bool:
return self.status == RequestStatus.FAILED return self.status == RequestStatus.FAILED
@property
def is_cancelled(self) -> bool:
return self.status == RequestStatus.CANCELLED
@classmethod @classmethod
def success( def success(
cls, cls,
@@ -219,6 +224,26 @@ class RequestResult:
is_stream=is_stream, is_stream=is_stream,
) )
@classmethod
def cancelled(
cls,
metadata: RequestMetadata,
response_time_ms: int,
usage: Optional[UsageInfo] = None,
is_stream: bool = False,
) -> "RequestResult":
"""创建客户端取消的请求结果"""
return cls(
status=RequestStatus.CANCELLED,
metadata=metadata,
status_code=499,
error_message="client_disconnected",
error_type="client_disconnected",
response_time_ms=response_time_ms,
usage=usage or UsageInfo(),
is_stream=is_stream,
)
@classmethod @classmethod
def from_exception( def from_exception(
cls, cls,