package handler import ( "context" "errors" "net" "strings" "sync" "syscall" "time" "github.com/qianfree/team-api/internal/dispatchadapter" "github.com/qianfree/team-api/internal/logic/monitor" "github.com/qianfree/team-api/relay/common" "github.com/qianfree/team-api/relay/constant" "github.com/qianfree/team-api/relaykit/dispatch" "github.com/qianfree/team-api/relaykit/types" ) // 本文件是 handler 与新调度引擎(relaykit/dispatch)之间的胶水层: // 送达状态标注、状态码提取、退避执行、租约续期、决策明细挂 trace。 // dispatchStatusCode 从错误中提取上游 HTTP 状态码。 // RelayError 直接携带;relaykit NewAPIError 由分类器自行解包,此处返回 0 即可。 func dispatchStatusCode(err error) int { var relayErr *constant.RelayError if errors.As(err, &relayErr) { return relayErr.StatusCode } return 0 } // retryAfterOf 从错误中提取上游 Retry-After(由共享 handler 在构造 RelayError 时附着)。 // FSM 据此在 429 时决定「原地等待重试」还是立即 failover;0 = 未携带。 func retryAfterOf(err error) time.Duration { var relayErr *constant.RelayError if errors.As(err, &relayErr) { return relayErr.RetryAfter } return 0 } // deliveryStateOfRequestErr DoRequest 阶段错误的送达状态标注: // 连接拒绝 / DNS 失败 / TLS 建连失败 = 请求确定未发出(可安全重试); // 其余(写出后 RST/EOF/读超时)= 可能已送达上游,非幂等请求禁止重放。 func deliveryStateOfRequestErr(err error) dispatch.DeliveryState { if err == nil { return dispatch.DeliveryNotSent } var dnsErr *net.DNSError if errors.As(err, &dnsErr) { return dispatch.DeliveryNotSent } if errors.Is(err, syscall.ECONNREFUSED) { return dispatch.DeliveryNotSent } msg := strings.ToLower(err.Error()) for _, marker := range []string{ "connection refused", "no such host", "dial tcp", "tls: handshake", "certificate", "proxyconnect", } { if strings.Contains(msg, marker) { return dispatch.DeliveryNotSent } } return dispatch.DeliveryMaybeSent } // reportChannelFatal 渠道级致命错误上报(请求未送达上游),让 FSM 决定换渠道或终止。 // code 必须为 channel: 前缀错误码(classify 归 ErrClassChannelFatal → failover)。 func reportChannelFatal(ctx context.Context, sess *dispatch.RouteSession, err error, code types.ErrorCode) dispatch.RetryDecision { failErr := types.NewError(err, code) decision, _ := sess.Report(ctx, 0, failErr, dispatch.DeliveryNotSent, 0, 0) trackRetryDecision(0, failErr, dispatch.DeliveryNotSent, decision) return decision } // reportMaterializeFailure 选择物化失败(Key 解密失败/目录缺失)时按渠道级致命上报, // 让 FSM 决定换渠道或终止。不发起过上游请求,送达状态为 NotSent。 func reportMaterializeFailure(ctx context.Context, sess *dispatch.RouteSession, mErr error) dispatch.RetryDecision { return reportChannelFatal(ctx, sess, mErr, types.ErrorCodeChannelNoAvailableKey) } // reportProtocolMismatch 渠道协议能力不匹配(如 responses 有状态请求落在 chat-only // 渠道)时上报:换渠道可能得到支持 Responses 协议的渠道,不等于渠道本身故障。 func reportProtocolMismatch(ctx context.Context, sess *dispatch.RouteSession, err error) dispatch.RetryDecision { return reportChannelFatal(ctx, sess, err, types.ErrorCodeChannelProtocolMismatch) } // sleepBackoff 执行退避等待,客户端断开时立即返回。 func sleepBackoff(ctx context.Context, d time.Duration) { if d <= 0 { return } timer := time.NewTimer(d) defer timer.Stop() select { case <-timer.C: case <-ctx.Done(): } } // appendSchedulerDecision 调度决策明细挂 ForwardingTrace + 选择指标计数。 func appendSchedulerDecision(trace *common.ForwardingTrace, d *dispatch.Decision, attempt int) { if d == nil { monitor.TrackDispatchNoCandidate() return } monitor.TrackDispatchSelection(string(d.Reason), string(d.Channel.Tier), string(d.SessionKey.Source)) if trace == nil { return } trace.Scheduler = append(trace.Scheduler, common.SchedulerDecision{ Attempt: attempt, ChannelID: d.Channel.ID, KeyID: d.KeyID, Reason: string(d.Reason), Tier: string(d.Channel.Tier), SessionSource: string(d.SessionKey.Source), Weights: map[string]float64{ "base": d.Breakdown.Base, "tier": d.Breakdown.Tier, "health": d.Breakdown.Health, "headroom": d.Breakdown.Headroom, "cost": d.Breakdown.Cost, "ramp": d.Breakdown.Ramp, "effective": d.Breakdown.Effective, }, Candidates: d.CandidateCount, ExcludedBreaker: d.Excluded.Breaker, ExcludedLease: d.Excluded.Lease, ExcludedRequest: d.Excluded.Request, }) } // trackRetryDecision 重试决策指标计数(dispatch_retry_total{err_class→decision})。 func trackRetryDecision(statusCode int, err error, delivery dispatch.DeliveryState, decision dispatch.RetryDecision) { monitor.TrackDispatchRetry(dispatch.Classify(statusCode, err, delivery).String(), decision.String()) } // dispatchLeaseRefresher 长请求(流式/websocket)的调度租约续期器。 // 直连 Redis 状态续期而非经 RouteSession(RouteSession 非并发安全)。 type dispatchLeaseRefresher struct { stop chan struct{} once sync.Once } // startDispatchLeaseRefresher 每 30s 续期一次调度租约(租约默认 90s,实例崩溃后自然过期)。 func startDispatchLeaseRefresher(channelID int64, requestID string) *dispatchLeaseRefresher { r := &dispatchLeaseRefresher{stop: make(chan struct{})} go func() { ticker := time.NewTicker(30 * time.Second) defer ticker.Stop() for { select { case <-ticker.C: ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) dispatchadapter.RefreshDispatchLease(ctx, channelID, requestID) cancel() case <-r.stop: return } } }() return r } // Stop 停止续期(幂等)。 func (r *dispatchLeaseRefresher) Stop() { if r == nil { return } r.once.Do(func() { close(r.stop) }) }