diff --git a/.env.example b/.env.example index f55d0df..161ff41 100644 --- a/.env.example +++ b/.env.example @@ -56,6 +56,8 @@ OT_PROXY_HEALTH_INTERVAL=60s OT_PROXY_HEALTH_FAIL_THRESHOLD=2 # 调试:记录管理员的原始请求与响应到请求明细(默认关闭;流式记录全部事件) +# 记录管理员请求的原始请求体/响应体到 usage_logs(调试用)。 +# 原始体会截断:请求体 256 KiB、响应体 1 MiB,避免单条日志几十 MB 拖慢查询与备份。 OT_PROXY_LOG_RAW=false # 限流(内存计数,Redis 后置):用户级每秒请求数上限(0=不限制) diff --git a/README.md b/README.md index 3ea85f9..5ad6f77 100644 --- a/README.md +++ b/README.md @@ -13,7 +13,7 @@ - `GET /v1/models` — 可用模型列表:只列调用方真能用的模型(叠加用户与密钥白名单),带上下文长度与能力位;Anthropic 客户端(带 `x-api-key`/`anthropic-version`)返回 Anthropic 形状 - **三协议互转**:客户端协议 × 渠道协议不匹配时自动转换(如 Chat 调用 Claude、Messages 调用 OpenAI、Responses 调用 Claude),流式逐事件转换;协议匹配时直通 - 错误按客户端协议返回(OpenAI 格式 / Anthropic 格式) -- **热路径开销**:模型/渠道目录进程内缓存(每请求 SQL 从 10 条降到 2 条,管理端写操作与健康状态变化即时失效)、流式正文按字节快路径提取、请求体定点改写(模型映射不再整包 unmarshal/marshal)、tiktoken 编码器复用 +- **热路径开销**:模型/渠道目录进程内缓存(每请求 SQL 从 10 条降到 2 条,管理端写操作与健康状态变化即时失效)、流式正文按字节快路径提取、请求体定点改写(模型映射不再整包 unmarshal/marshal)、tiktoken 编码器复用、会话亲和表 16 片无锁竞争(键为 64 位摘要、零分配)、访问日志走 64 KiB 缓冲(不再每请求一次 stderr 系统调用);记账落库按驱动自适应并发(SQLite 单 worker,PostgreSQL 4 个分片 worker) - **渠道体系**:按模型绑定选渠道 + 加权负载均衡;模型已有绑定时只用绑定渠道(都不可用则 503,不会把请求发给不认识它的渠道);每渠道并发信号量(满载排队 `OT_PROXY_CHANNEL_ACQUIRE_WAIT`,默认 150ms,仍满则 503 + Retry-After 而不是 502);后台健康检查(探测 `/models`,连续失败进 cooldown、恢复放回)+ 运行期健康(按失败原因分类冷却:网络错误/5xx 连续 3 次进冷却且指数退避,余额/配额类第一次即冷却 30 分钟,成功清零退避计数);可安全重试的失败自动故障转移(网络错误/429/5xx/超时/流式首个内容事件之前的任何失败);**会话亲和**(`OT_PROXY_AFFINITY_TTL`,默认 5m):同一会话粘住上次应答的渠道,命中上游提示词缓存而不是全价重算,冷却/下线自动回退加权随机 - **上游超时**:渠道 `timeout_ms` 生效——非流式为整段请求上限,流式仅约束首包(正文不限时长);上游请求绑定客户端 context,断连即取消 - **故障转移与流式细节**(借鉴 magpie 的 holdWriter):流式请求在**首个内容事件**到达前由网关缓冲——上游的错误事件、空流、断连、协议形态不符(要流给了整包,didNotStream)都在放行前换渠道消化,客户端无感知;`OT_PROXY_FIRST_TOKEN_WAIT` 限定放行前等内容的最长时间(0=不限);429 不一定是限流——错误体带余额/配额特征时按配额类长冷却;上游 `Retry-After` 会延长该渠道冷却(夹在 10 分钟内);重试受 `OT_PROXY_FAILOVER_BUDGET` 时间预算约束(首个渠道不受限,避免坏渠道把客户端拖到 N×timeout);流式空闲超过 `OT_PROXY_STREAM_HEARTBEAT` 补一行 SSE 注释心跳(扣住期间也照发),防止 nginx/Cloudflare 的空闲超时掐掉长思考;响应头已被心跳先行写出时,全部候选失败以 SSE 错误事件收尾 diff --git a/server/internal/channel/channel.go b/server/internal/channel/channel.go index d956e74..31ee145 100644 --- a/server/internal/channel/channel.go +++ b/server/internal/channel/channel.go @@ -4,10 +4,9 @@ package channel import ( "context" - "crypto/rand" "errors" "log" - "math/big" + mrand "math/rand/v2" "sync" "time" @@ -251,11 +250,13 @@ func (s *Service) Pick(cands []Candidate) *store.Channel { for _, c := range pool { total += pickWeight(c) } - n, _ := rand.Int(rand.Reader, big.NewInt(int64(total))) + // 负载均衡不需要密码学随机:math/rand/v2 的 IntN 无锁、零分配 + // (原来是 crypto/rand + big.Int,每请求一次系统调用加两次分配)。 + n := mrand.IntN(total) acc := 0 for _, c := range pool { acc += pickWeight(c) - if int(n.Int64()) < acc { + if n < acc { return c.Channel } } diff --git a/server/internal/proxy/affinity.go b/server/internal/proxy/affinity.go index 0594f8d..3bca6a1 100644 --- a/server/internal/proxy/affinity.go +++ b/server/internal/proxy/affinity.go @@ -1,9 +1,6 @@ package proxy import ( - "crypto/sha256" - "encoding/hex" - "fmt" "sync" "time" ) @@ -12,24 +9,33 @@ import ( // 同一会话粘住上次应答的渠道,让上游的提示词缓存(整段上下文每轮重发)被再次 // 命中,而不是被加权随机打散后全价重算。 // -// 会话身份 = 密钥 + 模型 + 请求文本稳定前缀的摘要。编码代理每轮重发全量上下文, -// 系统提示与最早的用户消息不变,取前缀即可区分会话;不需要解析各家客户端的 -// 会话头。命中亲和的渠道仍必须在候选集里(健康、未冷却、认识该模型)—— -// 冷却或下线后自然回退到加权随机。 +// 会话身份 = 密钥 + 模型 + 请求体稳定前缀的 64 位摘要。编码代理每轮重发全量上下文, +// 开头是稳定的(模型、系统提示、最早的几条消息),取前缀即可区分会话; +// 不需要解析各家客户端的会话头。命中亲和的渠道仍必须在候选集里(健康、未冷却、 +// 认识该模型)——冷却或下线后自然回退到加权随机。 // // TTL 默认 5 分钟(对齐 Anthropic 提示词缓存窗口),每次成功应答刷新; // 上游不报缓存的(OpenAI 自动前缀缓存)同样受益:粘住就是省。 +// +// 表按摘要分 16 片:每个请求都要查一次亲和,全局单锁会成为高并发下的串行点。 +// 键直接是 64 位摘要(而不是拼接的字符串),请求路径上零分配。 const ( - // affinityMax 亲和表上限:超过先清过期,仍超则不再新增(防御性,正常流量到不了)。 - affinityMax = 8192 - // affinityPrefix 会话身份摘要取请求文本的前缀字节数。 + // affinityShards 分片数(2 的幂,取低位做下标)。 + affinityShards = 16 + // affinityPerShard 每片的条目上限:超过先清过期,仍满则不再新增(防御性)。 + affinityPerShard = 8192 + // affinityPrefix 取请求体的前多少字节参与摘要。 affinityPrefix = 512 ) type affinityTable struct { + shards [affinityShards]affinityShard +} + +type affinityShard struct { mu sync.Mutex - m map[string]affinityEntry + m map[uint64]affinityEntry } type affinityEntry struct { @@ -38,47 +44,86 @@ type affinityEntry struct { } func newAffinityTable() *affinityTable { - return &affinityTable{m: map[string]affinityEntry{}} + t := &affinityTable{} + for i := range t.shards { + t.shards[i].m = map[uint64]affinityEntry{} + } + return t +} + +func (t *affinityTable) shard(key uint64) *affinityShard { + return &t.shards[key&(affinityShards-1)] } // get 返回该会话上次应答的渠道;过期条目顺手清理。 -func (t *affinityTable) get(key string) (uint64, bool) { - t.mu.Lock() - defer t.mu.Unlock() - e, ok := t.m[key] +func (t *affinityTable) get(key uint64) (uint64, bool) { + s := t.shard(key) + s.mu.Lock() + defer s.mu.Unlock() + e, ok := s.m[key] if !ok { return 0, false } if time.Now().After(e.until) { - delete(t.m, key) + delete(s.m, key) return 0, false } return e.channelID, true } // put 记录(或刷新)会话→渠道的绑定。 -func (t *affinityTable) put(key string, channelID uint64, ttl time.Duration) { - t.mu.Lock() - defer t.mu.Unlock() - if len(t.m) >= affinityMax { +func (t *affinityTable) put(key, channelID uint64, ttl time.Duration) { + s := t.shard(key) + s.mu.Lock() + defer s.mu.Unlock() + if len(s.m) >= affinityPerShard { now := time.Now() - for k, e := range t.m { + for k, e := range s.m { if now.After(e.until) { - delete(t.m, k) + delete(s.m, k) } } - if len(t.m) >= affinityMax { - return // 表满且全是活的:放弃本次记录,旧绑定继续服务 + if len(s.m) >= affinityPerShard { + return // 分片满且全是活的:放弃本次记录,旧绑定继续服务 } } - t.m[key] = affinityEntry{channelID: channelID, until: time.Now().Add(ttl)} + s.m[key] = affinityEntry{channelID: channelID, until: time.Now().Add(ttl)} } -// affinityKey 会话身份:密钥 + 模型 + 请求文本稳定前缀的摘要。 -func affinityKey(keyID uint64, model, firstText string) string { - if len(firstText) > affinityPrefix { - firstText = firstText[:affinityPrefix] +// len 当前条目数(测试与诊断用)。 +func (t *affinityTable) len() int { + n := 0 + for i := range t.shards { + t.shards[i].mu.Lock() + n += len(t.shards[i].m) + t.shards[i].mu.Unlock() } - h := sha256.Sum256([]byte(firstText)) - return fmt.Sprintf("%d|%s|%s", keyID, model, hex.EncodeToString(h[:8])) + return n +} + +// affinityKey 会话身份摘要:密钥 + 模型 + 请求体前 affinityPrefix 字节。 +// 内联 FNV-1a,避免在热路径上分配哈希器或拼接字符串。 +func affinityKey(keyID uint64, model string, prefix []byte) uint64 { + const ( + offset64 = 14695981039346656037 + prime64 = 1099511628211 + ) + h := uint64(offset64) + mix := func(b byte) { + h ^= uint64(b) + h *= prime64 + } + for i := 0; i < 8; i++ { + mix(byte(keyID >> (8 * i))) + } + for i := 0; i < len(model); i++ { + mix(model[i]) + } + if len(prefix) > affinityPrefix { + prefix = prefix[:affinityPrefix] + } + for _, b := range prefix { + mix(b) + } + return h } diff --git a/server/internal/proxy/affinity_test.go b/server/internal/proxy/affinity_test.go index e0ed42a..9d3607f 100644 --- a/server/internal/proxy/affinity_test.go +++ b/server/internal/proxy/affinity_test.go @@ -5,6 +5,7 @@ import ( "net/http" "net/http/httptest" "strings" + "sync" "sync/atomic" "testing" "time" @@ -75,7 +76,7 @@ func TestAffinityRespectsCooldown(t *testing.T) { // 预置亲和绑定指向渠道1,随后让渠道1 进运行期冷却 var ch1row store.Channel h.db.Where("name = ?", "c1").First(&ch1row) - h.gw.aff.put(affinityKey(1, "m1", "会话的第一条消息"), ch1row.ID, 5*time.Minute) + h.gw.aff.put(affinityKey(1, "m1", []byte("会话的第一条消息")), ch1row.ID, 5*time.Minute) for i := 0; i < 3; i++ { h.gw.ch.ReportFailure(&ch1row, channel.FailRate, 0) } @@ -85,3 +86,77 @@ func TestAffinityRespectsCooldown(t *testing.T) { t.Fatalf("冷却后亲和应失效并切换渠道2:%d %s", w.Code, w.Body.String()) } } + +// 亲和键:同输入稳定、任一要素不同即不同;请求路径上不分配。 +func TestAffinityKeyStable(t *testing.T) { + prefix := []byte(`{"model":"m1","messages":[{"role":"user","content":"hi"}]}`) + a := affinityKey(7, "m1", prefix) + if b := affinityKey(7, "m1", prefix); a != b { + t.Fatal("同一会话的键必须稳定") + } + if affinityKey(8, "m1", prefix) == a { + t.Fatal("不同密钥应得到不同键") + } + if affinityKey(7, "m2", prefix) == a { + t.Fatal("不同模型应得到不同键") + } + if affinityKey(7, "m1", []byte("other")) == a { + t.Fatal("不同前缀应得到不同键") + } + // 前缀只有前 affinityPrefix 字节参与:同一个开头、更长的尾部不影响键 + head := make([]byte, affinityPrefix) + for i := range head { + head[i] = byte('a' + i%26) + } + tailA := append(append([]byte{}, head...), []byte("AAAA")...) + tailB := append(append([]byte{}, head...), []byte("BBBB")...) + if affinityKey(7, "m1", tailA) != affinityKey(7, "m1", tailB) { + t.Fatal("超过前缀长度的部分不应影响键") + } + if affinityKey(7, "m1", head) != affinityKey(7, "m1", tailA) { + t.Fatal("尾部长短不同但前 512 字节相同,键应一致") + } + inner := append([]byte{}, head...) + inner[len(inner)-1] = 'z' // 改动前缀内最后一个字节 + if affinityKey(7, "m1", head) == affinityKey(7, "m1", inner) { + t.Fatal("前缀以内的差异必须影响键") + } +} + +func TestAffinityKeyNoAlloc(t *testing.T) { + prefix := []byte(`{"model":"m1"}`) + allocs := testing.AllocsPerRun(200, func() { _ = affinityKey(7, "m1", prefix) }) + if allocs != 0 { + t.Fatalf("affinityKey 分配了 %v 次,期望 0", allocs) + } +} + +// 分片表的过期与并发安全(-race 下验证)。 +func TestAffinityTableSharded(t *testing.T) { + tbl := newAffinityTable() + tbl.put(1, 10, 50*time.Millisecond) + tbl.put(2, 11, time.Minute) + if id, ok := tbl.get(1); !ok || id != 10 { + t.Fatalf("get(1) = %d/%v", id, ok) + } + if tbl.len() != 2 { + t.Fatalf("len = %d, 期望 2", tbl.len()) + } + time.Sleep(60 * time.Millisecond) + if _, ok := tbl.get(1); ok { + t.Fatal("过期条目应失效") + } + var wg sync.WaitGroup + for i := 0; i < 8; i++ { + wg.Add(1) + go func(n int) { + defer wg.Done() + for j := 0; j < 200; j++ { + k := uint64(n*1000 + j) + tbl.put(k, k, time.Minute) + tbl.get(k) + } + }(i) + } + wg.Wait() +} diff --git a/server/internal/proxy/handlers.go b/server/internal/proxy/handlers.go index 6372c76..71bf2ec 100644 --- a/server/internal/proxy/handlers.go +++ b/server/internal/proxy/handlers.go @@ -138,7 +138,7 @@ type sinkHolder struct { // 仅当开关开启且用户为管理员时记录;响应侧以 c.Get("raw_request") 是否非空判断是否需要捕获响应。 func (g *Gateway) recordRawRequest(c *gin.Context, u *store.User, body []byte) { if g.logRaw && u.Role == store.RoleAdmin { - c.Set("raw_request", string(body)) + c.Set("raw_request", capRaw(body, rawRequestCap)) } } diff --git a/server/internal/proxy/passthrough.go b/server/internal/proxy/passthrough.go index 92d11b6..c85a843 100644 --- a/server/internal/proxy/passthrough.go +++ b/server/internal/proxy/passthrough.go @@ -10,7 +10,6 @@ import ( "errors" "fmt" "io" - "log" "net/http" "strconv" "strings" @@ -242,9 +241,9 @@ func (g *Gateway) doProxy(c *gin.Context, cands []channel.Candidate, clientProto } // 会话亲和:上次应答过该会话的渠道优先(候选集已过滤健康/冷却/绑定), // 让上游提示词缓存被再次命中;命中不了(首次/冷却/下线)走加权随机。 - affKey := g.affinityKeyFor(c) + affKey, hasAff := g.affinityKeyFor(c) startIdx := -1 - if affKey != "" { + if hasAff { if chID, ok := g.aff.get(affKey); ok { for i, cand := range cands { if cand.Channel.ID == chID { @@ -318,7 +317,7 @@ func (g *Gateway) doProxy(c *gin.Context, cands []channel.Candidate, clientProto done, served, retry, st, b := g.proxyOne(c, ch, plan, stream, sink, sh, clientProto) release() if done { - if served && affKey != "" { + if served && hasAff { g.aff.put(affKey, ch.ID, g.affinityTTL) } return @@ -361,27 +360,27 @@ func (g *Gateway) doProxy(c *gin.Context, cands []channel.Candidate, clientProto g.recordError(c, lastCh, nil, now(), "all_channels_failed") } -// affinityKeyFor 计算本请求的会话亲和键;亲和关闭或缺少身份要素时返回空。 -func (g *Gateway) affinityKeyFor(c *gin.Context) string { +// affinityKeyFor 计算本请求的会话亲和键;亲和关闭或缺少身份要素时 has=false。 +func (g *Gateway) affinityKeyFor(c *gin.Context) (key uint64, has bool) { if g.affinityTTL <= 0 { - return "" + return 0, false } kid, ok := c.Get(CtxKeyID) kID, ok2 := kid.(uint64) if !ok || !ok2 { - return "" + return 0, false } mn, _ := c.Get("model_name") model, _ := mn.(string) // 亲和键取原始请求体的前缀:每轮重发全量上下文时开头是稳定的(模型、系统提示、 // 最早的几条消息),足够区分会话;不必为了它整包解析 JSON。 - var prefix string + var prefix []byte if v, ok := c.Get("req_body"); ok { if raw, ok2 := v.([]byte); ok2 { - prefix = string(raw) // affinityKey 内部只取前 512 字节 + prefix = raw // affinityKey 内部只取前 affinityPrefix 字节 } } - return affinityKey(kID, model, prefix) + return affinityKey(kID, model, prefix), true } // streamErrorEvent 把错误构造成客户端协议的 SSE 错误事件(响应头已写出时的唯一交付方式)。 @@ -556,7 +555,7 @@ func (g *Gateway) copyAndCapture(c *gin.Context, ch *store.Channel, resp *http.R c.Status(http.StatusOK) _, _ = c.Writer.Write(out) if _, ok := c.Get("raw_request"); ok { - c.Set("raw_response", string(data)) // 上游原始响应(未转换) + c.Set("raw_response", capRaw(data, rawResponseCap)) // 上游原始响应(未转换) } g.finishUsage(c, ch, start, store.UsageStatusSuccess, "") return true, true, false, 0, nil @@ -575,7 +574,7 @@ func (g *Gateway) streamHoldCopy(c *gin.Context, sh *streamHold, ch *store.Chann var rawResp strings.Builder commitRaw := func() { if capture { - c.Set("raw_response", rawResp.String()) + c.Set("raw_response", capRaw([]byte(rawResp.String()), rawResponseCap)) } } @@ -1036,7 +1035,7 @@ func (g *Gateway) finishUsage(c *gin.Context, ch *store.Channel, start time.Time } else { // 模型在记账时已被删除(请求进行中被下架):无法定价,按 0 记账留痕。 // 虚构单价扣用户的钱既不可审计也无法对账。 - log.Printf("proxy: trace=%s model %q not found at billing time, cost recorded as 0", trace, mn) + accessf("proxy: trace=%s model %q not found at billing time, cost recorded as 0", trace, mn) } // 落库的 input_tokens 存输入总量(含缓存读/写),与上游 message_start 口径一致,便于对账展示。 in += cacheRead + cacheCreate @@ -1112,7 +1111,7 @@ func (g *Gateway) finishUsage(c *gin.Context, ch *store.Channel, start time.Time }) // 请求级一行日志:usage_logs 落库是异步的且可能丢,排障时日志是唯一即时依据 - log.Printf("proxy: trace=%s user=%d key=%d model=%q served=%q channel=%d proto=%s in=%d out=%d r=%d cache_r=%d cache_w=%d(1h=%d) cost=%.6f status=%s http=%d latency=%dms err=%s", + accessf("proxy: trace=%s user=%d key=%d model=%q served=%q channel=%d proto=%s in=%d out=%d r=%d cache_r=%d cache_w=%d(1h=%d) cost=%.6f status=%s http=%d latency=%dms err=%s", traceStr, uidVal, kidVal, mn, servedModel, chID, p, in, out, reasoning, cacheRead, cacheCreate, cache1h, cost, status, c.Writer.Status(), latency, errCode) } diff --git a/server/internal/proxy/rawlog.go b/server/internal/proxy/rawlog.go new file mode 100644 index 0000000..f25f6b1 --- /dev/null +++ b/server/internal/proxy/rawlog.go @@ -0,0 +1,71 @@ +package proxy + +import ( + "bufio" + "fmt" + "log" + "os" + "sync" + "time" +) + +// 原始体会话与访问日志。 + +// 原始请求/响应落库的上限:开启 LogRaw 时流式响应可能是整段 SSE(百 KB~MB 级), +// 单条日志几十 MB 会让 usage_logs 膨胀、列表查询和备份都变慢。超出部分截断并标注。 +const ( + rawRequestCap = 256 << 10 // 256 KiB + rawResponseCap = 1 << 20 // 1 MiB +) + +// capRaw 把原始体截到 limit,并在末尾标注原始大小。 +func capRaw(b []byte, limit int) string { + if len(b) <= limit { + return string(b) + } + return string(b[:limit]) + fmt.Sprintf("\n…(truncated by openteam: %d bytes total)", len(b)) +} + +// accessLog 访问日志:每个请求一行,直接写 stderr 会在高 QPS 下变成系统调用热点。 +// 这里套一层 64KB 缓冲 + 每秒 flush,log.Logger 自带互斥,写入只是内存拷贝。 +var ( + accessLogOnce sync.Once + accessLog *log.Logger +) + +// syncBuf 让 bufio.Writer 的写与定时 flush 互斥: +// bufio 本身不支持并发访问,而 flush 在独立 goroutine 里跑。 +type syncBuf struct { + mu sync.Mutex + w *bufio.Writer +} + +func (s *syncBuf) Write(p []byte) (int, error) { + s.mu.Lock() + defer s.mu.Unlock() + return s.w.Write(p) +} + +func (s *syncBuf) Flush() error { + s.mu.Lock() + defer s.mu.Unlock() + return s.w.Flush() +} + +func initAccessLog() { + w := &syncBuf{w: bufio.NewWriterSize(os.Stderr, 64<<10)} + accessLog = log.New(w, "", log.LstdFlags) + go func() { + t := time.NewTicker(time.Second) + defer t.Stop() + for range t.C { + _ = w.Flush() + } + }() +} + +// accessf 写一行访问日志(带缓冲,不阻塞请求)。 +func accessf(format string, args ...any) { + accessLogOnce.Do(initAccessLog) + accessLog.Printf(format, args...) +} diff --git a/server/internal/proxy/rawlog_test.go b/server/internal/proxy/rawlog_test.go new file mode 100644 index 0000000..2daf310 --- /dev/null +++ b/server/internal/proxy/rawlog_test.go @@ -0,0 +1,31 @@ +package proxy + +import ( + "strings" + "testing" +) + +func TestCapRaw(t *testing.T) { + small := []byte("hello") + if got := capRaw(small, 10); got != "hello" { + t.Fatalf("未超限应原样返回,实际 %q", got) + } + big := []byte(strings.Repeat("x", 100)) + got := capRaw(big, 10) + if !strings.HasPrefix(got, strings.Repeat("x", 10)) { + t.Fatalf("应保留前 10 字节:%q", got) + } + if !strings.Contains(got, "truncated by openteam: 100 bytes total") { + t.Fatalf("应标注被截断与原始大小:%q", got) + } +} + +// 两个上限都远小于流式响应可能的体积,且响应上限比请求上限宽(响应通常是更大的那份) +func TestRawCapsSane(t *testing.T) { + if rawRequestCap <= 0 || rawResponseCap <= 0 { + t.Fatal("上限必须为正") + } + if rawRequestCap > rawResponseCap { + t.Fatalf("响应上限(%d)应不小于请求上限(%d)", rawResponseCap, rawRequestCap) + } +} diff --git a/server/internal/usage/usage.go b/server/internal/usage/usage.go index 4648e95..be74506 100644 --- a/server/internal/usage/usage.go +++ b/server/internal/usage/usage.go @@ -13,35 +13,76 @@ import ( ) // Recorder 异步记账器:缓冲队列 + 批量事务落库。 +// +// 多个 worker 按用户分片:同一用户的记录始终交给同一个 worker,落库顺序与扣款 +// 流水顺序保持一致;不同用户并行落库,PostgreSQL 上不再被单 goroutine 串行化。 type Recorder struct { db *gorm.DB - ch chan *store.UsageLog - wg sync.WaitGroup - closed chan struct{} - once sync.Once // 保证 Close 只执行一次(main 的 defer Shutdown 与显式 Close 双调用) + shards []chan *store.UsageLog + // workers 本实例的落库并发度(SQLite 为 1,PG 为 flushWorkersPG) + workers int + wg sync.WaitGroup + closed chan struct{} + once sync.Once // 保证 Close 只执行一次(main 的 defer Shutdown 与显式 Close 双调用) - // mu 保护 stopped:Close 之后 Record 不再进队(ch 永不 close,杜绝 send-on-closed panic), + // mu 保护 stopped:Close 之后 Record 不再进队(shard 永不 close,杜绝 send-on-closed panic), // 改走同步落库兜底。 mu sync.RWMutex stopped bool } -const batchSize = 32 +const ( + batchSize = 32 + // flushWorkersPG 落库 worker 数(按用户分片)。SQLite 只有一个写者, + // 多 worker 并发写只会互相抢锁、把 p99 拖长(实测 4 worker 时 p99 19–34ms), + // 所以按驱动自适应:PG 用多个,SQLite 只用一个。 + flushWorkersPG = 4 +) // flushRetries 批量落库的瞬时错误重试次数(SQLite busy / PG 连接抖动)。 const flushRetries = 3 func NewRecorder(db *gorm.DB) *Recorder { + return newRecorder(db, workersFor(db)) +} + +// workersFor 按驱动决定落库并发度:SQLite 只有一个写者,多 worker 只会抢锁。 +func workersFor(db *gorm.DB) int { + if db != nil && db.Dialector != nil && db.Dialector.Name() == "postgres" { + return flushWorkersPG + } + return 1 +} + +// newRecorder 指定并发度构造(测试用:在 SQLite 上也能验证多分片语义)。 +func newRecorder(db *gorm.DB, workers int) *Recorder { + if workers < 1 { + workers = 1 + } r := &Recorder{ - db: db, - ch: make(chan *store.UsageLog, 512), - closed: make(chan struct{}), + db: db, + closed: make(chan struct{}), + workers: workers, + shards: make([]chan *store.UsageLog, workers), + } + for i := range r.shards { + // 每片容量与原来的单队列等效(512 总量按片均分,至少 64) + size := 512 / workers + if size < 64 { + size = 64 + } + r.shards[i] = make(chan *store.UsageLog, size) + r.wg.Add(1) + go r.run(i) } - r.wg.Add(1) - go r.run() return r } +// shardOf 用户 → 队列分片:同一用户恒定落在同一片,保证其记账顺序。 +func (r *Recorder) shardOf(userID uint64) chan *store.UsageLog { + return r.shards[userID%uint64(len(r.shards))] +} + // Record 提交一条用量(非阻塞;队列满或已停止接收时同步写入,保证不丢账)。 func (r *Recorder) Record(l *store.UsageLog) { r.mu.RLock() @@ -50,8 +91,9 @@ func (r *Recorder) Record(l *store.UsageLog) { r.flushRetry([]*store.UsageLog{l}) return } + ch := r.shardOf(l.UserID) select { - case r.ch <- l: + case ch <- l: r.mu.RUnlock() default: r.mu.RUnlock() @@ -71,8 +113,9 @@ func (r *Recorder) Close() { }) } -func (r *Recorder) run() { +func (r *Recorder) run(shard int) { defer r.wg.Done() + ch := r.shards[shard] buf := make([]*store.UsageLog, 0, batchSize) flushBuf := func() { if len(buf) > 0 { @@ -84,7 +127,7 @@ func (r *Recorder) run() { defer tick.Stop() for { select { - case l := <-r.ch: + case l := <-ch: buf = append(buf, l) if len(buf) >= batchSize { flushBuf() @@ -94,7 +137,7 @@ func (r *Recorder) run() { // Record 已拿到 RLock 还没写进 channel,这里等它落地再收尾。 for { select { - case l := <-r.ch: + case l := <-ch: buf = append(buf, l) if len(buf) >= batchSize { flushBuf() diff --git a/server/internal/usage/usage_test.go b/server/internal/usage/usage_test.go index 4e297fe..1732d12 100644 --- a/server/internal/usage/usage_test.go +++ b/server/internal/usage/usage_test.go @@ -218,3 +218,68 @@ func TestBillingAggregatesBatch(t *testing.T) { t.Fatalf("最后一条流水 balance_after = %v,实际余额 %v", last.BalanceAfter, user.Balance) } } + +// 多 worker 落库:同一用户恒定落在同一分片(流水顺序与提交顺序一致), +// 不同用户并行落库,最终一条不丢。 +func TestRecorderShardedByUser(t *testing.T) { + db := openTestDB(t) + var users []uint64 + for i := 0; i < 3; i++ { + u := store.User{Username: fmt.Sprintf("shard-u%d", i), Email: fmt.Sprintf("shard%d@x.com", i), + PasswordHash: "x", Role: store.RoleUser, Balance: 100, Status: store.UserStatusActive} + if err := db.Create(&u).Error; err != nil { + t.Fatal(err) + } + users = append(users, u.ID) + } + // 显式用 3 个分片:SQLite 生产配置是单 worker,但分片语义本身仍要验证 + rec := newRecorder(db, 3) + const per = 40 + for i := 0; i < per; i++ { + for _, u := range users { + rec.Record(&store.UsageLog{ + TraceID: fmt.Sprintf("u%d-%d", u, i), UserID: u, ModelName: "m1", + Cost: 0.001, Status: store.UsageStatusSuccess, CreatedAt: time.Now().UTC(), + }) + } + } + rec.Close() + + var n int64 + db.Model(&store.UsageLog{}).Count(&n) + if n != int64(per*len(users)) { + t.Fatalf("落库 %d 条,期望 %d(一条不丢)", n, per*len(users)) + } + // 每个用户的分片一致 → 流水按 id 递增的顺序应与 trace 序号一致 + for _, u := range users { + var logs []store.BalanceLog + db.Where("user_id = ?", u).Order("id ASC").Find(&logs) + if len(logs) != per { + t.Fatalf("用户 %d 流水 %d 条,期望 %d", u, len(logs), per) + } + for i := 1; i < len(logs); i++ { + if logs[i].BalanceAfter > logs[i-1].BalanceAfter { + t.Fatalf("用户 %d 的流水余额应单调递减(顺序错乱):%v → %v", + u, logs[i-1].BalanceAfter, logs[i].BalanceAfter) + } + } + } +} + +// 分片函数:同一用户恒定同片,且覆盖所有分片。 +func TestShardOfStable(t *testing.T) { + rec := NewRecorder(openTestDB(t)) + defer rec.Close() + for _, u := range []uint64{1, 2, 3, 17, 1 << 40} { + if rec.shardOf(u) != rec.shardOf(u) { + t.Fatalf("用户 %d 的分片不稳定", u) + } + } + seen := map[chan *store.UsageLog]bool{} + for u := uint64(0); u < 64; u++ { + seen[rec.shardOf(u)] = true + } + if len(seen) != rec.workers { + t.Fatalf("只用到 %d 个分片,期望 %d", len(seen), rec.workers) + } +}