proxy

package
v0.0.1-beta.8 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 12, 2026 License: AGPL-3.0 Imports: 48 Imported by: 0

Documentation

Overview

SPDX-License-Identifier: AGPL-3.0-or-later

SPDX-License-Identifier: AGPL-3.0-or-later

SPDX-License-Identifier: AGPL-3.0-or-later

SPDX-License-Identifier: AGPL-3.0-or-later

异步续接绑定 worker（REST 流式）：删除 REST 闸门后，业务帧不再等待 Redis 绑定确认——首个有效响应 id 帧在写出接缝处**同步快照**为 continuation.BindRequest 投入有界队列，本 worker 常驻批量落库。绑定语义由"生产侧 fail-closed（绑完才 可见）"改为"消费侧 fail-closed（lookup/pin 校验）"；worker 的丢弃/失败/冲突 **只计数、绝不回写当前响应**。

纪律对齐 internal/usage/errlog.go：有界队列 + 非阻塞投递（队列满 → 丢弃 + 原子计数）；worker.GoLoop 托管；停机**先禁入队再排空**。批节奏：从首条入批起 最多等 BatchWait，或攒够 BatchSize；每批 Redis 操作受 BatchTimeout 约束。

SPDX-License-Identifier: AGPL-3.0-or-later

SPDX-License-Identifier: AGPL-3.0-or-later

Package proxy 是 AI 请求热路径：分组 key 鉴权 → 调度器选号 → SDK 转发 → 用量采集。 规格 §6/§9。不变量：热路径零 DB、零 per-request 锁。

SPDX-License-Identifier: AGPL-3.0-or-later

SPDX-License-Identifier: AGPL-3.0-or-later

SPDX-License-Identifier: AGPL-3.0-or-later

SPDX-License-Identifier: AGPL-3.0-or-later

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrAttemptObserverNotDispatched = errors.New("attempt observer: outcome is not dispatched")
	ErrAttemptObserverHealthKind    = errors.New("attempt observer: invalid health kind")
)

Functions

func AIRouter 

func AIRouter(p *Proxy) http.Handler

AIRouter 挂载 AI 端点（规格 §6.1/§9）：路径决定请求格式，全部走通用转发 骨架 handleFormat（UpstreamCaller 注册表分发）；resp-ws 与 HTTP responses 同路径（真实客户端无 /ws 后缀）——/v1/responses 带 upgrade 头 → 按 resp-ws 处理，走专用编排 HandleResponsesWS（caller_responses_ws.go）。

func AdaptOutcomeToObservation 

func AdaptOutcomeToObservation(o AttemptOutcome) (quality.Observation, bool)

AdaptOutcomeToObservation maps a canonical AttemptOutcome onto a quality Observation. The outcome must pass Validate and IsCountedForQuality; cancel/local/reservation and invalid outcomes are excluded (ok=false). Failures — malformed included, i.e. post-commit failures — are classified by HTTP status: 429 / ordinary 4xx / 5xx / status-0 network.

func CanRetry 

func CanRetry(cat CallerCategory, o AttemptOutcome) bool

Types

type AttemptFlowAppend 

type AttemptFlowAppend func(AttemptOutcome)

type AttemptHealthEvent 

type AttemptHealthEvent struct {
	Kind         rule.Kind
	ResetAt      *time.Time
	ErrorMessage string
	Retryable    bool
}

type AttemptHealthMark 

type AttemptHealthMark func(AttemptOutcome, AttemptHealthEvent)

type AttemptID 

type AttemptID string

AttemptID is a per-attempt identifier. Dispatched metadata (RouteClassID/QualityClassID/Fingerprint/Lane/Generation/IdentityRevision) is plan-canonical: it comes from the scheduler.Attempt recorded by ReserveAttempt (pipelineBase), never from placeholder values.

func (AttemptID) Validate 

func (a AttemptID) Validate() error

type AttemptLeaseRelease 

type AttemptLeaseRelease func()

type AttemptObserver 

type AttemptObserver struct {
	// contains filtered or unexported fields
}

func NewAttemptObserver 

func NewAttemptObserver(qualityContext *quality.AttemptContext, markHealth AttemptHealthMark, appendFlow AttemptFlowAppend, release AttemptLeaseRelease) *AttemptObserver

func (*AttemptObserver) Abandon 

func (o *AttemptObserver) Abandon()

Abandon is the deferred owner cleanup for a dispatch that ended without a terminal outcome (local reject after reservation, panic mid-call). It is not an observation: no quality attempt, no flow record, no health, no release — it only releases the AttemptContext pin taken at dispatch time so the recorder's bounded inflight cannot leak.

func (*AttemptObserver) Cancel 

func (o *AttemptObserver) Cancel(outcome AttemptOutcome) error

func (*AttemptObserver) Complete 

func (o *AttemptObserver) Complete(outcome AttemptOutcome, health *AttemptHealthEvent) error

type AttemptOutcome 

type AttemptOutcome struct {
	ID             AttemptID
	RouteClassID   RouteClassID
	QualityClassID QualityClassID
	Fingerprint    CandidateFingerprint
	TemplateID     int64
	AccountID      int64
	RequestedModel string
	MappedModel    string
	CallerCategory CallerCategory
	OperationTag   OperationTag
	Ordinal        uint8
	// IdentityRevision carries K on dispatched outcomes (plan-canonical, copied
	// from scheduler.Attempt). Values built by the synthetic caller/WS/retry
	// constructors are a placeholder (1): those outcomes never dispatched through
	// a plan attempt, feed only the observation/wire surface, and do not take part
	// in the continuation (I,K) comparison (contPin compares the plan-canonical
	// Attempt), so the placeholder cannot become a wrong judgement.
	IdentityRevision  IdentityRevision
	Lane              LaneID
	Generation        Generation
	Commit            CommitState
	Result            AttemptResult
	HTTPStatus        AttemptStatus
	Timing            AttemptTiming
	Usage             AttemptUsage
	PreviousAttemptID *AttemptID
	Terminal          bool
	HardContinuation  bool
	BusinessFrameSent bool
	IsMalformed       bool
}

func (AttemptOutcome) HasPossiblyWrittenBytes 

func (o AttemptOutcome) HasPossiblyWrittenBytes() bool

func (AttemptOutcome) IsCountedForQuality 

func (o AttemptOutcome) IsCountedForQuality() bool

func (AttemptOutcome) IsDispatched 

func (o AttemptOutcome) IsDispatched() bool

func (AttemptOutcome) IsFailed 

func (o AttemptOutcome) IsFailed() bool

func (AttemptOutcome) Validate 

func (o AttemptOutcome) Validate() error

type AttemptResult 

type AttemptResult int
const (
	ResultUnknown AttemptResult = iota
	ResultSuccess
	ResultFailed
	ResultClientCancel
	ResultLocalReject
	ResultReservationReject
)

func (AttemptResult) String 

func (r AttemptResult) String() string

func (AttemptResult) Valid 

func (r AttemptResult) Valid() bool

type AttemptStatus 

type AttemptStatus int

type AttemptTiming 

type AttemptTiming struct {
	LatencyMS int64
	TTFTMS    *int64
}

type AttemptUsage 

type AttemptUsage struct {
	InputTokens         int64
	OutputTokens        int64
	CacheReadTokens     int64
	CacheCreationTokens int64
	CallCount           int64
}

type Auth 

type Auth struct {
	// contains filtered or unexported fields
}

Auth 鉴权快照：key_raw（明文）→ KeyMeta（含归属用户门禁字段）+ 用户快照表 （status+role）+ 两级并发/额度内存计数（gate）。热路径零 DB、零 per-request 锁（RWMutex 读多写少，规格 §10.3）。用户变更（禁用/降权/并发/额度调整） 走 invalidate 回调 → Reload 全量刷新，JWT 24h 长时效仅作快照 失效后的最终兜底。

func NewAuth 

func NewAuth(loader KeyLoader, users UserStatusLoader, mgmtLoader ManagementKeyLoader, log *logx.Logger, quotaEnabled bool) *Auth

NewAuth 构造鉴权快照（空表——首载统一由快照注册表 ReloadAll 承担，单一启动 入口，消灭"构造即载 + 注册表再刷"双重加载冗余；构造到首刷之间无请求流量， 见 main 装配序）。

func (*Auth) Acquire 

func (a *Auth) Acquire(meta domain.KeyMeta) (int, bool)

Acquire 两级并发门禁：user → key 依次 CAS 抢占；key 失败回滚 user 计数 （防泄漏）。返回已 acquire 层级位掩码（release 仅释放已 acquire 层级）。未设置上限（max=0）或计数器缺失（跨 reload 竞态窗口）→ 该层跳过。

func (*Auth) Authenticate 

func (a *Auth) Authenticate(r *http.Request) (domain.KeyMeta, bool)

Authenticate 解析网关 key 并返回 KeyMeta。兼容两种客户端口径： OpenAI 客户端发 Authorization: Bearer；Anthropic 官方 SDK / Claude Code 发 x-api-key 头。两者同时提供时以 Authorization 为准。 key 或归属用户被禁用 → 快照直接拒绝（401，即时失效）。 快照 map key = 明文，等值直查（零哈希）；meta 无 key 字符串字段—— 鉴权失败日志天然不落明文。

func (*Auth) AuthenticateManagement 

func (a *Auth) AuthenticateManagement(r *http.Request) (domain.ManagementKeyMeta, bool)

AuthenticateManagement 解析管理面凭证并返回管理 key 快照条目。仅认 Authorization: Bearer，**要求前缀 mk-**（无前缀直接 false）。前缀先判、失败 不回退：token 以 mk- 开头但查表失败/禁用 ⇒ 直接 false（中间件 401），绝不 回落 JWT 校验（否则前缀判别不成立，A3 由此时立）。owner 状态由中间件经 UserSnapshot 统一校验，fail-closed。

func (*Auth) DeductQuota 

func (a *Auth) DeductQuota(keyID, cost int64) int64

DeductQuota 请求结束扣减（后扣模型；usage 已知；无额度 key 无计数器 → no-op）。

func (*Auth) Delete 

func (a *Auth) Delete(raw string)

func (*Auth) DeleteManagementKey 

func (a *Auth) DeleteManagementKey(raw string)

DeleteManagementKey 增量移除单个管理 key（删除后本实例即时 401）。

func (*Auth) InFlightUsers 

func (a *Auth) InFlightUsers() map[int64]int64

InFlightUsers 门禁在途并发只读快照（/api/admin/users-top 端点用；spec 2026-08-14 gateSnapshot.users 未导出，经本访问器只读暴露）：gateSnapshot 整体 原子换入换出（reload/upsert 重建，不可变），store.Load() 零锁取当前引用后 遍历 + 原子读各计数器 → map[int64]int64 拷贝（含 0——过滤由调用方做）。 冷面调用（管理端聚合，不涉请求热路径）；多实例部署下为本实例在途计数。

func (*Auth) QuotaExhausted 

func (a *Auth) QuotaExhausted(meta domain.KeyMeta) bool

QuotaExhausted 额度检查：本地预算快读（零锁零 DB）；预算耗尽触发 DB 复核 认领（§3.2——复核成功续预算继续放行，复核确认真尽才 429）。检查在并发 acquire 之前（评审提醒①：失败无并发槽副作用）；未设置额度 key 短路零成本。

func (*Auth) Release 

func (a *Auth) Release(meta domain.KeyMeta, level int)

Release 释放并发计数（仅释放 acquire 返回的层级；跨 reload 命中新快照的 继承计数，与 scheduler Release 同语义）。

func (*Auth) Reload 

func (a *Auth) Reload(ctx context.Context) error

Reload 全量刷新鉴权快照（注册表首刷/周期 auth-sync/用户变更 invalidate）： keys 元数据 + 用户状态 + 门禁计数器（在途值跨 reload 继承）。 失败必打 Warn（含调用方是否忽略错误——invalidate 回调等吞错路径）：加载 失败若被忽略，快照保持旧值/空表 → 鉴权全部 401 或用旧 key 放行，静默恶化 （IN 超限事故的"运行中静默失败"形态即此类）。

func (*Auth) RemoveUser 

func (a *Auth) RemoveUser(userID int64)

RemoveUser 增量移除用户（暂未使用，防御性；与 UpsertUser 对称）。

func (*Auth) SetInstancesProvider 

func (a *Auth) SetInstancesProvider(p InstancesProvider)

SetInstancesProvider 注入集群实例数 N 提供者（多实例预算分摊；discovery 装配——main 装配点，spec 2026-08-25-redis-instance-discovery-design §2.2）。 注入即触发预算重算（幂等 reload，在途值继承）；此后 N 在每次预算分配现读， 心跳计数变化 ≤1 tick 天然生效。

func (*Auth) Upsert 

func (a *Auth) Upsert(raw string, meta domain.KeyMeta)

Upsert 增量刷新单个 key（key 创建/轮换/更新后调用；门禁计数器同步）。

func (*Auth) UpsertManagementKey 

func (a *Auth) UpsertManagementKey(raw string, meta domain.ManagementKeyMeta)

UpsertManagementKey 增量刷新单个管理 key（创建/启用/禁用后调用；禁用即时 401——查表按 Status==active 门控）。懒初始化 map（loader 未注入时也安全）。

func (*Auth) UpsertUser 

func (a *Auth) UpsertUser(userID int64, snap domain.UserSnapshot)

UpsertUser 增量刷新单个用户状态（本地立即可见，不等去抖窗口）。 供 admin 创建用户 / 注册 / 状态变更后本地实例立即对 RequireJWT 可见， 消除 200ms 窗口内新建用户 401；远端实例仍经 NOTIFY → 全量 Reload 收敛。

func (*Auth) UserSnapshot 

func (a *Auth) UserSnapshot(userID int64) (domain.UserSnapshot, bool)

UserSnapshot 用户快照（RequireJWT 状态校验 + adminAuth 快照 role 校验共用； 用户变更走 invalidate → Reload，不用 DB 直查）。单次查找同时取 status+role （热路径零分配）。

type BillingHooks 

type BillingHooks struct {
	Resolver   PriceResolver
	Balances   *billing.Balances // 余额只读快照（预检 + 扣费后定向刷新）
	Flusher    *billing.Flusher  // 批量扣费落库（billed 路由终点）
	TierPolicy func(tier billing.Tier) billing.TierPolicyMode
}

BillingHooks 计费钩子（proxy.New 参数；nil = 计费全关：不查价、不记 BillingTier、不处理 service_tier 转发策略、不做余额预检）。

type CallerCategory 

type CallerCategory string
const (
	CallerChat        CallerCategory = "chat"
	CallerResponses   CallerCategory = "responses"
	CallerAnthropic   CallerCategory = "anthropic"
	CallerConverted   CallerCategory = "converted"
	CallerImages      CallerCategory = "images"
	CallerImagesCodex CallerCategory = "images_codex"
	CallerCodexHTTP   CallerCategory = "codex_http"
	CallerResponsesWS CallerCategory = "responses_ws"
	CallerCodexWS     CallerCategory = "codex_ws"
	CallerSearch      CallerCategory = "search"
)

func AllCallerCategories 

func AllCallerCategories() []CallerCategory

func (CallerCategory) Valid 

func (c CallerCategory) Valid() bool

type CandidateFingerprint 

type CandidateFingerprint string

type CommitState 

type CommitState int
const (
	CommitNotSent CommitState = iota
	CommitSentAmbiguous
	CommitUpstreamResponded
	CommitResponseStarted
	CommitClientCommitted
)

func (CommitState) String 

func (c CommitState) String() string

func (CommitState) Valid 

func (c CommitState) Valid() bool

type ConcSyncWorker 

type ConcSyncWorker struct {
	// contains filtered or unexported fields
}

ConcSyncWorker 并发门双向同步 worker：实现 worker.Worker。客户端经 main 注入 （pkg/redisx 单构造点纪律，本包不自建连接）；gate 引用取自 Auth（同包直访 私有字段，无需导出接口）。nil client = Start no-op（测试/降级形态零专门分支）。

func NewConcSyncWorker 

func NewConcSyncWorker(auth *Auth, client *redis.Client, self string, log *logx.Logger) *ConcSyncWorker

NewConcSyncWorker 构造（零副作用：Start 才起 goroutine，Close 未 Start 时安全）。

func (*ConcSyncWorker) Close 

func (w *ConcSyncWorker) Close(ctx context.Context) error

Close 幂等优雅停机：停 tick 即完事（spec §1.5 协调态可丢、无排空义务—— 在途字段由 ts 新鲜度 ≤4s 出局、整键 EXPIRE 自灭，无清理命令）。

func (*ConcSyncWorker) Name 

func (w *ConcSyncWorker) Name() string

Name worker 名（worker.Worker 契约）。

func (*ConcSyncWorker) Start 

func (w *ConcSyncWorker) Start(_ context.Context) error

Start 非阻塞启动同步循环（幂等；自持 ctx，同 discovery baseCtx 惯用法）。 nil client 直接短路：无视图装配能力即全额本地语义，循环无存在意义。

func (*ConcSyncWorker) Stats 

func (w *ConcSyncWorker) Stats() any

Stats worker 观测（协调面冻结可见性）：fail-open 静默退化时这是运维面唯一 痕迹——视图停止换入则 last_tick_ok 翻 false、consecutive_errors 增长。

type Config 

type Config struct {
	MaxBodySize           int64
	UpstreamTimeout       time.Duration // codex 非流式上游超时（resp/images 各自包 ctx——HTTPClient.Timeout 不可用：流式/非流式四方法共享，覆盖整响应体读取会切断长流式 SSE）。同源同值 cfg.Proxy.UpstreamTimeout（aiclient.Config.UpstreamTimeout 管 typed 面）
	UpstreamStreamTimeout time.Duration // 流式 backstop（非流式超时在 aiclient.Config/cfg.Proxy.UpstreamTimeout）
	FailoverAttempts      int
	UsageCapture          bool
	BillingCapture        bool // 计费开关（config.Billing.Enabled 映射；余额预检门控 + billable 行 Billed 出生标记取反——单写点后不再路由分流）
	// BehindCDN 客户端 IP 识别开关（config.proxy.behind_cdn 映射；默认 true——
	// clientIP 提取门控：true 按序采信供应商头，false 完全不读直取 RemoteAddr）。
	// 部署前提见 config.go 注释与 clientip.go：源站只对 CDN/反代暴露时方可采信。
	BehindCDN bool
	// LowBalanceConcCap 低余额用户级并发上限（config.proxy.low_balance_max_concurrency
	// 映射；0 = 关闭该特性——guardPipeline 保持逐位等价现状）。
	LowBalanceConcCap int
	// LowBalanceConcThresholdMilli 触发阈值（毫分；config.proxy.low_balance_threshold_usd
	// × 1e5 取整）。余额快照存在且严格小于它时把用户级上限钳为 LowBalanceConcCap。
	LowBalanceConcThresholdMilli int64
	// StreamKeepaliveInterval SSE 客户端保活注释间隔（config.proxy.stream_keepalive_interval
	// 映射；默认 10s，0 = 关闭通用定时心跳）。五路 + images 经 Output 的保活由它
	// 驱动（images 自身 SDK 保活不受影响）。
	StreamKeepaliveInterval time.Duration
}

type ContBindConfig 

type ContBindConfig struct {
	QueueSize    int           // 有界队列容量（背压面：满 → 丢弃计数）
	Workers      int           // 并发消费 goroutine 数
	BatchSize    int           // 每批最多条数
	BatchWait    time.Duration // 从首条入批起的最大等待（凑批）
	BatchTimeout time.Duration // 每批 Redis 操作超时
}

ContBindConfig 异步绑定 worker 构造参数（零值字段取内部默认）。

type ContBindWorker 

type ContBindWorker struct {
	// contains filtered or unexported fields
}

ContBindWorker 异步绑定 worker（worker.Worker + handler.StatsProvider 契约， Name="cont-bind"）。

func NewContBindWorker 

func NewContBindWorker(store *continuation.Store, log *logx.Logger, cfg ContBindConfig) *ContBindWorker

NewContBindWorker 构造异步绑定 worker；store 为绑定存储（须非 nil）， log 可 nil（静默）。

func (*ContBindWorker) Bound 

func (w *ContBindWorker) Bound() int64

Bound 成功落库（created/refreshed）计数（观测/测试）。

func (*ContBindWorker) Close 

func (w *ContBindWorker) Close(ctx context.Context) error

Close 幂等排空：置位 closed（此后 Enqueue 丢弃计数）→ 等 loop 退出（受 ctx 预算约束）→ 排空剩余（每批 ≤ BatchSize，受 ctx 预算约束）。预算到期时：**已 在手批次按 dropped 计**（未提交 Redis，等价于队列剩余丢弃——与已发给 Redis 的 批次失败计 failed 区分），并对剩余条数 Warn 一次（对齐 errlog 停机截断先例）。 未 Start 也安全（跳过 loop 等待直接排空）。

func (*ContBindWorker) Conflicts 

func (w *ContBindWorker) Conflicts() int64

Conflicts CAS conflict 计数（保留旧绑定；不等于下次必然 410——Lookup 可能 命中旧记录）。

func (*ContBindWorker) DropUnattributed 

func (w *ContBindWorker) DropUnattributed()

DropUnattributed 记一条可归属拒绝（contEnqueue 在写出接缝取得 id 后，因缺少 loop 派发观测/请求元数据或指纹不可解码而未写 Redis）。

func (*ContBindWorker) Dropped 

func (w *ContBindWorker) Dropped() int64

Dropped 队列满/已关闭丢弃计数（背压观测）。

func (*ContBindWorker) Enqueue 

func (w *ContBindWorker) Enqueue(req continuation.BindRequest)

Enqueue 非阻塞有界投递：队列满或已关闭 → 丢弃 + 计数（绝不阻塞请求热路径）。

func (*ContBindWorker) Failed 

func (w *ContBindWorker) Failed() int64

Failed 批量失败/逐条 I/O 错误计数（观测；不影响当前响应）。

func (*ContBindWorker) Name 

func (w *ContBindWorker) Name() string

Name worker.Worker 契约。

func (*ContBindWorker) Queued 

func (w *ContBindWorker) Queued() int

Queued 当前队列积压条数（背压观测）。

func (*ContBindWorker) Start 

func (w *ContBindWorker) Start(ctx context.Context) error

Start 非阻塞启动消费 goroutine（worker.Loop 托管，panic 不崩进程）。幂等： 二次 Start 返回错误。

func (*ContBindWorker) Stats 

func (w *ContBindWorker) Stats() any

Stats 满足 handler.StatsProvider（观测面：队列/落库/丢弃/失败/冲突/可归属拒绝）。

func (*ContBindWorker) Unattributed 

func (w *ContBindWorker) Unattributed() int64

Unattributed 可归属拒绝计数（缺 dispatch/reqMeta/指纹；未装配路径不计数）。

type Deps 

type Deps struct {
	// Codex codex SDK 适配层（nil = 未装配 → codex 类型请求 501）。
	Codex *sdkbridge.Codex
	// Recorder 质量记录器（休眠依赖；nil = 未装配）。
	Recorder *quality.Recorder
	// Continuation 硬续接绑定存储（nil = 未装配 → continuation 请求 fail-closed）。
	Continuation *continuation.Store
	// ContBind REST 流式异步绑定 worker（nil = 未装配 → REST 流式绑定 no-op，
	// 未装配形态零行为变化；与 Continuation 同处注入）。
	ContBind *ContBindWorker
}

Deps New 的尾部一次性协作者（SetCodex / SetQualityRecorder / SetContinuationStore 三个事后回填折叠进构造，零语义变化——各字段 nil 语义 与原 setter 完全一致：Codex nil → codex 类型请求 501 显式拒绝；Recorder nil → 休眠依赖；Continuation nil → continuation 请求 fail-closed）。尾部 struct 而非 3 个位置参数：New 本就 9 参，位置参数会冲到 12 个（>3 参 smell），具名字段自文档且调用点可只填所需（Deps 同款）。

type Generation 

type Generation int64

type IdentityRevision 

type IdentityRevision int64

IdentityRevision is the account identity generation (K) carried by an attempt: the same K the plan attempt was compiled with. It is NOT the lifecycle token (C), which only fences client writes.

type InstancesProvider 

type InstancesProvider interface {
	ClusterInstances() int
}

InstancesProvider 集群实例数 N 提供者（discovery.Discovery 实现 ClusterInstances； 多实例预算分摊 §3.1——N = Redis 心跳活体数，spec 2026-08-25-redis-instance-discovery-design；原 DB settings 手工设置已删）。 nil（未装配）按 N=1（单实例语义）。N 在每次预算分配现读（instancesN），心跳 计数变化 ≤1 tick 天然生效。

type KeyLoader 

type KeyLoader interface {
	LoadKeys(ctx context.Context) (map[string]domain.KeyMeta, error)
}

KeyLoader 由 repository.KeyRepo 实现（keys 独立表鉴权快照）。

type LaneID 

type LaneID string
const (
	LanePrimary  LaneID = "primary"
	LaneExplore  LaneID = "explore"
	LaneDegraded LaneID = "degraded"
)

func (LaneID) Valid 

func (l LaneID) Valid() bool

type ManagementKeyLoader 

type ManagementKeyLoader interface {
	LoadManagementKeys(ctx context.Context) (map[string]domain.ManagementKeyMeta, error)
}

ManagementKeyLoader 由 repository.ManagementKeyRepo 实现（管理 key mk- 鉴权快照数据源，spec 2026-10-09）。

type OperationTag 

type OperationTag string

type PriceResolver 

type PriceResolver interface {
	ResolvePrices(model string, promptTokens int64, tier string, at time.Time) (domain.ResolvedPrices, bool)
}

PriceResolver 统一价格解析（零 DB 快照读，首中即停变体解析）。

type Proxy 

type Proxy struct {
	// contains filtered or unexported fields
}

func New 

func New(cfg Config, sched *scheduler.Scheduler, creds *credential.Registry, rec *usage.Recorder, clients *aiclient.Factory, auth *Auth, log *logx.Logger, bill *BillingHooks, errlog *usage.ErrLogWorker, deps Deps) *Proxy

New 构造代理。creds 为凭据注册表（直接参数注入，编译期强制； 不用 Config 字段——避免 nil 运行时才炸）。bill 为计费钩子（； nil = 计费全关——现有调用点/测试兼容）。errlog 为错误明细落盘 worker （分表设计；nil = 未装配——拒绝/异常路径只聚统计不落 err_logs 明细）。 deps 为尾部一次性协作者（零值 = 三者皆未装配，各 nil 语义见 Deps）。

func (*Proxy) CloseAllWS 

func (p *Proxy) CloseAllWS()

CloseAllWS closes all hijacked WS client connections. Uses CloseNow for immediate TCP close — shutdown path, no handshake. Closed sessions unwind through existing classify→finish→rec.Record and inflight drops naturally. Idempotent.

func (*Proxy) HandleAnthropic 

func (p *Proxy) HandleAnthropic(w http.ResponseWriter, r *http.Request)

HandleAnthropic 转发 /v1/messages（anthropic 格式）。全部逻辑在 通用骨架 handleFormat + anthropicCaller（转发骨架通用化），本方法 只是端点入口委托。测试按处理函数直接调用（forward_ext_test.go 等），故入口保留。

func (*Proxy) HandleChat 

func (p *Proxy) HandleChat(w http.ResponseWriter, r *http.Request)

HandleChat 转发 /v1/chat/completions（openai-chat 格式）。全部逻辑在 通用骨架 handleFormat + chatCaller（转发骨架通用化），本方法 只是端点入口委托。测试按处理函数直接调用（proxy_test.go 等），故入口保留。

func (*Proxy) HandleImagesEdits 

func (p *Proxy) HandleImagesEdits(w http.ResponseWriter, r *http.Request)

HandleImagesEdits 转发 POST /v1/images/edits（openai-images 格式；上游子路径 由 handleFormat 内 imagesCallerFor 按请求路径选择）。

func (*Proxy) HandleImagesGenerations 

func (p *Proxy) HandleImagesGenerations(w http.ResponseWriter, r *http.Request)

HandleImagesGenerations 转发 POST /v1/images/generations（openai-images 格式； 全部逻辑在 handleFormat + imagesCaller——端点入口委托，同 HandleChat 形态）。

func (*Proxy) HandleModels 

func (p *Proxy) HandleModels(w http.ResponseWriter, r *http.Request)

HandleModels GET /v1/models：OpenAI 兼容模型列表（OpenAI SDK / codex 等 客户端用网关 key 拉取可用模型）。数据源 = 调度器内存快照（零 DB）： key → meta.GroupID → 组快照 routes → 模型去重排序（scheduler.GroupModels）。 端点冷面（非转发热路径）：鉴权通过即放行——不走计费/限流/并发门禁（只读 列表端点，OpenAI 语义——声明于注释）；不建新缓存/新快照。空组/无模型 → 空 data 数组（200 不 404）；组不存在/快照未加载 → 404（对齐 Select 的 ErrGroupNotFound 语义——鉴权已过但组失效）。

func (*Proxy) HandleResponses 

func (p *Proxy) HandleResponses(w http.ResponseWriter, r *http.Request)

HandleResponses 转发 /v1/responses（openai-responses 格式）。全部逻辑在 通用骨架 handleFormat + responsesCaller（转发骨架通用化），本方法 只是端点入口委托。测试按处理函数直接调用（forward_ext_test.go 等），故入口保留。

func (*Proxy) HandleResponsesWS 

func (p *Proxy) HandleResponsesWS(w http.ResponseWriter, r *http.Request)

HandleResponsesWS 处理 resp-ws 升级请求（/v1/responses 带 upgrade 头—— 真实客户端无 /ws 后缀，WS 与 POST /v1/responses 同路径，按协议分流）。 与 handleFormat 同构：guardPipeline（鉴权 → 额度/余额预检 → 两级并发门禁 → 限流，见 pipeline.go）→ 升级 → 首帧（= 请求体）→ 模型提取 → 选号 → failoverLoop（wsAttempt + wsSink，precheck=true）→ 双向 relay（usage 嗅探） → 记录。差异段留本文件：

  • 无 HTTP body：请求体 = 升级后首个 WS 帧（response.create）
  • 选号在首帧之后（模型来自首帧；挂死不占账号槽）
  • 本地拒绝在升级后无 HTTP 状态码 → 错误事件帧承载（wsWriteError）
  • 门禁覆盖整个长会话：guard 释放 defer 到 relay 结束（同现状语义）

func (*Proxy) HandleSearch 

func (p *Proxy) HandleSearch(w http.ResponseWriter, r *http.Request)

HandleSearch 转发 codex /v1/alpha/search：codex CLI 以 独立 unary POST 调 web search（模型发 web.run tool call 时触发，与主 /responses 流并发）。透传语义：请求体/响应体原样（opaque results/ encrypted_output 网关零解析——alpha 端点实验性，上游变更网关免疫）。

与主 handleFormat 的差异（search 专属语义）：

  • 账号选择：body.model → selectWithPlan(groupID, openai-responses, model, ApplyModelMapping=false)（复用主流 resp 路由面——四类型全可达；独立选号 无会话绑定：search 请求自包含，上游鉴权 = 有效 Bearer，无会话亲和机制）
  • 不走计费预检（余额/缺价 402 均不执行——search 无预检语义；按次价在 2xx 落账时结算，零余额透支扣费为产品语义）
  • 四类型分派（用户裁决 2026-08-13）：codex-oauth/codex-pat → codex-sdk Search（适配层 clientFor 缓存客户端直接复用；固定 SDK 官方端点 https://chatgpt.com/backend-api/codex/alpha/search，网关零拼装， test transport 仅 host 重写保留官方 path；Auth 注入/刷新/fatal 生命周期 复用既有 SDK 面）；api_key/responses-special → 静态透传（Bearer upstream key 直连上游；URL 裸根派生 base/v1/alpha/search，固定 /v1/alpha/search）。组内混合类型路由允许
  • x-codex-turn-metadata：静态路径如实透传，SDK 路径不转发——静态 rawPostCT 现收 r.Header 过 RelayHeaders，该键不在全仓唯一清单 relayDeny ⇒ 客户端带就发（与主流静态路径同源）；SDK 路径结构上不采客户端头
  • 不做 ModelMapping 改写：请求体原样 = 映射对 search 不生效（上游收客户端模型名）——零解析是显式约束
  • 计费：2xx → usage_logs 行（format=openai-search + call_count=1 + price_per_call_millis=PriceResolver call 档（codex-search 模型） + cost=按次价×整单 倍率）；非 2xx/网络错误 → 不计费

复用面：guardPipeline（鉴权/配额/并发门禁/限流序列）、 selectWithPlan + handleSelectError、信封分类（statusOf/upstreamBody）、 failoverLoopWithPlan （每轮按当轮 sel.CredentialType 重新分派——跨类型换账号复用旧调用器会把健康账号路由到错误凭据路径）、 recordRejected/finish/buildLog/MarkResult 全部既有机制。

func (*Proxy) Inflight 

func (p *Proxy) Inflight() int64

func (*Proxy) QualityRecorder 

func (p *Proxy) QualityRecorder() *quality.Recorder

QualityRecorder returns the dormant quality recorder (may be nil in tests).

func (*Proxy) SetInstancesProvider 

func (p *Proxy) SetInstancesProvider(inst InstancesProvider)

SetInstancesProvider 注入集群实例数 N 提供者（多实例预算分摊；discovery 构造后调用——main 装配点：px.SetInstancesProvider(disco)，spec 2026-08-25-redis-instance-discovery-design §2.2）。转发给 auth（gate 预算 ceil(剩余/N)）；N 在每次预算分配现读，心跳计数变化 ≤1 tick 天然生效。

func (*Proxy) SetSupplierSnapshot 

func (p *Proxy) SetSupplierSnapshot(s *supplier.SupplierSnapshot)

SetSupplierSnapshot 注入供应商视图快照（main 装配；nil = 未装配/关闭态—— 热路径字段保持出生定态 credited=true）。归属写入/刷新由 main 的 ticker 触发 Store（§6.4）。快照类型归属 internal/supplier（ViewSink/准入同处）。

func (*Proxy) SupplierSnapshotRef 

func (p *Proxy) SupplierSnapshotRef() *supplier.SupplierSnapshot

SupplierSnapshotRef 返回已装配的供应商快照（nil = 未装配；ops 面读取）。

type QualityClassID 

type QualityClassID string

type QuotaUsedReader 

type QuotaUsedReader interface {
	QuotaUsed(ctx context.Context, keyID int64) (int64, error)
}

QuotaUsedReader 预算复核的 DB 只读接口（repository.KeyRepo 实现）：预算耗尽 时读 key 当前已用额度（DB 权威值——usage.Recorder 批量增量回写 quota_used， 复核时刻分配预算用）。热路径不调（仅复核慢路径）。

type RouteClassID 

type RouteClassID string

type ToolView 

type ToolView struct {
	Raw       []byte
	Type      []byte
	Name      []byte
	Namespace []byte
	Result    []byte
	ID        []byte
}

ToolView 工具视图：Raw 为 body 子切片（真零拷贝——引用 tools 值区间字节）； Type/Name/Namespace 为顶层键值（去引号、\uXXXX 已解码；未提取 = 空）。 Result/ID 为 resp 响应检测旁路（spec §6）复用提取：ID = id 字符串值； Result = result 值裸字节（字符串值去引号/\u 解码；非字符串值——数组/null/ 数字——取原始值区间，空语义在判定处 imageResultNonEmpty 处理）。

type UpstreamCaller 

type UpstreamCaller interface {
	Call(ctx context.Context, w http.ResponseWriter, r *http.Request, reqID string, groupID int64,
		start time.Time, sel *scheduler.Selection, cred string, body []byte, stream bool) (code int, respBody []byte, handled bool, err error)
}

UpstreamCaller 一格式一实现：完成单次上游调用（含流式写出、客户端断开判定 与 usage 记录）。记录职责全在 caller（finish/buildLog/recordStreamAbort/ MarkResult 直接可用）；骨架只做 code 分支（429/5xx 转移、4xx 透传记录）、handled 短路与耗尽 record。凭据值经 aiclient 格式方法传入 （头名 aiclient 内组装，正交延续）。

语义：

  • handled == true → 请求已处理完毕（成功/客户端断开/流中止已记录；本地拒绝 已写出无记录），骨架直接 return（不可转移）
  • handled == false → 上游未接受，骨架接手： code 429 → MarkResult(Kind429) + Release + 转移 code >= 500 或 code == 0（连接级/凭据错）→ MarkResult(RuleKindOf(code)) + Release + 转移 code 4xx（err == nil）→ 骨架 finish(buildLog(Err4xx)) + 透传 respBody （空 → 网关文案 "upstream rejected request"）
  • err 非 nil 仅在错误路径返回（分类由 code 承载）；骨架用它提取错误文本 （部署故障修复）：code==0 → err.Error() 落 ErrorMessage/last_error + Warn（err 全文），4xx → respBody 原文落 ErrorMessage。成功路径 err 恒 nil（零新增分配）。
  • 例外（首字节前客户端断连，分类正确性）：code==0 且 r.Context().Err()!=nil （客户端已断开）→ 记 499+ErrAbort 立即返回——不 failover、不 MarkResult、 不冷却（否则连接级误分类把无辜账号冷却 + failover 空转）。

type UserStatusLoader 

type UserStatusLoader interface {
	LoadUsers(ctx context.Context) (map[int64]domain.UserSnapshot, error)
}

UserStatusLoader 由 repository.UserRepo 实现（RequireJWT 用户状态 + adminAuth 快照 role 校验的数据源）。

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.